# Flink CDC Job 배포

이미 실행 중인 Flink Session Cluster에 CDC 파이프라인 Job을 제출하는 chart입니다.  
Session Cluster를 재시작하지 않고 파이프라인만 독립적으로 배포/재배포할 수 있습니다.

> **사전 조건**: `flink-cdc-session` chart로 Session Cluster가 RUNNING 상태여야 합니다.

## 1. 배포 방법

Release 이름이 Job, ConfigMap, ServiceAccount, Role 이름으로 사용됩니다.

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

### 여러 파이프라인 동시 배포

```sh
# 사용자 테이블 파이프라인
helm install cdc-users ./ -n flink-test -f values.yaml -f values-users.yaml

# 주문 테이블 파이프라인
helm install cdc-orders ./ -n flink-test -f values.yaml -f values-orders.yaml
```

### 파이프라인 재제출

Kubernetes Job은 완료 후 재실행이 불가하므로 uninstall → install로 재제출합니다.

```sh
helm uninstall cdc-job -n flink-test
helm install cdc-job ./ -n flink-test -f values.yaml -f custom-values.yaml
```

## 2. custom-values.yaml 설명

### 1) ServiceAccount / RBAC

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

### 2) Session Cluster 연결

| Name | 설명 | 기본값 |
| --- | --- | --- |
| `sessionCluster.restAddress` | Session Cluster REST 서비스명 | `flink-cdc-session-session-rest` |
| `sessionCluster.restPort` | REST 포트 | `8081` |

> `restAddress`는 FlinkDeployment 이름에 `-rest`를 붙인 값입니다.  
> `flink-cdc-session` chart는 FlinkDeployment를 `{release-name}-session`으로 생성하므로  
> REST 서비스명은 `{release-name}-session-rest`가 됩니다.  
> 예: release name `flink-cdc-session` → `sessionCluster.restAddress: flink-cdc-session-session-rest`

### 3) PostgreSQL 소스

| 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` |

### 4) 파이프라인

| Name | 설명 | 기본값 |
|------|------|--------|
| `pipeline.name` | 파이프라인 이름 | `cdc-pipeline` |
| `pipeline.parallelism` | 파이프라인 병렬도 | `2` |
| `pipeline.checkpointInterval` | 체크포인트 주기 | `60s` |

### 5) Iceberg 싱크

| 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` |

### 6) 라우팅

```yaml
route:
  - sourceTable: public.orders      # PostgreSQL schema.table
    sinkTable: testdb_public.orders # Iceberg namespace.table
  - sourceTable: public.users
    sinkTable: testdb_public.users
```

## 3. values 파일 예시 (복수 파이프라인)

### values-users.yaml (사용자/상품 테이블)

```yaml
sessionCluster:
  restAddress: flink-cdc-session-session-rest

pipeline:
  name: cdc-users
  parallelism: 2
  checkpointInterval: 60s

postgres:
  hostname: postgres.example.svc.cluster.local
  port: 5432
  username: flink_cdc
  password: "password"
  slotName: flink_cdc_users_slot
  decodingPlugin: pgoutput
  tables: mydb.public.users,mydb.public.products

route:
  - sourceTable: public.users
    sinkTable: mydb_public.users
  - sourceTable: public.products
    sinkTable: mydb_public.products
```

### values-orders.yaml (주문 테이블)

```yaml
sessionCluster:
  restAddress: flink-cdc-session-session-rest

pipeline:
  name: cdc-orders
  parallelism: 4
  checkpointInterval: 60s

postgres:
  hostname: postgres.example.svc.cluster.local
  slotName: flink_cdc_orders_slot
  tables: mydb.public.orders,mydb.public.order_items

route:
  - sourceTable: public.orders
    sinkTable: mydb_public.orders
  - sourceTable: public.order_items
    sinkTable: mydb_public.order_items
```

## 4. 배포 후 확인

```bash
# Job 상태 확인
kubectl get job -n flink-test

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

# Flink REST API로 Job 상태 확인
kubectl exec -n flink-test <jobmanager-pod> -- curl -s http://localhost:8081/jobs
```

## 5. 삭제

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

> **삭제 시 주의사항**: `cdc-job`을 삭제하여도 Flink Session Cluster에 제출된 Job은 삭제되지 않습니다.

## 6. 참고

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