# Flink CDC Session 배포

## 1. 배포 방법

```sh
helm upgrade --install flink-cdc-session ./ \
  --namespace flink-test \
  --create-namespace \
  -f values.yaml \
  -f custom-values.yaml
```

### Session Cluster 먼저 배포 후 Job 활성화 (권장)

```sh
# 1단계: Session Cluster만 배포 (custom-values.yaml에 job.enabled: false 설정)
helm upgrade --install flink-cdc-session ./ \
  --namespace flink-test \
  --create-namespace \
  -f values.yaml \
  -f custom-values.yaml

# 2단계: Cluster RUNNING 확인
kubectl get flinkdeployment -n flink-test -w

# 3단계: Job 포함해서 upgrade
helm upgrade flink-cdc-session ./ \
  --namespace flink-test \
  -f values.yaml \
  -f custom-values.yaml \
  --set job.enabled=true
```

## 2. custom-values.yaml 설명

### 1) ServiceAccount / RBAC

| Name | 설명 | 기본값 |
|------|------|--------|
| `serviceAccount.create` | ServiceAccount 생성 여부 | `true` |
| `serviceAccount.name` | SA 이름. 비어있으면 Release.Name 사용 | `""` |
| `rbac.create` | Role / RoleBinding 생성 여부 | `true` |

> `rbac.create: true` 시 JobManager가 TaskManager(Deployment)를 관리하기 위한 RBAC 권한이 함께 생성됩니다.

### 2) 체크포인트 저장소

| Name | 설명 | 기본값 |
|------|------|--------|
| `checkpointStorage.storageType` | 저장소 유형 (`s3` \| `gcs`) | `s3` |
| `checkpointStorage.checkpointDir` | 체크포인트 저장 경로 | `s3://flink/checkpoints` |
| `checkpointStorage.savepointDir` | 세이브포인트 저장 경로 | `s3://flink/savepoints` |
| `checkpointStorage.s3.endpoint` | S3/MinIO 엔드포인트 | `http://minio.minio.svc.cluster.local:9000` |
| `checkpointStorage.s3.pathStyleAccess` | Path-style 접근 여부 (MinIO는 `"true"`) | `"true"` |
| `checkpointStorage.s3.existingSecret` | 자격증명 Secret 이름 | `minio-credentials` |

### 3) Ingress 설정

| Name | 설명 | 기본값 |
|------|------|--------|
| `ingress.enabled` | Ingress 활성화 여부 | `true` |
| `ingress.host` | Flink Web UI 호스트명 | `flink-cdc-session.example.org` |
| `ingress.className` | IngressClass 이름 | `kong` |
| `ingress.annotations` | Ingress 어노테이션 | cert-manager + Kong HTTPS 설정 |

### 4) Truststore (사설 CA 환경)

| Name | 설명 | 기본값 |
| --- | --- | --- |
| `truststore.enabled` | 외부 JKS 사용 여부 | `true` |
| `truststore.secretName` | 기존 Secret 이름 지정. 비어있으면 `{release-name}-flink-truststore` 자동 사용 | `""` |

> `truststore.secretName: flink-truststore` 처럼 기존 Secret을 지정하면 chart가 Secret을 새로 생성하지 않고 재사용합니다.

### 5) Job 설정

| Name | 설명 | 기본값 |
|------|------|--------|
| `job.enabled` | CDC Job 활성화 여부 | `false` |

> **주의**: Session Cluster가 RUNNING 상태가 된 후 `job.enabled: true`로 upgrade하세요.

### 6) Flink 클러스터 리소스 (values.yaml에서 조정)

| Name | 설명 | 기본값 |
|------|------|--------|
| `jobManager.memory` | JobManager 메모리 | `1024m` |
| `jobManager.cpu` | JobManager CPU | `0.5` |
| `taskManager.replicas` | TaskManager 인스턴스 수 | `3` |
| `taskManager.memory` | TaskManager 메모리 | `2048m` |
| `taskManager.cpu` | TaskManager CPU | `1` |
| `taskManager.taskSlots` | TaskManager 슬롯 수 | `2` |

### 7) PostgreSQL 소스 (values.yaml에서 조정)

| Name | 설명 | 기본값 |
|------|------|--------|
| `postgres.hostname` | PostgreSQL 호스트 | `postgres-postgresql.flink-test.svc.cluster.local` |
| `postgres.port` | PostgreSQL 포트 | `5432` |
| `postgres.username` | CDC 전용 사용자 | `flink_cdc` |
| `postgres.password` | 사용자 패스워드 | `password` |
| `postgres.slotName` | Replication slot 이름 | `flink_cdc_slot` |
| `postgres.tables` | CDC 대상 테이블 목록 | `testdb.public.orders,testdb.public.users` |

### 8) Iceberg 싱크 (values.yaml에서 조정)

| Name | 설명 | 기본값 |
|------|------|--------|
| `sink.catalog.uri` | Lakekeeper REST URI | `https://lakekeeper.example.org/catalog` |
| `sink.catalog.warehouse` | 카탈로그 warehouse 이름 | `minio` |
| `sink.catalog.s3Endpoint` | Iceberg 데이터 저장소 엔드포인트 | `https://minio.example.org` |
| `sink.catalog.oauth2Uri` | Keycloak OAuth2 토큰 URI | `https://keycloak.example.org/…` |
| `sink.catalog.credential` | `client_id:client_secret` | - |
| `sink.catalog.scope` | OAuth2 scope | `lakekeeper` |

## 3. 사전 준비: Secret 생성

### 3.1 MinIO 자격증명 Secret

```bash
kubectl create secret generic minio-credentials \
  --from-literal=access-key-id=<ACCESS_KEY> \
  --from-literal=secret-access-key=<SECRET_KEY> \
  -n flink-test
```

### 3.2 Truststore Secret (`truststore.enabled: true` 인 경우)

```bash
# JVM 기본 cacerts를 베이스로 truststore 생성
IMAGE="paasup/flink-cdc-pipeline:3.5.0-flink1.20-r1"
docker run --rm --entrypoint="" "${IMAGE}" \
  cat /opt/java/openjdk/lib/security/cacerts > truststore.jks

# 사설 CA 추가 (cert-manager 사용 시)
kubectl get secret root-ca-secret -n cert-manager \
  -o jsonpath='{.data.ca\.crt}' | base64 -d > /tmp/internal-ca.pem

keytool -import -noprompt \
  -keystore truststore.jks \
  -storepass changeit \
  -alias internal-root-ca \
  -file /tmp/internal-ca.pem

# Kubernetes Secret 생성
kubectl create secret generic flink-truststore \
  --from-file=truststore.jks=./truststore.jks \
  -n flink-test
```

> 공인 CA 환경(Let's Encrypt 등)에서는 `truststore.enabled: false`로 설정하면 truststore Secret 불필요합니다.

## 4. 배포 후 확인

```bash
# Session Cluster 상태 확인
kubectl get flinkdeployment -n flink-test

# Pod 상태 확인
kubectl get pods -n flink-test

# Job 로그 확인 (release name: flink-cdc-session, job.name: submit-cdc)
kubectl logs -n flink-test job/flink-cdc-session-submit-cdc

# Flink Web UI (포트포워딩)
kubectl port-forward svc/flink-cdc-session-session-rest 8081:8081 -n flink-test
```

## 5. 삭제

```bash
helm uninstall flink-cdc-session -n flink-test
```

## 6. 참고

자세한 파라미터 설명, truststore 생성 방법, 렌더링 확인 방법은 [README.md](README.md)를 참조하세요.