diff --git a/manifests/helm/flink-cdc-session/0.1.0/BUILD-README.md b/manifests/helm/flink-cdc-session/0.1.0/BUILD-README.md new file mode 100644 index 0000000..adfcae5 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/BUILD-README.md @@ -0,0 +1,105 @@ +# flink-cdc-session 버전 갱신 가이드 + +## 1. git 작업 환경 구성 + +- DIP 카탈로그 git 다운로드 +``` +$ git clone https://github.com/paasup/dip-catalog.git +``` + +- 작업 브랜치로 체크아웃 +``` +$ git checkout -b update-flink-cdc-session/0.1.0 +``` + +## 2. helm 차트 개발 및 업데이트 + +- 현재 flink-cdc-session 차트는 직접 작성하여 업데이트 중입니다. +- 기존 오픈소스 차트를 기반으로 하지 않고, 프로젝트 요구사항에 맞게 커스텀 개발하고 있습니다. + +### 2.1 차트 구조 +``` +flink-cdc-session/0.1.0/ +├── Chart.yaml # 차트 메타데이터 +├── values.yaml # 기본 설정값 +├── custom-values.yaml # 커스텀 설정값 +├── BUILD-README.md # 빌드 가이드 (본 파일) +├── CUSTOM-README.md # 배포 가이드 +├── README.md # 차트 상세 문서 +└── templates/ + ├── _helpers.tpl # 헬퍼 템플릿 + ├── configmap.yaml # flink-cdc.yaml + pipeline.yaml + ├── flinkdeployment.yaml # Flink Session Cluster + ├── ingress.yaml # Flink Web UI Ingress + ├── job.yaml # CDC 파이프라인 제출 Job (조건부) + └── secrets.yaml # truststore / gcs-key Secret (조건부) +``` + +### 2.2 개발 중인 주요 기능 +- PostgreSQL → Apache Iceberg CDC 파이프라인 (Flink Session Mode) +- Flink CDC 3.5.0 기반 +- S3/MinIO 및 GCS 체크포인트 저장소 지원 +- 사설 CA 인증서(JKS truststore) 지원 +- Iceberg REST 카탈로그(Lakekeeper) + OAuth2 연동 +- Ingress 지원 (Flink Web UI) + +## 3. 이미지 정보 + +- 기본 이미지: `paasup/flink-cdc-pipeline:3.5.0-flink1.20-r1` +- Flink CDC 3.5.0 + Flink 1.20 기반 커스텀 이미지 +- PostgreSQL CDC 커넥터 및 Iceberg 싱크 포함 + +## 4. git push 및 tag 추가 + +- 갱신작업 진행후 commit +```sh +$ git add . +$ git commit -m "update flink-cdc-session/0.1.0" +``` + +- main 브랜치에 체크아웃 후 merge +```sh +$ git checkout main +$ git merge update-flink-cdc-session/0.1.0 +``` + +- git에 push 후 작업 브랜치 삭제 +```sh +$ git push -u origin main +$ git branch -d update-flink-cdc-session/0.1.0 +``` + +- git tag 추가 후 push +```sh +$ git tag flink-cdc-session/0.1.0 +$ git push origin flink-cdc-session/0.1.0 +``` + +## 5. 차트 버전 정보 + +- flink-cdc-session/0.1.0 + - Apache Flink CDC 3.5.0 + Flink 1.20 기반 + - PostgreSQL WAL → Apache Iceberg REST 카탈로그 CDC 파이프라인 + - S3/MinIO 및 GCS 체크포인트 저장소 지원 + - 사설 CA 인증서(JKS truststore) 지원 + - Iceberg REST 카탈로그(Lakekeeper) + Keycloak OAuth2 연동 + - 서비스 배포를 위하여 custom-values.yaml에 정의하였다. + - 차트의 빌드 방법과 배포 방법을 BUILD-README.md, CUSTOM-README.md 문서에 작성하였다. + +## 6. 개발 참고사항 + +### 6.1 주요 의존성 +- Flink Kubernetes Operator 설치 필요 +- Kubernetes 1.25+ 권장 +- Helm 3.x 필요 +- PostgreSQL logical replication 활성화 필요 (`wal_level=logical`) + +### 6.2 보안 설정 +- JKS 기반 truststore로 사설 CA 인증서 관리 +- Kubernetes Secret을 통한 MinIO/GCS 자격증명 관리 +- OAuth2 client_id:client_secret을 통한 Lakekeeper 인증 + +## 참고) 관련 문서 +- [Flink CDC 공식 문서](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.5/) +- [Flink Kubernetes Operator 문서](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/) +- [Apache Iceberg 문서](https://iceberg.apache.org/docs/latest/) diff --git a/manifests/helm/flink-cdc-session/0.1.0/CUSTOM-README.md b/manifests/helm/flink-cdc-session/0.1.0/CUSTOM-README.md new file mode 100644 index 0000000..3c4d647 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/CUSTOM-README.md @@ -0,0 +1,165 @@ +# 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) 체크포인트 저장소 + +| 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` | + +### 2) 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 설정 | + +### 3) Truststore (사설 CA 환경) + +| Name | 설명 | 기본값 | +|------|------|--------| +| `truststore.enabled` | 외부 JKS 사용 여부 | `true` | +| `truststore.secretName` | truststore Secret 이름 | `flink-truststore` | + +### 4) Job 설정 + +| Name | 설명 | 기본값 | +|------|------|--------| +| `job.enabled` | CDC Job 활성화 여부 | `false` | + +> **주의**: Session Cluster가 RUNNING 상태가 된 후 `job.enabled: true`로 upgrade하세요. + +### 5) 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` | + +### 6) 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` | + +### 7) 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= \ + --from-literal=secret-access-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 로그 확인 +kubectl logs -n flink-test job/submit-cdc + +# Flink Web UI (포트포워딩) +kubectl port-forward svc/flink-cdc-session-rest 8081:8081 -n flink-test +``` + +## 5. 삭제 + +```bash +helm uninstall flink-cdc-session -n flink-test +``` + +## 6. 참고 + +자세한 파라미터 설명, truststore 생성 방법, 렌더링 확인 방법은 [README.md](README.md)를 참조하세요. diff --git a/manifests/helm/flink-cdc-session/0.1.0/Chart.yaml b/manifests/helm/flink-cdc-session/0.1.0/Chart.yaml new file mode 100644 index 0000000..a4238bc --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/Chart.yaml @@ -0,0 +1,6 @@ +apiVersion: v2 +appVersion: 3.5.0-flink1.20 +description: Flink CDC Session Mode cluster with PostgreSQL-to-Iceberg pipeline +name: flink-cdc-session +type: application +version: 0.1.0 diff --git a/manifests/helm/flink-cdc-session/0.1.0/README.md b/manifests/helm/flink-cdc-session/0.1.0/README.md new file mode 100644 index 0000000..d19b2b3 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/README.md @@ -0,0 +1,472 @@ +# flink-cdc-session Helm Chart + +PostgreSQL → Apache Iceberg CDC 파이프라인을 Flink Session Mode로 배포하는 Helm chart입니다. + +## 아키텍처 + +``` +PostgreSQL (WAL) + │ pgoutput + ▼ +Flink Session Cluster (FlinkDeployment) + │ Flink CDC 3.5.0 + ▼ +Iceberg REST Catalog (Lakekeeper) + │ S3FileIO + ▼ +Object Storage (MinIO / GCS / S3) +``` + +## 사전 요구사항 + +| 항목 | 버전 | +|------|------| +| Helm | 3.x | +| Flink Kubernetes Operator | 1.x | +| Kubernetes | 1.25+ | + +## 파일 구조 + +``` +flink-cdc-session/ +├── Chart.yaml +├── values.yaml +├── custom-values.yaml # 환경별 오버라이드 (배포 시 사용) +└── templates/ + ├── _helpers.tpl + ├── configmap.yaml # flink-cdc.yaml + pipeline.yaml + ├── flinkdeployment.yaml # Flink Session Cluster + ├── job.yaml # CDC 파이프라인 제출 Job (조건부) + └── secrets.yaml # truststore / gcs-key Secret (조건부) +``` + +--- + +## 외부 JKS(truststore) 사용 여부 결정 + +이 chart는 `truststore.enabled` 플래그로 외부 JKS 사용 여부를 제어합니다. +**배포 전에 반드시 아래 기준에 따라 값을 결정하세요.** + +### 판단 기준 + +Flink는 다음 엔드포인트와 HTTPS로 통신합니다. + +| 엔드포인트 | 용도 | +|-----------|------| +| Lakekeeper (`sink.catalog.uri`) | Iceberg REST 카탈로그 | +| Keycloak (`sink.catalog.oauth2Uri`) | OAuth2 토큰 발급 | +| MinIO (`sink.catalog.s3Endpoint`) | Iceberg 데이터 파일 읽기/쓰기 | + +**`truststore.enabled: true` (외부 JKS 사용)** +→ 위 엔드포인트 중 하나라도 **사설 CA**가 서명한 인증서를 사용하는 경우 + +**`truststore.enabled: false` (외부 JKS 불필요)** +→ 모든 엔드포인트가 **공인 CA**(Let's Encrypt, DigiCert 등)로 서명된 인증서를 사용하는 경우 +→ JVM 기본 cacerts만으로 TLS 검증이 가능하며, `flink-truststore` Secret도 불필요 + +> **왜 외부 JKS가 필요한가** +> Flink JVM 옵션 `-Djavax.net.ssl.trustStore`는 JVM 기본 cacerts를 **완전히 교체**합니다. +> 따라서 사설 CA 인증서뿐만 아니라 공인 CA 체인도 포함된 truststore를 직접 만들어야 합니다. +> 반대로 이 옵션을 설정하지 않으면 JVM 기본 cacerts가 그대로 사용됩니다. + +### 엔드포인트 인증서 확인 방법 + +로컬 또는 클러스터 내부에서 아래 명령으로 인증서 발급자를 확인합니다. + +```bash +# 인증서 발급자(Issuer) 확인 +openssl s_client -connect lakekeeper.example.org:443 -showcerts /dev/null \ + | openssl x509 -noout -issuer -subject + +openssl s_client -connect keycloak.example.org:443 -showcerts /dev/null \ + | openssl x509 -noout -issuer -subject +``` + +출력 예시: + +``` +# 사설 CA → truststore.enabled: true 필요 +issuer=O=paasup, CN=paasup-root-ca +subject=CN=lakekeeper.example.org + +# 공인 CA → truststore.enabled: false 가능 +issuer=C=US, O=Let's Encrypt, CN=R11 +subject=CN=lakekeeper.example.org +``` + +클러스터 내부에서 확인하는 경우: + +```bash +kubectl run ssl-check --rm -it --image=alpine/openssl --restart=Never -- \ + s_client -connect lakekeeper.example.org:443 -showcerts /dev/null \ + | grep -E "issuer|subject" +``` + +--- + +## Secret 준비 + +### 1. truststore.jks 생성 (`truststore.enabled: true` 인 경우만 필요) + +#### Step 1 — JVM 기본 cacerts 추출 + +Docker 이미지 내부의 JVM cacerts를 truststore 베이스로 사용합니다. + +```bash +IMAGE="wbsong111/flink-cdc-pipeline:3.5.0-flink1.20-r1" + +docker run --rm --entrypoint="" "${IMAGE}" \ + cat /opt/java/openjdk/lib/security/cacerts > truststore.jks +``` + +#### Step 2 — 사설 CA 추가 + +**cert-manager ClusterIssuer에서 추출하는 경우:** + +```bash +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 + +rm -f /tmp/internal-ca.pem +``` + +**PEM 파일에서 직접 추가하는 경우:** + +```bash +keytool -import -noprompt \ + -keystore truststore.jks \ + -storepass changeit \ + -alias my-ca \ + -file ./certs/my-ca.crt +``` + +추가된 인증서 목록 확인: + +```bash +keytool -list -keystore truststore.jks -storepass changeit | tail -5 +``` + +#### Step 3 — truststore 동작 검증 + +배포 전에 truststore로 실제 엔드포인트 TLS 연결이 성공하는지 확인합니다. + +```bash +# 사설 CA PEM으로 curl 검증 +curl -v --cacert /tmp/internal-ca.pem https://lakekeeper.example.org/catalog 2>&1 \ + | grep -E "SSL|TLS|certificate" +``` + +#### Step 4 — Kubernetes Secret 생성 + +```bash +kubectl create secret generic flink-truststore \ + --from-file=truststore.jks=./truststore.jks \ + -n flink-test +``` + +기존 Secret 업데이트: + +```bash +kubectl create secret generic flink-truststore \ + --from-file=truststore.jks=./truststore.jks \ + -n flink-test \ + --dry-run=client -o yaml | kubectl apply -f - +``` + +> `create-truststore.sh` 스크립트로 Step 1~4를 한번에 실행할 수 있습니다: +> ```bash +> bash ../../create-truststore.sh +> ``` + +#### 배포 후 마운트 확인 + +```bash +# truststore.jks가 Pod에 정상 마운트되었는지 확인 +kubectl exec -n flink-test deploy/flink-cdc-session -- ls -la /opt/flink/certs/ + +# JVM이 truststore를 인식하는지 확인 (JobManager 로그에서 SSL 관련 출력) +kubectl logs -n flink-test -l component=jobmanager | grep -i "truststore\|ssl\|tls" | head -20 +``` + +--- + +### 2. 체크포인트 저장소 자격증명 + +`checkpointStorage.storageType` 값에 따라 필요한 Secret이 다릅니다. + +#### MinIO / S3 — existingSecret 방식 (권장) + +```bash +kubectl create secret generic minio-credentials \ + --from-literal=access-key-id= \ + --from-literal=secret-access-key= \ + -n flink-test +``` + +`custom-values.yaml`에서 참조: + +```yaml +checkpointStorage: + storageType: s3 + checkpointDir: s3://flink/checkpoints + savepointDir: s3://flink/savepoints + s3: + endpoint: "http://minio.minio.svc.cluster.local:9000" # 클러스터 내부 MinIO + pathStyleAccess: "true" + existingSecret: minio-credentials +``` + +클러스터 내부 MinIO를 사용할 경우 plain HTTP 엔드포인트(`http://`)를 사용하면 TLS 없이 통신할 수 있습니다. + +#### GCS — 서비스 계정 키 방식 + +```bash +kubectl create secret generic gcs-key \ + --from-file=key.json=./gcs_key/key.json \ + -n flink-test +``` + +`custom-values.yaml`에서 참조: + +```yaml +checkpointStorage: + storageType: gcs + checkpointDir: gs://my-bucket/flink/checkpoints + savepointDir: gs://my-bucket/flink/savepoints + gcs: + credentialSecret: gcs-key + credentialPath: /opt/flink/gcs-key/key.json +``` + +--- + +## 배포 + +### custom-values.yaml 사용 (권장) + +환경별 설정은 `values.yaml`을 수정하지 않고 `custom-values.yaml`에 오버라이드합니다. + +```bash +helm install flink-cdc-session ./flink/helm/flink-cdc-session \ + -n flink-test \ + -f values.yaml \ + -f custom-values.yaml +``` + +### Session Cluster만 먼저 배포 (Job 제외) + +```bash +# 1단계: Session Cluster만 배포 (custom-values.yaml에 job.enabled: false 설정) +helm install flink-cdc-session ./flink/helm/flink-cdc-session \ + -n flink-test \ + -f values.yaml \ + -f custom-values.yaml + +# 2단계: Cluster 기동 확인 +kubectl get flinkdeployment -n flink-test -w + +# 3단계: Job 포함해서 업그레이드 +helm upgrade flink-cdc-session ./flink/helm/flink-cdc-session \ + -n flink-test \ + -f values.yaml \ + -f custom-values.yaml \ + --set job.enabled=true +``` + +### truststore 불필요한 경우 (공인 CA 환경) + +```bash +helm install flink-cdc-session ./flink/helm/flink-cdc-session \ + -n flink-test \ + -f values.yaml \ + -f custom-values.yaml \ + --set truststore.enabled=false +``` + +`truststore.enabled=false` 시 다음이 모두 제거됩니다: +- `env.java.opts.jobmanager/taskmanager`의 SSL JVM 옵션 +- Pod의 truststore volumeMount 및 volume + +### Secret도 chart에서 생성하는 경우 + +```bash +# truststore Secret chart 생성 +helm install flink-cdc-session ./flink/helm/flink-cdc-session \ + -n flink-test \ + --set truststore.create=true \ + --set truststore.base64jks="$(base64 < truststore.jks)" + +# GCS Secret chart 생성 +helm install flink-cdc-session ./flink/helm/flink-cdc-session \ + -n flink-test \ + --set checkpointStorage.storageType=gcs \ + --set checkpointStorage.gcs.create=true \ + --set "checkpointStorage.gcs.base64json=$(base64 < gcs_key/key.json)" +``` + +### 삭제 + +```bash +helm uninstall flink-cdc-session -n flink-test +``` + +--- + +## values.yaml 파라미터 + +### 이미지 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `image.repository` | `wbsong111/flink-cdc-pipeline` | 컨테이너 이미지 저장소 | +| `image.tag` | `3.5.0-flink1.20-r1` | 이미지 태그 | +| `image.pullPolicy` | `Always` | 이미지 풀 정책 | + +### Flink 클러스터 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `flinkVersion` | `v1_20` | Flink 버전 | +| `serviceAccount` | `flink` | Pod 서비스 어카운트 | +| `jobManager.replicas` | `1` | JobManager 인스턴스 수 | +| `jobManager.memory` | `1024m` | JobManager 메모리 | +| `jobManager.cpu` | `0.5` | JobManager CPU | +| `taskManager.replicas` | `3` | TaskManager 인스턴스 수 | +| `taskManager.memory` | `2048m` | TaskManager 메모리 | +| `taskManager.cpu` | `1` | TaskManager CPU | +| `taskManager.taskSlots` | `2` | TaskManager 슬롯 수 | + +### CheckpointStorage (저장 위치 + 자격증명) + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `checkpointStorage.storageType` | `s3` | 저장소 유형 (`gcs` \| `s3`) | +| `checkpointStorage.checkpointDir` | `s3://my-bucket/…` | 체크포인트 저장 경로 | +| `checkpointStorage.savepointDir` | `s3://my-bucket/…` | 세이브포인트 저장 경로 | +| `checkpointStorage.s3.endpoint` | `""` | S3/MinIO 엔드포인트. 비워두면 AWS S3 | +| `checkpointStorage.s3.pathStyleAccess` | `"false"` | Path-style 접근 여부. MinIO는 `"true"` | +| `checkpointStorage.s3.existingSecret` | `""` | 자격증명 Secret 이름. 설정 시 아래 두 항목보다 우선 | +| `checkpointStorage.s3.accessKeyIdKey` | `access-key-id` | Secret 내 Access Key ID 필드명 | +| `checkpointStorage.s3.secretAccessKeyKey` | `secret-access-key` | Secret 내 Secret Access Key 필드명 | +| `checkpointStorage.s3.accessKeyId` | `""` | Access Key ID 직접 입력 (테스트용) | +| `checkpointStorage.s3.secretAccessKey` | `""` | Secret Access Key 직접 입력 (테스트용) | +| `checkpointStorage.gcs.credentialSecret` | `gcs-key` | GCS 키 Secret 이름 | +| `checkpointStorage.gcs.credentialPath` | `/opt/flink/gcs-key/key.json` | 컨테이너 내 키 파일 경로 | +| `checkpointStorage.gcs.create` | `false` | `true`이면 chart가 Secret 생성 | +| `checkpointStorage.gcs.base64json` | `""` | `create: true` 시 필수, base64 인코딩된 key.json | + +### 체크포인트 동작 설정 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `checkpoint.interval` | `60s` | 체크포인트 주기 | +| `checkpoint.mode` | `EXACTLY_ONCE` | 체크포인트 모드 | + +### Truststore (외부 JKS) + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `truststore.enabled` | `true` | 외부 JKS 사용 여부. 사설 CA 환경이면 `true` | +| `truststore.secretName` | `flink-truststore` | truststore Secret 이름 | +| `truststore.password` | `changeit` | JKS 패스워드 | +| `truststore.create` | `false` | `true`이면 chart가 Secret 생성 | +| `truststore.base64jks` | `""` | `create: true` 시 필수, base64 인코딩된 .jks | + +### PostgreSQL 소스 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `postgres.hostname` | `postgres-postgresql.flink-test.svc…` | PostgreSQL 호스트 | +| `postgres.port` | `5432` | PostgreSQL 포트 | +| `postgres.username` | `flink_cdc` | CDC 전용 사용자 | +| `postgres.password` | `password` | 사용자 패스워드 | +| `postgres.slotName` | `flink_cdc_slot` | Replication slot 이름 | +| `postgres.decodingPlugin` | `pgoutput` | WAL 디코딩 플러그인 | +| `postgres.tables` | `testdb.public.orders,…` | CDC 대상 테이블 목록 | + +### 파이프라인 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `flinkCdc.parallelism` | `1` | Flink CDC 글로벌 병렬도 | +| `flinkCdc.schemaChangeBehavior` | `EVOLVE` | 스키마 변경 처리 방식 | +| `pipeline.name` | `pg-to-iceberg-test` | 파이프라인 이름 | +| `pipeline.parallelism` | `2` | 파이프라인 병렬도 | +| `pipeline.checkpointInterval` | `60s` | 파이프라인 체크포인트 주기 | + +### Iceberg 싱크 + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `sink.catalog.uri` | `https://lakekeeper.example.org/catalog` | Lakekeeper REST URI | +| `sink.catalog.warehouse` | `minio` | 카탈로그 warehouse 이름 | +| `sink.catalog.s3Endpoint` | `https://minio.example.org` | S3 호환 오브젝트 스토리지 엔드포인트 | +| `sink.catalog.s3PathStyleAccess` | `"true"` | Path-style S3 접근 여부 | +| `sink.catalog.oauth2Uri` | `https://keycloak.example.org/…` | Keycloak OAuth2 토큰 URI | +| `sink.catalog.credential` | `lakekeeper-admin:…` | `client_id:client_secret` | +| `sink.catalog.scope` | `lakekeeper` | OAuth2 scope | + +### 라우팅 + +```yaml +route: + - sourceTable: public.orders # PostgreSQL schema.table + sinkTable: testdb_public.orders # Iceberg namespace.table + - sourceTable: public.users + sinkTable: testdb_public.users +``` + +### Job + +| 파라미터 | 기본값 | 설명 | +|----------|--------|------| +| `job.enabled` | `true` | `false`이면 Job 리소스 미생성 | +| `job.name` | `submit-cdc` | Job 리소스 이름 | +| `job.restAddress` | `flink-cdc-session-rest` | Session Cluster REST 서비스명 | +| `job.restPort` | `8081` | REST 포트 | +| `job.pipelineFile` | `./conf/pipeline.yaml` | 파이프라인 설정 파일 경로 | + +--- + +## 렌더링 확인 + +```bash +# 전체 렌더링 확인 +helm template flink-cdc-session . -n flink-test \ + -f values.yaml -f custom-values.yaml + +# lint 검사 +helm lint . + +# S3/MinIO 렌더링 확인 +helm template flink-cdc-session . -n flink-test \ + -f values.yaml -f custom-values.yaml \ + | grep -E "checkpointDir|savepointDir|AWS_ACCESS|s3.endpoint" + +# truststore 비활성화 시 SSL 옵션이 제거되는지 확인 +helm template flink-cdc-session . -n flink-test \ + -f values.yaml -f custom-values.yaml \ + --set truststore.enabled=false | grep -A3 "java.opts" +``` + +## 배포 후 확인 + +```bash +# Session Cluster 상태 확인 +kubectl get flinkdeployment -n flink-test + +# Job 로그 확인 +kubectl logs -n flink-test job/submit-cdc + +# Job 삭제 후 재제출 +kubectl delete job submit-cdc -n flink-test +helm upgrade flink-cdc-session . -n flink-test \ + -f values.yaml -f custom-values.yaml \ + --set job.enabled=true +``` diff --git a/manifests/helm/flink-cdc-session/0.1.0/custom-values.yaml b/manifests/helm/flink-cdc-session/0.1.0/custom-values.yaml new file mode 100644 index 0000000..067a340 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/custom-values.yaml @@ -0,0 +1,31 @@ +# custom-values.yaml +# 실제 배포에 사용하는 환경별 오버라이드 값 +# 사용법: helm install ... -f values.yaml -f custom-values.yaml + +checkpointStorage: + storageType: s3 + checkpointDir: s3://flink/checkpoints + savepointDir: s3://flink/savepoints + s3: + endpoint: "http://minio.minio.svc.cluster.local:9000" + pathStyleAccess: "true" + existingSecret: minio-credentials + +ingress: + enabled: true + host: "flink-cdc-session.example.org" + className: kong + annotations: + cert-manager.io/cluster-issuer: root-ca-issuer + cert-manager.io/duration: 8760h + cert-manager.io/renew-before: 720h + konghq.com/protocols: https + konghq.com/https-redirect-status-code: "301" + +truststore: + enabled: true + secretName: flink-truststore + +# Session Cluster만 먼저 배포. RUNNING 확인 후 job.enabled: true로 upgrade. +job: + enabled: false diff --git a/manifests/helm/flink-cdc-session/0.1.0/templates/_helpers.tpl b/manifests/helm/flink-cdc-session/0.1.0/templates/_helpers.tpl new file mode 100644 index 0000000..b0648d8 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/templates/_helpers.tpl @@ -0,0 +1,19 @@ +{{/* +Chart label +*/}} +{{- define "flink-cdc-session.chart" -}} +{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }} +{{- end }} + +{{/* +Common labels +*/}} +{{- define "flink-cdc-session.labels" -}} +helm.sh/chart: {{ include "flink-cdc-session.chart" . }} +app.kubernetes.io/name: {{ .Chart.Name }} +app.kubernetes.io/instance: {{ .Release.Name }} +{{- if .Chart.AppVersion }} +app.kubernetes.io/version: {{ .Chart.AppVersion | quote }} +{{- end }} +app.kubernetes.io/managed-by: {{ .Release.Service }} +{{- end }} diff --git a/manifests/helm/flink-cdc-session/0.1.0/templates/configmap.yaml b/manifests/helm/flink-cdc-session/0.1.0/templates/configmap.yaml new file mode 100644 index 0000000..ff4073b --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/templates/configmap.yaml @@ -0,0 +1,45 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: cdc-pipeline-config + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-session.labels" . | nindent 4 }} +data: + flink-cdc.yaml: | + parallelism: {{ .Values.flinkCdc.parallelism }} + schema.change.behavior: {{ .Values.flinkCdc.schemaChangeBehavior }} + + pipeline.yaml: | + source: + type: postgres + hostname: {{ .Values.postgres.hostname }} + port: {{ .Values.postgres.port }} + username: {{ .Values.postgres.username }} + password: {{ .Values.postgres.password }} + decoding.plugin.name: {{ .Values.postgres.decodingPlugin }} + slot.name: {{ .Values.postgres.slotName }} + tables: {{ .Values.postgres.tables }} + + pipeline: + name: {{ .Values.pipeline.name }} + parallelism: {{ .Values.pipeline.parallelism }} + execution.checkpointing.interval: {{ .Values.pipeline.checkpointInterval }} + + sink: + type: iceberg + catalog.properties.type: rest + catalog.properties.uri: {{ .Values.sink.catalog.uri | quote }} + catalog.properties.warehouse: {{ .Values.sink.catalog.warehouse }} + catalog.properties.io-impl: org.apache.iceberg.aws.s3.S3FileIO + catalog.properties.s3.endpoint: {{ .Values.sink.catalog.s3Endpoint | quote }} + catalog.properties.s3.path-style-access: {{ .Values.sink.catalog.s3PathStyleAccess | quote }} + catalog.properties.oauth2-server-uri: {{ .Values.sink.catalog.oauth2Uri | quote }} + catalog.properties.credential: {{ .Values.sink.catalog.credential }} + catalog.properties.scope: {{ .Values.sink.catalog.scope }} + + route: + {{- range .Values.route }} + - source-table: {{ .sourceTable }} + sink-table: {{ .sinkTable }} + {{- end }} diff --git a/manifests/helm/flink-cdc-session/0.1.0/templates/flinkdeployment.yaml b/manifests/helm/flink-cdc-session/0.1.0/templates/flinkdeployment.yaml new file mode 100644 index 0000000..2f2cc17 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/templates/flinkdeployment.yaml @@ -0,0 +1,103 @@ +apiVersion: flink.apache.org/v1beta1 +kind: FlinkDeployment +metadata: + name: flink-cdc-session + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-session.labels" . | nindent 4 }} +spec: + image: {{ .Values.image.repository }}:{{ .Values.image.tag }} + imagePullPolicy: {{ .Values.image.pullPolicy }} + flinkVersion: {{ .Values.flinkVersion }} + flinkConfiguration: + classloader.resolve-order: parent-first + taskmanager.numberOfTaskSlots: {{ .Values.taskManager.taskSlots | quote }} + execution.checkpointing.interval: {{ .Values.checkpoint.interval }} + execution.checkpointing.mode: {{ .Values.checkpoint.mode }} + execution.checkpointing.dir: {{ .Values.checkpointStorage.checkpointDir }} + execution.checkpointing.savepoint-dir: {{ .Values.checkpointStorage.savepointDir }} + classloader.parent-first-patterns.additional: org.apache.iceberg. + {{- if eq .Values.checkpointStorage.storageType "s3" }} + {{- if .Values.checkpointStorage.s3.endpoint }} + s3.endpoint: {{ .Values.checkpointStorage.s3.endpoint }} + s3.path-style-access: {{ .Values.checkpointStorage.s3.pathStyleAccess | quote }} + {{- end }} + {{- end }} + {{- if .Values.truststore.enabled }} + env.java.opts.jobmanager: >- + -Djavax.net.ssl.trustStore=/opt/flink/certs/truststore.jks + -Djavax.net.ssl.trustStorePassword={{ .Values.truststore.password }} + -Djavax.net.ssl.trustStoreType=JKS + env.java.opts.taskmanager: >- + -Djavax.net.ssl.trustStore=/opt/flink/certs/truststore.jks + -Djavax.net.ssl.trustStorePassword={{ .Values.truststore.password }} + -Djavax.net.ssl.trustStoreType=JKS + {{- end }} + serviceAccount: {{ .Values.serviceAccount }} + jobManager: + replicas: {{ .Values.jobManager.replicas }} + resource: + memory: {{ .Values.jobManager.memory }} + cpu: {{ .Values.jobManager.cpu }} + taskManager: + replicas: {{ .Values.taskManager.replicas }} + resource: + memory: {{ .Values.taskManager.memory }} + cpu: {{ .Values.taskManager.cpu }} + podTemplate: + spec: + containers: + - name: flink-main-container + {{- if eq .Values.checkpointStorage.storageType "gcs" }} + env: + - name: GOOGLE_APPLICATION_CREDENTIALS + value: {{ .Values.checkpointStorage.gcs.credentialPath }} + {{- else if eq .Values.checkpointStorage.storageType "s3" }} + {{- if or .Values.checkpointStorage.s3.existingSecret .Values.checkpointStorage.s3.accessKeyId }} + env: + {{- if .Values.checkpointStorage.s3.existingSecret }} + - name: AWS_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: {{ .Values.checkpointStorage.s3.existingSecret }} + key: {{ .Values.checkpointStorage.s3.accessKeyIdKey }} + - name: AWS_SECRET_ACCESS_KEY + valueFrom: + secretKeyRef: + name: {{ .Values.checkpointStorage.s3.existingSecret }} + key: {{ .Values.checkpointStorage.s3.secretAccessKeyKey }} + {{- else }} + - name: AWS_ACCESS_KEY_ID + value: {{ .Values.checkpointStorage.s3.accessKeyId }} + - name: AWS_SECRET_ACCESS_KEY + value: {{ .Values.checkpointStorage.s3.secretAccessKey }} + {{- end }} + {{- end }} + {{- end }} + volumeMounts: + - name: pipeline-config + mountPath: /opt/flink/flink-cdc-3.5.0/conf + {{- if .Values.truststore.enabled }} + - name: truststore + mountPath: /opt/flink/certs + readOnly: true + {{- end }} + {{- if eq .Values.checkpointStorage.storageType "gcs" }} + - name: gcs-key + mountPath: /opt/flink/gcs-key + readOnly: true + {{- end }} + volumes: + - name: pipeline-config + configMap: + name: cdc-pipeline-config + {{- if .Values.truststore.enabled }} + - name: truststore + secret: + secretName: {{ .Values.truststore.secretName }} + {{- end }} + {{- if eq .Values.checkpointStorage.storageType "gcs" }} + - name: gcs-key + secret: + secretName: {{ .Values.checkpointStorage.gcs.credentialSecret }} + {{- end }} diff --git a/manifests/helm/flink-cdc-session/0.1.0/templates/ingress.yaml b/manifests/helm/flink-cdc-session/0.1.0/templates/ingress.yaml new file mode 100644 index 0000000..2f1b85a --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/templates/ingress.yaml @@ -0,0 +1,32 @@ +{{- if .Values.ingress.enabled }} +apiVersion: networking.k8s.io/v1 +kind: Ingress +metadata: + name: {{ .Release.Name }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-session.labels" . | nindent 4 }} + {{- if .Values.ingress.annotations }} + annotations: + {{- toYaml .Values.ingress.annotations | nindent 4 }} + {{- end }} +spec: + {{- if .Values.ingress.className }} + ingressClassName: {{ .Values.ingress.className }} + {{- end }} + rules: + - host: {{ .Values.ingress.host }} + http: + paths: + - path: / + pathType: ImplementationSpecific + backend: + service: + name: {{ .Release.Name }}-rest + port: + number: 8081 + tls: + - hosts: + - {{ .Values.ingress.host }} + secretName: {{ .Release.Name }}-tls +{{- end }} diff --git a/manifests/helm/flink-cdc-session/0.1.0/templates/job.yaml b/manifests/helm/flink-cdc-session/0.1.0/templates/job.yaml new file mode 100644 index 0000000..f719712 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/templates/job.yaml @@ -0,0 +1,80 @@ +{{- if .Values.job.enabled }} +apiVersion: batch/v1 +kind: Job +metadata: + name: {{ .Values.job.name }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-session.labels" . | nindent 4 }} +spec: + backoffLimit: 0 + template: + spec: + serviceAccountName: {{ .Values.serviceAccount }} + containers: + - name: submit + image: {{ .Values.image.repository }}:{{ .Values.image.tag }} + {{- if eq .Values.checkpointStorage.storageType "gcs" }} + env: + - name: GOOGLE_APPLICATION_CREDENTIALS + value: {{ .Values.checkpointStorage.gcs.credentialPath }} + {{- else if eq .Values.checkpointStorage.storageType "s3" }} + {{- if or .Values.checkpointStorage.s3.existingSecret .Values.checkpointStorage.s3.accessKeyId }} + env: + {{- if .Values.checkpointStorage.s3.existingSecret }} + - name: AWS_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: {{ .Values.checkpointStorage.s3.existingSecret }} + key: {{ .Values.checkpointStorage.s3.accessKeyIdKey }} + - name: AWS_SECRET_ACCESS_KEY + valueFrom: + secretKeyRef: + name: {{ .Values.checkpointStorage.s3.existingSecret }} + key: {{ .Values.checkpointStorage.s3.secretAccessKeyKey }} + {{- else }} + - name: AWS_ACCESS_KEY_ID + value: {{ .Values.checkpointStorage.s3.accessKeyId }} + - name: AWS_SECRET_ACCESS_KEY + value: {{ .Values.checkpointStorage.s3.secretAccessKey }} + {{- end }} + {{- end }} + {{- end }} + command: ["/bin/sh", "-c"] + args: + - | + cd /opt/flink/flink-cdc-3.5.0 + ./bin/flink-cdc.sh \ + {{ .Values.job.pipelineFile }} \ + -Dexecution.checkpointing.interval={{ .Values.pipeline.checkpointInterval }} \ + -Drest.address={{ .Values.job.restAddress }} \ + -Drest.port={{ .Values.job.restPort }} + volumeMounts: + - name: pipeline-config + mountPath: /opt/flink/flink-cdc-3.5.0/conf + {{- if .Values.truststore.enabled }} + - name: truststore + mountPath: /opt/flink/certs + readOnly: true + {{- end }} + {{- if eq .Values.checkpointStorage.storageType "gcs" }} + - name: gcs-key + mountPath: /opt/flink/gcs-key + readOnly: true + {{- end }} + volumes: + - name: pipeline-config + configMap: + name: cdc-pipeline-config + {{- if .Values.truststore.enabled }} + - name: truststore + secret: + secretName: {{ .Values.truststore.secretName }} + {{- end }} + {{- if eq .Values.checkpointStorage.storageType "gcs" }} + - name: gcs-key + secret: + secretName: {{ .Values.checkpointStorage.gcs.credentialSecret }} + {{- end }} + restartPolicy: Never +{{- end }} diff --git a/manifests/helm/flink-cdc-session/0.1.0/templates/secrets.yaml b/manifests/helm/flink-cdc-session/0.1.0/templates/secrets.yaml new file mode 100644 index 0000000..f1f0d09 --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/templates/secrets.yaml @@ -0,0 +1,25 @@ +{{- if .Values.truststore.create }} +apiVersion: v1 +kind: Secret +metadata: + name: {{ .Values.truststore.secretName }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-session.labels" . | nindent 4 }} +type: Opaque +data: + truststore.jks: {{ required "truststore.base64jks is required when truststore.create is true" .Values.truststore.base64jks }} +{{- end }} +{{- if and (eq .Values.checkpointStorage.storageType "gcs") .Values.checkpointStorage.gcs.create }} +--- +apiVersion: v1 +kind: Secret +metadata: + name: {{ .Values.checkpointStorage.gcs.credentialSecret }} + namespace: {{ .Release.Namespace }} + labels: + {{- include "flink-cdc-session.labels" . | nindent 4 }} +type: Opaque +data: + key.json: {{ required "checkpointStorage.gcs.base64json is required when checkpointStorage.gcs.create is true" .Values.checkpointStorage.gcs.base64json }} +{{- end }} diff --git a/manifests/helm/flink-cdc-session/0.1.0/values.yaml b/manifests/helm/flink-cdc-session/0.1.0/values.yaml new file mode 100644 index 0000000..6d76f5f --- /dev/null +++ b/manifests/helm/flink-cdc-session/0.1.0/values.yaml @@ -0,0 +1,123 @@ +image: + repository: paasup/flink-cdc-pipeline + tag: 3.5.0-flink1.20-r1 + pullPolicy: Always + +flinkVersion: v1_20 +serviceAccount: flink + +ingress: + enabled: false + host: "flink-cdc-session.example.org" + className: "" + annotations: {} + +jobManager: + replicas: 1 + memory: 1024m + cpu: 0.5 + +taskManager: + replicas: 3 + memory: 2048m + cpu: 1 + taskSlots: 2 + +# ────────────────────────────────────────────────────────────── +# CheckpointStorage: 체크포인트/세이브포인트 저장 위치 + 자격증명 (WHERE) +# storageType에 해당하는 하위 블록(gcs 또는 s3)만 설정 +# ────────────────────────────────────────────────────────────── +checkpointStorage: + storageType: s3 # gcs | s3 + checkpointDir: s3://my-bucket/flink/checkpoints + savepointDir: s3://my-bucket/flink/savepoints + + # S3 (storageType: s3 일 때 설정) + # endpoint 비어있음 → AWS S3 / MinIO 등: https://minio.example.org + # accessKeyId/secretAccessKey 모두 비어있음 → IAM Role / IRSA 사용 + # existingSecret 지정 시 → 기존 Secret에서 자격증명 참조 (accessKeyId 값보다 우선) + s3: + endpoint: "" + pathStyleAccess: "false" + accessKeyId: "" + secretAccessKey: "" + existingSecret: "" + accessKeyIdKey: access-key-id + secretAccessKeyKey: secret-access-key + # GCS (storageType: gcs 일 때 설정) + gcs: + credentialSecret: gcs-key + credentialPath: /opt/flink/gcs-key/key.json + create: false + # create: true 시 필요: base64 인코딩된 key.json 파일 내용 + # base64json: "" + +# 체크포인트 동작 설정 (HOW) +checkpoint: + interval: 60s + mode: EXACTLY_ONCE + +# Secret 관리 +# create: false → 기존 Secret 참조 (kubectl create secret으로 별도 생성 필요) +# create: true → 차트가 Secret 리소스를 직접 생성 (base64 값 필수) +truststore: + # enabled: true → 외부 JKS를 사용. 싱크/소스 엔드포인트가 사설 CA 인증서를 사용할 때 필요. + # enabled: false → 외부 JKS 미사용. 모든 엔드포인트가 공인 CA 인증서를 사용할 때 설정. + # JVM 기본 cacerts로 TLS 검증을 수행하며, truststore Secret도 불필요. + enabled: true + secretName: flink-truststore + password: changeit + create: false + # create: true 시 필요: base64 인코딩된 truststore.jks 파일 내용 + # base64jks: "" + +# Flink CDC 글로벌 설정 (flink-cdc.yaml) +flinkCdc: + parallelism: 1 + schemaChangeBehavior: EVOLVE + +# PostgreSQL 소스 +postgres: + hostname: postgres-postgresql.flink-test.svc.cluster.local + port: 5432 + username: flink_cdc + password: password + slotName: flink_cdc_slot + decodingPlugin: pgoutput + tables: testdb.public.orders,testdb.public.users + +# 파이프라인 설정 +pipeline: + name: pg-to-iceberg-test + parallelism: 2 + checkpointInterval: 60s + +# Iceberg REST 카탈로그 싱크 +sink: + catalog: + uri: https://lakekeeper.example.org/catalog + warehouse: minio + s3Endpoint: https://minio.example.org + s3PathStyleAccess: "true" + oauth2Uri: https://keycloak.example.org/realms/paasup/protocol/openid-connect/token + credential: lakekeeper-admin:wMyE9WNn50DH8yycLXwLTmMmu66JU7GV + scope: lakekeeper + +# 소스→싱크 테이블 라우팅 +# source-table: PostgreSQL CDC emit 형식 (schema.table) +# sink-table: Iceberg namespace.table 형식 +route: + - sourceTable: public.orders + sinkTable: testdb_public.orders + - sourceTable: public.users + sinkTable: testdb_public.users + +# CDC 파이프라인 Job 제출 +# enabled: false → Session Cluster만 배포 (Job 생략) +# enabled: true → Session Cluster + Job 함께 배포 +job: + enabled: true + name: submit-cdc + restAddress: flink-cdc-session-rest + restPort: 8081 + pipelineFile: ./conf/pipeline.yaml