Move file path and Add flink sql gateway document

2026-01-30 15:57:37 +09:00
parent d4fecb4ae5
commit b2855ccba3
16 changed files with 366 additions and 4 deletions
@@ -0,0 +1,100 @@
+# flink-sql-gateway 버전 갱신 가이드
+
+## 1. git 작업 환경 구성
+
+- DIP 카탈로그 git 다운로드
+```
+$ git clone https://github.com/paasup/dip-catalog.git
+```
+
+- 작업 브랜치로 체크아웃
+```
+$ git checkout -b update-flink-sql-gateway/0.1.0
+```
+
+## 2. helm 차트 개발 및 업데이트
+
+- 현재 flink-sql-gateway 차트는 직접 작성하여 업데이트 중입니다.
+- 기존 오픈소스 차트를 기반으로 하지 않고, 프로젝트 요구사항에 맞게 커스텀 개발하고 있습니다.
+
+### 2.1 차트 구조
+```
+flink-sql-gateway/0.1.0/
+├── Chart.yaml                 # 차트 메타데이터
+├── values.yaml                # 기본 설정값
+├── custom-values.yaml         # 커스텀 설정값
+├── BUILD-README.md           # 빌드 가이드 (본 파일)
+├── CUSTOM-README.md          # 배포 가이드
+├── deployment.md             # 상세 배포 문서
+└── templates/
+    ├── _helpers.tpl          # 헬퍼 템플릿
+    ├── configmaps.yaml       # 설정 맵
+    ├── flink-session-cluster.yaml  # Flink 클러스터
+    ├── flink-sql-gateway.yaml      # SQL Gateway
+    ├── rbac.yaml            # 권한 설정
+    ├── services.yaml        # 서비스 정의
+    └── sql-client.yaml      # SQL 클라이언트
+```
+
+### 2.2 개발 중인 주요 기능
+- Apache Flink SQL Gateway 배포
+- Flink Session Cluster 통합
+- 외부 Python 클라이언트 접근 최적화
+- SSL/TLS 인증서 지원
+- Kafka 연동 설정
+- MinIO/S3 스토리지 연동
+
+## 3. git push 및 tag 추가
+
+- 갱신작업 진행후 commit
+```sh
+$ git add .
+$ git commit -m "update flink-sql-gateway/0.1.0"
+```
+
+- main 브랜치에 체크아웃 후 merge
+```sh
+$ git checkout main
+$ git merge update-flink-sql-gateway/0.1.0
+```
+
+- git에 push 후 작업 브랜치 삭제
+```sh
+$ git push -u origin main
+$ git branch -d update-flink-sql-gateway/0.1.0
+```
+
+- git tag 추가 후 push
+```sh
+$ git tag flink-sql-gateway/0.1.0
+$ git push origin flink-sql-gateway/0.1.0
+```
+
+## 4. 차트 버전 정보
+
+- flink-sql-gateway/0.1.0
+  - Apache Flink SQL Gateway 2.0.1 기반
+  - 외부 Python 클라이언트 접근에 최적화된 구성
+  - SSL/TLS 인증서 및 Truststore 지원
+  - Kafka, MinIO 연동 설정 포함
+  - 서비스 배포를 위하여 custom-values.yaml에 정의하였다.
+  - 차트의 빌드 방법과 배포 방법을 BUILD-README.md, CUSTOM-README.md 문서에 작성하였다.
+
+## 5. 개발 참고사항
+
+### 5.1 커스텀 이미지
+- 기본 이미지: `paasup/flink-sql:2.0.1`
+- Kafka 커넥터 및 SSL 인증서 지원 포함
+
+### 5.2 주요 의존성
+- Flink Kubernetes Operator 필요
+- Kubernetes 1.20+ 권장
+- Helm 3.x 필요
+
+### 5.3 보안 설정
+- Truststore 기반 SSL/TLS 인증서 관리
+- Kubernetes Secret을 통한 인증서 배포
+- 환경 변수를 통한 보안 정보 주입
+
+## 참고) 관련 문서
+- [Flink SQL Gateway 공식 문서](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/sql-gateway/)
@@ -0,0 +1,262 @@
+# Flink SQL Gateway 배포
+
+## 1. 배포 방법
+
+- 명령어
+
+  ``` sh
+  $ helm upgrade --install flink-sql-gateway ./ \
+    --namespace flink-sql-test \
+    --create-namespace \
+    -f custom-values.yaml
+  ```
+
+## 2. custom-values.yaml 설명
+
+- custom-values.yaml에 정의된 값에 대한 설명이다.
+
+### 1) 글로벌 설정
+| Name                   | 설명                                                         | 기본값 |
+| ---------------------- | ------------------------------------------------------------ | ------ |
+| `global.namespace` | 배포할 네임스페이스 | `flink-sql-test`   |
+| `global.image.repository` | Flink SQL Gateway 이미지 저장소 | `paasup/flink-sql` |
+| `global.image.tag` | Flink SQL Gateway 이미지 태그 | `2.0.1` |
+| `global.image.pullPolicy` | 이미지 풀 정책 | `IfNotPresent` |
+
+### 2) Flink Session Cluster 설정
+| Name                   | 설명                                                         | 기본값 |
+| ---------------------- | ------------------------------------------------------------ | ------ |
+| `sessionCluster.flinkVersion` | Flink 버전 | `v2_0` |
+| `sessionCluster.jobManager.resources.memory` | JobManager 메모리 할당량 | `2048m` |
+| `sessionCluster.jobManager.resources.cpu` | JobManager CPU 할당량 | `1` |
+| `sessionCluster.taskManager.resources.memory` | TaskManager 메모리 할당량 | `4096m` |
+| `sessionCluster.taskManager.resources.cpu` | TaskManager CPU 할당량 | `2` |
+| `sessionCluster.flinkConfiguration.taskmanager.numberOfTaskSlots` | TaskManager 슬롯 수 | `"4"` |
+
+### 3) SQL Gateway 설정
+| Name                   | 설명                                                         | 기본값 |
+| ---------------------- | ------------------------------------------------------------ | ------ |
+| `sqlGateway.resources.requests.memory` | SQL Gateway 요청 메모리 | `1Gi` |
+| `sqlGateway.resources.requests.cpu` | SQL Gateway 요청 CPU | `0.5` |
+| `sqlGateway.resources.limits.memory` | SQL Gateway 제한 메모리 | `2Gi` |
+| `sqlGateway.resources.limits.cpu` | SQL Gateway 제한 CPU | `1` |
+
+### 4) S3/MinIO 설정
+| Name                   | 설명                                                         | 기본값 |
+| ---------------------- | ------------------------------------------------------------ | ------ |
+| `sessionCluster.flinkConfiguration.fs.s3.impl` | S3 파일시스템 구현체 | `org.apache.hadoop.fs.s3a.S3AFileSystem` |
+| `sessionCluster.flinkConfiguration.fs.s3a.impl` | S3A 파일시스템 구현체 | `org.apache.hadoop.fs.s3a.S3AFileSystem` |
+| `sessionCluster.flinkConfiguration.fs.s3a.endpoint` | MinIO 엔드포인트 | `"https://minio.example.org"` |
+| `sessionCluster.flinkConfiguration.fs.s3a.path.style.access` | Path Style Access 사용 여부 | `"true"` |
+
+### 5) SQL Client 설정
+| Name                   | 설명                                                         | 기본값 |
+| ---------------------- | ------------------------------------------------------------ | ------ |
+| `sqlClient.enabled` | SQL Client 활성화 여부 | `true` |
+
+## 3. 사설 인증서 등록
+
+### 3.1 Kubernetes 클러스터
+- Kubernetes 1.20+ 권장
+- 충분한 리소스 (최소 10GB 메모리, 5 CPU 코어)
+
+### 3.2 필수 Operator 설치
+- Flink Kubernetes Operator 설치 필요
+  ```bash
+  helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.8.0/
+  helm install flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
+  ```
+
+### 3.3 SSL/TLS 인증서 설정
+- 외부 시스템(Kafka, MinIO 등)과의 HTTPS 통신을 위한 사설 인증서 등록이 필요합니다.
+- Flink의 모든 컴포넌트(JobManager, TaskManager, SQL Gateway)에서 동일한 Truststore를 사용합니다.
+
+#### 3.3.1 인증서 파일 준비
+포함되어야 할 인증서들:
+
+| Namespace | Secret Name | 파일 | 설명 |
+|-----------|-------------|------|------|
+| `kafka-cluster` | `kafka-cluster-cluster-ca-cert` | `ca.crt` | Kafka 클러스터 CA 인증서 |
+| `cert-manager` | `root-ca-secret` | `ca.crt` | Ingress에 사용되는 Root CA 인증서 |
+
+#### 3.3.2 인증서 추출 및 Truststore 생성
+```bash
+# 1. Kafka CA 인증서 추출
+kubectl get secret kafka-cluster-cluster-ca-cert -n kafka-cluster -o jsonpath='{.data.ca\.crt}' | base64 -d > kafka-ca.crt
+
+# 2. Root CA 인증서 추출
+kubectl get secret root-ca-secret -n cert-manager -o jsonpath='{.data.ca\.crt}' | base64 -d > root-ca.crt
+
+# 3. 빈 truststore 생성
+keytool -genkeypair -alias temp -keystore ca.p12 -storetype PKCS12 -storepass YOUR_TRUSTSTORE_PASSWORD -keypass YOUR_TRUSTSTORE_PASSWORD -dname "CN=temp" -keyalg RSA
+keytool -delete -alias temp -keystore ca.p12 -storetype PKCS12 -storepass YOUR_TRUSTSTORE_PASSWORD
+
+# 4. Kafka CA 인증서 추가
+keytool -import -trustcacerts -keystore ca.p12 \
+        -storetype PKCS12 \
+        -storepass YOUR_TRUSTSTORE_PASSWORD \
+        -alias kafka-ca \
+        -file kafka-ca.crt \
+        -noprompt
+
+# 5. Root CA 인증서 추가
+keytool -import -trustcacerts -keystore ca.p12 \
+        -storetype PKCS12 \
+        -storepass YOUR_TRUSTSTORE_PASSWORD \
+        -alias root-ca \
+        -file root-ca.crt \
+        -noprompt
+
+# 6. truststore 내용 확인
+keytool -list -keystore ca.p12 -storetype PKCS12 -storepass YOUR_TRUSTSTORE_PASSWORD
+
+# 7. Kubernetes Secret 생성
+kubectl create secret generic truststore-secret \
+  --from-file=ca.p12=ca.p12 \
+  --from-literal=ca.password=YOUR_TRUSTSTORE_PASSWORD \
+  -n flink-sql-test
+```
+
+#### 3.3.3 custom-values.yaml에서 SSL 설정 적용
+
+```yaml
+# Flink Session Cluster SSL 설정
+sessionCluster:
+  flinkConfiguration:
+    # JVM SSL 시스템 속성 설정
+    env.java.opts.jobmanager: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=YOUR_TRUSTSTORE_PASSWORD"
+    env.java.opts.taskmanager: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=YOUR_TRUSTSTORE_PASSWORD"
+  
+  # 환경 변수 설정
+  env:
+    - name: TRUSTSTORE_PASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: truststore-secret
+          key: ca.password
+  
+  # 볼륨 마운트 설정
+  volumeMounts:
+    - name: truststore-certs
+      mountPath: /opt/flink/certs
+      readOnly: true
+  
+  # 볼륨 정의
+  volumes:
+    - name: truststore-certs
+      secret:
+        secretName: truststore-secret
+
+# SQL Gateway SSL 설정
+sqlGateway:
+  flinkConfiguration:
+    # JVM SSL 시스템 속성 설정
+    env.java.opts: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=YOUR_TRUSTSTORE_PASSWORD"
+  
+  # 환경 변수, 볼륨 마운트, 볼륨 설정 (sessionCluster와 동일)
+  env:
+    - name: TRUSTSTORE_PASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: truststore-secret
+          key: ca.password
+  volumeMounts:
+    - name: truststore-certs
+      mountPath: /opt/flink/certs
+      readOnly: true
+  volumes:
+    - name: truststore-certs
+      secret:
+        secretName: truststore-secret
+
+# SQL Client SSL 설정 (선택적)
+sqlClient:
+  enabled: true
+  env:
+    - name: TRUSTSTORE_PASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: truststore-secret
+          key: ca.password
+  volumeMounts:
+    - name: truststore-certs
+      mountPath: /opt/flink/certs
+      readOnly: true
+  volumes:
+    - name: truststore-certs
+      secret:
+        secretName: truststore-secret
+```
+
+#### 3.3.4 SSL 설정 동작 원리
+1. **Secret 생성**: Truststore 파일(`ca.p12`)과 패스워드를 Kubernetes Secret으로 생성
+2. **볼륨 마운트**: Secret을 각 컴포넌트의 `/opt/flink/certs` 경로에 마운트
+3. **환경 변수 주입**: Secret의 패스워드를 `TRUSTSTORE_PASSWORD` 환경 변수로 주입
+4. **JVM 옵션 설정**: Java 시스템 속성을 통해 SSL Truststore 경로와 설정 지정
+5. **SSL 통신**: 외부 시스템(Kafka, MinIO 등)과의 HTTPS 통신 시 Truststore 사용
+
+## 4. 배포 후 확인
+
+### 4.1 Pod 상태 확인
+```bash
+kubectl get pods -n flink-sql-test
+```
+
+### 4.2 서비스 접근
+```bash
+# SQL Gateway API 접근을 위한 포트 포워딩
+kubectl port-forward svc/flink-sql-gateway 8083:8083 -n flink-sql-test
+
+# Flink Web UI 접근을 위한 포트 포워딩
+kubectl port-forward svc/flink-session-cluster-rest 8081:8081 -n flink-sql-test
+```
+
+### 4.3 API 테스트
+- pod 내 실행
+```bash
+# SQL Gateway 정보 확인
+curl http://localhost:8083/v1/info
+
+# 세션 생성 테스트
+curl -X POST http://localhost:8083/v1/sessions \
+  -H "Content-Type: application/json" \
+  -d '{"properties": {"execution.runtime-mode": "streaming"}}'
+```
+
+## 5. Python 클라이언트 사용 예제
+
+```python
+import requests
+import json
+
+# SQL Gateway 연결
+gateway_url = "http://localhost:8083"
+
+# 1. 세션 생성
+response = requests.post(f"{gateway_url}/v1/sessions")
+session_handle = response.json()["sessionHandle"]
+print(f"Session created: {session_handle}")
+
+# 2. SQL 실행
+sql_request = {
+    "statement": "SHOW TABLES"
+}
+response = requests.post(
+    f"{gateway_url}/v1/sessions/{session_handle}/statements",
+    json=sql_request
+)
+operation_handle = response.json()["operationHandle"]
+
+# 3. 결과 조회
+result_response = requests.get(
+    f"{gateway_url}/v1/sessions/{session_handle}/operations/{operation_handle}/result/0"
+)
+print("Query result:", result_response.json())
+```
+
+## 6. 주요 특징
+
+- **외부 Python 클라이언트 최적화**: Port Forward를 통한 직접 REST API 접근
+- **SSL/TLS 지원**: 사설 인증서를 사용하는 외부 시스템 연동 지원
+- **S3/MinIO 연동**: 객체 스토리지 연동을 위한 설정 포함
+- **유연한 구성**: 필요한 컴포넌트만 선택적 활성화 가능
@@ -0,0 +1,21 @@
+annotations:
+  category: Analytics
+  licenses: Apache-2.0
+apiVersion: v2
+appVersion: 2.0.1
+description: A Helm chart for Flink SQL Gateway with Kafka integration
+home: https://github.com/your-org/flink-sql-gateway-helm
+keywords:
+- flink
+- sql
+- gateway
+- kafka
+- streaming
+maintainers:
+- email: team@example.com
+  name: Flink SQL Gateway Team
+name: flink-sql-gateway
+sources:
+- https://github.com/your-org/flink-sql-gateway-helm
+type: application
+version: 0.1.0
@@ -0,0 +1,411 @@
+# Flink SQL Gateway Helm Chart
+
+Apache Flink SQL Gateway를 쉽게 배포하기 위한 Helm Chart입니다. 외부 Python 클라이언트 접근에 최적화되어 있습니다.
+
+## 📋 개요
+
+이 Helm Chart는 다음 구성 요소를 배포합니다:
+
+### 필수 컴포넌트 (항상 배포됨)
+- **Flink Session Cluster**: SQL Gateway가 연결할 Flink 클러스터
+- **Flink SQL Gateway**: SQL 쿼리를 실행할 수 있는 REST API 서버
+- **Service**: SQL Gateway 접근을 위한 Kubernetes 서비스
+- **RBAC**: Flink 리소스 관리를 위한 권한 설정
+- **ConfigMap**: SQL Gateway 설정
+
+### 선택적 컴포넌트 (설정으로 활성화)
+- **SQL Client Pod**: 대화형 SQL 클라이언트 (`sqlClient.enabled: true`)
+- **Kafka Integration**: Kafka 연동 설정 (`kafka.enabled: true`)
+- **SQL Scripts ConfigMap**: SQL 스크립트 모음 (SQL Client 활성화 시)
+
+## 🏗️ 아키텍처
+
+### 외부 Python 클라이언트 접근 (권장)
+```
+┌──────────────────┐
+│  Python Client   │
+│  (External)      │
+└──────────────────┘
+         │
+         ▼ HTTP REST API
+┌──────────────────┐
+│   SQL Gateway    │
+│   (REST API)     │
+│   Port: 8083     │
+└──────────────────┘
+         │
+         ▼ SQL 실행 요청
+┌─────────────────────┐
+│  Flink Session      │
+│  Cluster            │
+│  (Data Processing)  │
+└─────────────────────┘
+         │
+         ▼ 데이터 처리 (선택적)
+┌─────────────────┐
+│   Kafka Cluster │
+│   (Streaming)   │
+└─────────────────┘
+```
+
+### 내부 SQL Client 사용 (선택적)
+```
+┌──────────────────┐
+│   SQL Client     │
+│   (Pod)          │
+└──────────────────┘
+         │
+         ▼ SQL 쿼리 전송
+┌──────────────────┐
+│   SQL Gateway    │
+│   (REST API)     │
+└──────────────────┘
+         │
+         ▼ SQL 실행 요청
+┌─────────────────────┐
+│  Flink Session      │
+│  Cluster            │
+└─────────────────────┘
+```
+
+**주요 특징:**
+- **외부 Python 클라이언트 최적화**: Port Forward를 통한 직접 REST API 접근
+- **유연한 구성**: 필요한 컴포넌트만 선택적 배포
+- **Session Mode**: 대화형 SQL 개발 환경 제공
+- **검증된 안정성**: YAML 파싱 오류 해결 및 템플릿 검증 완료
+
+## 🚀 빠른 시작
+
+### 1. 사전 요구사항
+
+- Kubernetes 클러스터
+- Helm 3.x
+- Flink Kubernetes Operator 설치됨
+- Strimzi Kafka Operator 설치됨 (Kafka 사용 시)
+
+### 2. 설치
+
+#### 기본 설치 (외부 Python 클라이언트용 - 권장)
+```bash
+# 최소 리소스로 기본 설치
+helm install flink-sql-gateway ./ \
+  --namespace flink-sql-gateway \
+  --create-namespace
+
+# 외부 Python 클라이언트용 최적화 설치 (권장)
+helm install flink-sql-gateway ./ \
+  --namespace flink-sql-gateway \
+  --create-namespace \
+  --values ./custom-values.yaml
+```
+
+#### 고급 설치 옵션
+```bash
+# SQL Client 포함 설치 (대화형 SQL 사용)
+helm install flink-sql-gateway ./ \
+  --namespace flink-sql-gateway \
+  --create-namespace \
+  --set sqlClient.enabled=true
+
+# Kafka 연동 포함 설치
+helm install flink-sql-gateway ./ \
+  --namespace flink-sql-gateway \
+  --create-namespace \
+  --set kafka.enabled=true \
+  --set kafka.user.password="your-kafka-password"
+
+# 완전한 환경 설치 (모든 컴포넌트)
+helm install flink-sql-gateway ./ \
+  --namespace flink-sql-gateway \
+  --create-namespace \
+  --values ./custom-values.yaml \
+  --set sqlClient.enabled=true \
+  --set kafka.enabled=true \
+  --set kafka.user.password="your-kafka-password"
+```
+
+### 3. 외부 Python 클라이언트 접근 (권장)
+
+#### Port Forward 설정
+```bash
+# SQL Gateway API 접근
+kubectl port-forward -n flink-sql-gateway svc/flink-sql-gateway 8083:8083
+
+# Flink Web UI 접근 (선택적)
+kubectl port-forward -n flink-sql-gateway svc/flink-session-cluster-rest 8081:8081
+```
+
+#### Python 클라이언트 예제
+```python
+import requests
+import json
+
+# SQL Gateway 연결
+gateway_url = "http://localhost:8083"
+
+# 1. 세션 생성
+response = requests.post(f"{gateway_url}/v1/sessions")
+session_handle = response.json()["sessionHandle"]
+print(f"Session created: {session_handle}")
+
+# 2. SQL 실행
+sql_request = {
+    "statement": "SHOW TABLES"
+}
+response = requests.post(
+    f"{gateway_url}/v1/sessions/{session_handle}/statements",
+    json=sql_request
+)
+operation_handle = response.json()["operationHandle"]
+
+# 3. 결과 조회
+result_response = requests.get(
+    f"{gateway_url}/v1/sessions/{session_handle}/operations/{operation_handle}/result/0"
+)
+print("Query result:", result_response.json())
+```
+
+### 4. 내부 SQL Client 접근 (선택적)
+
+```bash
+# SQL Client에 연결 (sqlClient.enabled=true인 경우)
+kubectl exec -it flink-sql-client -n flink-sql-gateway -- \
+  /opt/flink/bin/sql-client.sh gateway \
+  --endpoint http://flink-sql-gateway:8083
+```
+
+## ⚙️ 설정
+
+### 주요 설정 항목
+
+| 파라미터 | 설명 | 기본값 | custom-values.yaml |
+|----------|------|--------|-------------------|
+| `global.namespace` | 배포할 네임스페이스 | `flink-sql-gateway` | `flink-sql-gateway` |
+| `global.image.repository` | Flink 이미지 저장소 | `paasup/flink-sql` | `paasup/flink-sql` |
+| `global.image.tag` | Flink 이미지 태그 | `1.20-kafka` | `1.20-kafka` |
+| `sessionCluster.enabled` | Session Cluster 활성화 | `true` | `true` |
+| `sqlGateway.enabled` | SQL Gateway 활성화 | `true` | `true` |
+| `sqlClient.enabled` | SQL Client 활성화 | `false` | `false` |
+| `kafka.enabled` | Kafka 연동 활성화 | `false` | `false` |
+
+### 리소스 설정 비교
+
+#### 기본 설정 (values.yaml)
+```yaml
+# 최소 리소스 - 테스트용
+sessionCluster:
+  jobManager:
+    resources:
+      memory: 1024m
+      cpu: 0.5
+  taskManager:
+    resources:
+      memory: 2048m
+      cpu: 1
+  flinkConfiguration:
+    taskmanager.numberOfTaskSlots: "2"
+
+sqlGateway:
+  resources:
+    requests:
+      memory: 512Mi
+      cpu: 0.25
+    limits:
+      memory: 1Gi
+      cpu: 0.5
+```
+
+#### 최적화 설정 (custom-values.yaml)
+```yaml
+# 외부 Python 클라이언트용 최적화 - 프로덕션 권장
+sessionCluster:
+  jobManager:
+    resources:
+      memory: 2048m  # 2배 증가
+      cpu: 1         # 2배 증가
+  taskManager:
+    resources:
+      memory: 4096m  # 2배 증가
+      cpu: 2         # 2배 증가
+  flinkConfiguration:
+    taskmanager.numberOfTaskSlots: "4"  # 2배 증가
+
+sqlGateway:
+  resources:
+    requests:
+      memory: 1Gi    # 2배 증가
+      cpu: 0.5       # 2배 증가
+    limits:
+      memory: 2Gi    # 2배 증가
+      cpu: 1         # 2배 증가
+
+# SQL Client 비활성화 (외부 Python 클라이언트 사용)
+sqlClient:
+  enabled: false
+```
+
+### Kafka 설정 (선택적)
+
+```yaml
+kafka:
+  enabled: true
+  bootstrapServers: "kafka-cluster-kafka-external-bootstrap.kafka.svc.cluster.local:9094"
+  topics:
+    input:
+      name: test.input
+    output:
+      name: test.output
+  user:
+    name: flink-sql-gateway
+    password: "your-kafka-password"
+    # 또는 External Secret 사용
+    externalSecret:
+      enabled: true
+      secretName: "kafka-user-credentials"
+      usernameKey: "username"
+      passwordKey: "password"
+```
+
+### External Secret 설정
+
+```yaml
+kafka:
+  user:
+    externalSecret:
+      enabled: true
+      secretName: "flink-sql-gateway"  # Kafka namespace의 기존 secret
+      usernameKey: "username"
+      passwordKey: "password"
+```
+
+## 📝 사용 예시
+
+### 1. 테이블 생성
+
+```sql
+-- 소스 테이블 생성
+CREATE TABLE test_input (
+    id STRING,
+    message STRING,
+    timestamp_field TIMESTAMP(3),
+    WATERMARK FOR timestamp_field AS timestamp_field - INTERVAL '5' SECOND
+) WITH (
+    'connector' = 'kafka',
+    'topic' = 'test.input',
+    'properties.bootstrap.servers' = 'kafka-cluster-kafka-external-bootstrap.kafka.svc.cluster.local:9094',
+    'properties.group.id' = 'flink-sql-gateway',
+    'properties.security.protocol' = 'SASL_SSL',
+    'properties.sasl.mechanism' = 'SCRAM-SHA-512',
+    'properties.sasl.jaas.config' = 'org.apache.kafka.common.security.scram.ScramLoginModule required username="flink-sql-gateway" password="YOUR_PASSWORD";',
+    'properties.ssl.truststore.location' = '/opt/flink/certs/truststore.jks',
+    'properties.ssl.truststore.password' = 'changeit',
+    'properties.ssl.truststore.type' = 'JKS',
+    'format' = 'json',
+    'json.timestamp-format.standard' = 'ISO-8601'
+);
+```
+
+### 2. SQL 쿼리 실행
+
+```sql
+-- 간단한 필터링 쿼리
+INSERT INTO test_output
+SELECT 
+    id,
+    CONCAT('Processed: ', message) as processed_message,
+    timestamp_field as original_timestamp,
+    CURRENT_TIMESTAMP as processing_time
+FROM test_input
+WHERE LENGTH(message) > 5;
+```
+
+## 🔧 관리
+
+### 업그레이드
+
+```bash
+helm upgrade flink-sql-gateway . \
+  --namespace flink-sql-gateway \
+  --values values.yaml
+```
+
+### 제거
+
+```bash
+helm uninstall flink-sql-gateway --namespace flink-sql-gateway
+```
+
+### 상태 확인
+
+```bash
+# Helm 릴리스 상태
+helm status flink-sql-gateway -n flink-sql-gateway
+
+# Pod 상태
+kubectl get pods -n flink-sql-gateway
+
+# FlinkDeployment 상태
+kubectl get flinkdeployment -n flink-sql-gateway
+
+# 로그 확인
+kubectl logs -n flink-sql-gateway -l app=flink-sql-gateway
+```
+
+## 🐛 문제 해결
+
+### 일반적인 문제
+
+1. **SQL Gateway 연결 실패**
+   ```bash
+   # Gateway 상태 확인
+   kubectl get deployment flink-sql-gateway -n flink-sql-gateway
+   kubectl logs -n flink-sql-gateway -l app=flink-sql-gateway
+   
+   # Port Forward 확인
+   kubectl port-forward -n flink-sql-gateway svc/flink-sql-gateway 8083:8083
+   curl http://localhost:8083/v1/info
+   ```
+
+2. **Python 클라이언트 연결 실패**
+   ```python
+   # 연결 테스트
+   import requests
+   try:
+       response = requests.get("http://localhost:8083/v1/info", timeout=5)
+       print("SQL Gateway 연결 성공:", response.json())
+   except requests.exceptions.ConnectionError:
+       print("Port Forward가 실행 중인지 확인하세요")
+   except requests.exceptions.Timeout:
+       print("SQL Gateway가 시작되지 않았을 수 있습니다")
+   ```
+
+3. **Flink Session Cluster 시작 실패**
+   ```bash
+   # FlinkDeployment 상태 확인
+   kubectl describe flinkdeployment flink-session-cluster -n flink-sql-gateway
+   ```
+
+4. **Kafka 연결 오류**
+   - Kafka 사용자 비밀번호 확인
+   - 인증서 설정 확인
+   - 네트워크 연결 확인
+
+### 디버깅 명령어
+
+```bash
+# 전체 환경 상태 확인
+kubectl get all -n flink-sql-gateway
+
+# 이벤트 확인
+kubectl get events -n flink-sql-gateway --sort-by='.lastTimestamp'
+
+# Helm 템플릿 확인
+helm template flink-sql-gateway . --values values.yaml
+```
+
+## 📚 참고 자료
+
+- [Apache Flink Documentation](https://nightlies.apache.org/flink/flink-docs-release-1.20/)
+- [Flink SQL Gateway Documentation](https://nightlies.apache.org/flink/flink-docs-release-1.20/docs/dev/table/sql-gateway/)
+- [Flink Kubernetes Operator](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-release-1.13/)
+- [Helm Documentation](https://helm.sh/docs/)
@@ -0,0 +1,109 @@
+# Custom Values for Flink SQL Gateway Helm Chart
+# Configuration for external Python client access with resource settings
+
+global:
+  namespace: flink-sql-test
+  image:
+    repository: wbsong111/flink-sql
+    tag: 2.0.1
+    pullPolicy: IfNotPresent
+
+
+# Flink Session Cluster resource configuration
+sessionCluster:
+  flinkVersion: v2_0
+  jobManager:
+    resources:
+      memory: 2048m  # Increased from default 1024m
+      cpu: 1         # Increased from default 0.5
+  taskManager:
+    resources:
+      memory: 4096m  # Increased from default 2048m
+      cpu: 2         # Increased from default 1
+  flinkConfiguration:
+    taskmanager.numberOfTaskSlots: "4"  # Increased from default "2"
+    # SSL/TLS Configuration for truststore
+    # security.ssl.internal.enabled: "true"
+    # security.ssl.internal.truststore: "/opt/flink/certs/ca.p12"
+    # security.ssl.internal.truststore-type: "PKCS12"
+    # JVM SSL system properties for JobManager and TaskManager
+    env.java.opts.jobmanager: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=YOUR_TRUSTSTORE_PASSWORD"
+    env.java.opts.taskmanager: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=YOUR_TRUSTSTORE_PASSWORD"
+    # Iceberg REST Catalog SSL Configuration
+    table.exec.resource.default-parallelism: "2"
+    
+    # [추가] s3 스키마를 Hadoop S3AFileSystem으로 처리하도록 매핑
+    fs.s3.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
+    fs.s3a.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
+    
+    # [추가] MinIO 사용 시 Path Style Access 필수
+    fs.s3a.endpoint: "https://minio.example.org"
+    fs.s3a.path.style.access: "true"
+
+  env:
+    - name: TRUSTSTORE_PASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: truststore-secret
+          key: ca.password
+  volumeMounts:
+    - name: truststore-certs
+      mountPath: /opt/flink/certs
+      readOnly: true
+  volumes:
+    - name: truststore-certs
+      secret:
+        secretName: truststore-secret
+
+# SQL Gateway resource configuration
+sqlGateway:
+  resources:
+    requests:
+      memory: 1Gi    # Increased from default 512Mi
+      cpu: 0.5       # Increased from default 0.25
+    limits:
+      memory: 2Gi    # Increased from default 1Gi
+      cpu: 1         # Increased from default 0.5
+  flinkConfiguration:
+    # JVM SSL system properties for SQL Gateway
+    # env.java.opts.all: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=${TRUSTSTORE_PASSWORD}"
+    env.java.opts: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=YOUR_TRUSTSTORE_PASSWORD"
+  env:
+    - name: TRUSTSTORE_PASSWORD
+      valueFrom:
+        secretKeyRef:
+          name: truststore-secret
+          key: ca.password
+  volumeMounts:
+    - name: truststore-certs
+      mountPath: /opt/flink/certs
+      readOnly: true
+  volumes:
+    - name: truststore-certs
+      secret:
+        secretName: truststore-secret
+
+sqlClient:
+   enabled: true
+   env:
+     - name: TRUSTSTORE_PASSWORD
+       valueFrom:
+         secretKeyRef:
+           name: truststore-secret
+           key: ca.password
+   volumeMounts:
+     - name: truststore-certs
+       mountPath: /opt/flink/certs
+       readOnly: true
+   volumes:
+     - name: truststore-certs
+       secret:
+         secretName: truststore-secret
+# Optional: Enable Ingress for external access
+# ingress:
+#   enabled: true
+#   hosts:
+#     - host: flink-sql-gateway.example.com
+#       paths:
+#         - path: /
+#           pathType: Prefix
@@ -0,0 +1,107 @@
+{{/*
+Expand the name of the chart.
+*/}}
+{{- define "flink-sql-gateway.name" -}}
+{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Create a default fully qualified app name.
+We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
+If release name contains chart name it will be used as a full name.
+*/}}
+{{- define "flink-sql-gateway.fullname" -}}
+{{- if .Values.fullnameOverride }}
+{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- $name := default .Chart.Name .Values.nameOverride }}
+{{- if contains $name .Release.Name }}
+{{- .Release.Name | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
+{{- end }}
+{{- end }}
+{{- end }}
+
+{{/*
+Create chart name and version as used by the chart label.
+*/}}
+{{- define "flink-sql-gateway.chart" -}}
+{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Common labels
+*/}}
+{{- define "flink-sql-gateway.labels" -}}
+helm.sh/chart: {{ include "flink-sql-gateway.chart" . }}
+{{ include "flink-sql-gateway.selectorLabels" . }}
+{{- if .Chart.AppVersion }}
+app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
+{{- end }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- with .Values.global.labels }}
+{{ toYaml . }}
+{{- end }}
+{{- end }}
+
+{{/*
+Selector labels
+*/}}
+{{- define "flink-sql-gateway.selectorLabels" -}}
+app.kubernetes.io/name: {{ include "flink-sql-gateway.name" . }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+{{- end }}
+
+{{/*
+Create the name of the service account to use
+*/}}
+{{- define "flink-sql-gateway.serviceAccountName" -}}
+{{- if .Values.rbac.create }}
+{{- default (include "flink-sql-gateway.fullname" .) .Values.rbac.serviceAccountName }}
+{{- else }}
+{{- default "default" .Values.rbac.serviceAccountName }}
+{{- end }}
+{{- end }}
+
+{{/*
+Create namespace name
+*/}}
+{{- define "flink-sql-gateway.namespace" -}}
+{{- default .Values.global.namespace .Release.Namespace }}
+{{- end }}
+
+{{/*
+Create image name
+*/}}
+{{- define "flink-sql-gateway.image" -}}
+{{- printf "%s:%s" .Values.global.image.repository .Values.global.image.tag }}
+{{- end }}
+
+
+{{/*
+Session Cluster labels
+*/}}
+{{- define "flink-sql-gateway.sessionCluster.labels" -}}
+{{ include "flink-sql-gateway.labels" . }}
+app: {{ .Values.sessionCluster.name }}
+component: flink-session-cluster
+{{- end }}
+
+{{/*
+SQL Gateway labels
+*/}}
+{{- define "flink-sql-gateway.sqlGateway.labels" -}}
+{{ include "flink-sql-gateway.labels" . }}
+app: {{ .Values.sqlGateway.name }}
+component: flink-sql-gateway
+{{- end }}
+
+{{/*
+SQL Client labels
+*/}}
+{{- define "flink-sql-gateway.sqlClient.labels" -}}
+{{ include "flink-sql-gateway.labels" . }}
+app: {{ .Values.sqlClient.name }}
+component: flink-sql-client
+{{- end }}
@@ -0,0 +1,148 @@
+---
+# ConfigMap for SQL Gateway Configuration
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: sql-gateway-config
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.labels" . | nindent 4 }}
+  {{- with .Values.global.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+data:
+  sql-gateway-defaults.yaml: |
+    # SQL Gateway Configuration
+    sql-gateway:
+      endpoint:
+        rest:
+          address: {{ .Values.sqlGateway.config.endpoint.rest.address }}
+          port: {{ .Values.sqlGateway.config.endpoint.rest.port }}
+          bind-address: {{ .Values.sqlGateway.config.endpoint.rest.bindAddress }}
+      session:
+        max-num: {{ .Values.sqlGateway.config.session.maxNum }}
+        idle-timeout: {{ .Values.sqlGateway.config.session.idleTimeout }}
+        check-interval: {{ .Values.sqlGateway.config.session.checkInterval }}
+        plan-cache:
+          enabled: {{ .Values.sqlGateway.config.session.planCache.enabled }}
+          max-size: {{ .Values.sqlGateway.config.session.planCache.maxSize }}
+          ttl: {{ .Values.sqlGateway.config.session.planCache.ttl }}
+{{- if .Values.sqlClient.enabled }}
+---
+# ConfigMap with sample SQL scripts
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: sql-test-scripts
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.labels" . | nindent 4 }}
+  {{- with .Values.global.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+data:
+  01-create-tables.sql: |
+    -- Example SQL script for creating tables
+    -- This is a template - modify according to your data sources
+    
+    -- Example: Create source table (replace with your connector)
+    CREATE TABLE source_table (
+        id STRING,
+        message STRING,
+        timestamp_field TIMESTAMP(3),
+        WATERMARK FOR timestamp_field AS timestamp_field - INTERVAL '5' SECOND
+    ) WITH (
+        'connector' = 'your-connector',  -- e.g., 'kafka', 'filesystem', etc.
+        -- Add your connector-specific properties here
+        'format' = 'json'
+    );
+
+    -- Example: Create sink table (replace with your connector)
+    CREATE TABLE sink_table (
+        id STRING,
+        processed_message STRING,
+        original_timestamp TIMESTAMP(3),
+        processing_time TIMESTAMP(3)
+    ) WITH (
+        'connector' = 'your-connector',  -- e.g., 'kafka', 'filesystem', etc.
+        -- Add your connector-specific properties here
+        'format' = 'json'
+    );
+
+  02-simple-query.sql: |
+    -- Simple filtering and transformation query
+    INSERT INTO sink_table
+    SELECT
+        id,
+        CONCAT('Processed: ', message) as processed_message,
+        timestamp_field as original_timestamp,
+        CURRENT_TIMESTAMP as processing_time
+    FROM source_table
+    WHERE LENGTH(message) > 5;
+
+  03-windowed-aggregation.sql: |
+    -- Windowed aggregation example
+    SELECT
+        TUMBLE_START(timestamp_field, INTERVAL '1' MINUTE) as window_start,
+        TUMBLE_END(timestamp_field, INTERVAL '1' MINUTE) as window_end,
+        COUNT(*) as message_count,
+        COUNT(DISTINCT id) as unique_ids
+    FROM source_table
+    GROUP BY TUMBLE(timestamp_field, INTERVAL '1' MINUTE);
+
+  test-data-generator.sql: |
+    -- Generate test data (for manual testing)
+    -- This would typically be done by external producer
+    -- Example JSON format for input data:
+    -- {"id": "msg001", "message": "Hello World", "timestamp_field": "2024-12-10T10:00:00.000Z"}
+    -- {"id": "msg002", "message": "Test Message", "timestamp_field": "2024-12-10T10:01:00.000Z"}
+
+  README.md: |
+    # Flink SQL Client Usage
+
+    ## Connect to SQL Gateway with Remote Cluster
+    ```bash
+    # Connect to SQL Gateway and specify remote cluster
+    kubectl exec -it {{ .Values.sqlClient.name }} -n {{ include "flink-sql-gateway.namespace" . }} -- /opt/flink/bin/sql-client.sh gateway \
+      --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }} \
+      -Dexecution.target=remote \
+      -Drest.address={{ .Values.sessionCluster.name }}-rest \
+      -Drest.port=8081
+    ```
+
+    ## Alternative: Set execution config in SQL
+    ```bash
+    # Connect to SQL Gateway
+    kubectl exec -it {{ .Values.sqlClient.name }} -n {{ include "flink-sql-gateway.namespace" . }} -- /opt/flink/bin/sql-client.sh gateway --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}
+
+    # Then in SQL Client, set execution config:
+    SET 'execution.target' = 'remote';
+    SET 'rest.address' = '{{ .Values.sessionCluster.name }}-rest';
+    SET 'rest.port' = '8081';
+    ```
+
+    ## Run SQL Scripts
+    ```bash
+    # Inside the SQL Client container
+    kubectl exec -it {{ .Values.sqlClient.name }} -n {{ include "flink-sql-gateway.namespace" . }} -- bash
+
+    # View available scripts
+    ls /opt/sql-scripts/
+
+    # Execute SQL file with remote cluster config
+    /opt/flink/bin/sql-client.sh gateway \
+      --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }} \
+      -Dexecution.target=remote \
+      -Drest.address={{ .Values.sessionCluster.name }}-rest \
+      -Drest.port=8081 \
+      -f /opt/sql-scripts/01-create-tables.sql
+    ```
+
+    ## Test Data
+    Send JSON messages to your input source:
+    ```json
+    {"id": "msg001", "message": "Hello World", "timestamp_field": "2024-12-10T10:00:00.000Z"}
+    ```
+{{- end }}
@@ -0,0 +1,67 @@
+{{- if .Values.sessionCluster.enabled }}
+---
+# Flink Session Cluster (for SQL Gateway to connect to)
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkDeployment
+metadata:
+  name: {{ .Values.sessionCluster.name }}
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.sessionCluster.labels" . | nindent 4 }}
+  {{- with .Values.global.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  image: {{ include "flink-sql-gateway.image" . }}
+  flinkVersion: {{ .Values.sessionCluster.flinkVersion }}
+  flinkConfiguration:
+    {{- toYaml .Values.sessionCluster.flinkConfiguration | nindent 4 }}
+  serviceAccount: {{ include "flink-sql-gateway.serviceAccountName" . }}
+  jobManager:
+    resource:
+      memory: {{ .Values.sessionCluster.jobManager.resources.memory }}
+      cpu: {{ .Values.sessionCluster.jobManager.resources.cpu }}
+    replicas: {{ .Values.sessionCluster.jobManager.replicas }}
+  taskManager:
+    resource:
+      memory: {{ .Values.sessionCluster.taskManager.resources.memory }}
+      cpu: {{ .Values.sessionCluster.taskManager.resources.cpu }}
+    replicas: {{ .Values.sessionCluster.taskManager.replicas }}
+  podTemplate:
+    spec:
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.podSecurityContext }}
+      securityContext:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      containers:
+        - name: flink-main-container
+          {{- with .Values.securityContext }}
+          securityContext:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          {{- with .Values.sessionCluster.env }}
+          env:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          {{- with .Values.sessionCluster.volumeMounts }}
+          volumeMounts:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+      {{- with .Values.sessionCluster.volumes }}
+      volumes:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+{{- end }}
@@ -0,0 +1,173 @@
+{{- if .Values.sqlGateway.enabled }}
+---
+# SQL Gateway Deployment (separate from Flink cluster)
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .Values.sqlGateway.name }}
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.sqlGateway.labels" . | nindent 4 }}
+  {{- with .Values.global.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  replicas: {{ .Values.sqlGateway.replicas }}
+  selector:
+    matchLabels:
+      app: {{ .Values.sqlGateway.name }}
+  template:
+    metadata:
+      labels:
+        app: {{ .Values.sqlGateway.name }}
+        {{- include "flink-sql-gateway.selectorLabels" . | nindent 8 }}
+    spec:
+      serviceAccountName: {{ include "flink-sql-gateway.serviceAccountName" . }}
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.podSecurityContext }}
+      securityContext:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      containers:
+        - name: sql-gateway
+          image: {{ include "flink-sql-gateway.image" . }}
+          imagePullPolicy: {{ .Values.global.image.pullPolicy }}
+          {{- with .Values.securityContext }}
+          securityContext:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          command: ["/bin/bash"]
+          args:
+            - -c
+            - |
+              # Wait for Flink session cluster to be ready with timeout
+              echo "Waiting for Flink session cluster..."
+              TIMEOUT=300  # 5 minutes
+              ELAPSED=0
+              INTERVAL=5
+              
+              while [ $ELAPSED -lt $TIMEOUT ]; do
+                if curl -f --connect-timeout 5 --max-time 10 http://{{ .Values.sessionCluster.name }}-rest:8081/v1/overview >/dev/null 2>&1; then
+                  echo "Flink session cluster is ready!"
+                  break
+                fi
+                echo "Waiting for Flink cluster... (${ELAPSED}s/${TIMEOUT}s)"
+                sleep $INTERVAL
+                ELAPSED=$((ELAPSED + INTERVAL))
+              done
+              
+              if [ $ELAPSED -ge $TIMEOUT ]; then
+                echo "ERROR: Timeout waiting for Flink session cluster"
+                echo "Checking cluster status..."
+                kubectl get pods -n {{ include "flink-sql-gateway.namespace" . }} -l app={{ .Values.sessionCluster.name }} || true
+                exit 1
+              fi
+              
+              echo "Flink cluster is ready. Starting SQL Gateway..."
+              
+              # Create flink configuration with environment variable substitution
+              cat > /opt/flink/conf/config.yaml << EOF
+              # Remote Flink Cluster Configuration
+              rest.bind-address: 0.0.0.0
+              rest.address: {{ .Values.sessionCluster.name }}-rest
+              rest.port: 8081
+              jobmanager.rpc.address: {{ .Values.sessionCluster.name }}
+              jobmanager.rpc.port: 6123
+              
+              # Session Cluster at Kubernetes
+              execution.target: kubernetes-session
+              kubernetes.namespace: {{ .Release.Namespace }}
+              kubernetes.cluster-id: {{ .Values.sessionCluster.name }}
+              
+              # SQL Gateway Configuration
+              sql-gateway.endpoint.rest.address: 0.0.0.0
+              sql-gateway.endpoint.rest.port: {{ .Values.sqlGateway.port }}
+              
+              # Table/SQL Configuration
+              table.exec.source.idle-timeout: 30s
+              table.exec.resource.default-parallelism: 1
+              
+              # Additional Flink Configuration from values
+              {{- with .Values.sqlGateway.flinkConfiguration }}
+              {{- range $key, $value := . }}
+              {{ $key }}: {{ $value }}
+              {{- end }}
+              {{- end }}
+              EOF
+              
+              # Start SQL Gateway
+              /opt/flink/bin/sql-gateway.sh start
+              
+              echo "SQL Gateway started successfully"
+              
+              # Wait for log file to be created and then tail it
+              LOG_FILE=""
+              for i in {1..30}; do
+                LOG_FILE=$(find /opt/flink/log -name "flink--sql-gateway-*.log" 2>/dev/null | head -1)
+                if [ -n "$LOG_FILE" ]; then
+                  echo "Found log file: $LOG_FILE"
+                  break
+                fi
+                echo "Waiting for log file to be created... ($i/30)"
+                sleep 2
+              done
+              
+              if [ -n "$LOG_FILE" ]; then
+                echo "Tailing SQL Gateway log file..."
+                tail -f "$LOG_FILE"
+              else
+                echo "Log file not found, keeping container alive..."
+                # Keep container running without tailing logs
+                while true; do
+                  echo "SQL Gateway is running on port {{ .Values.sqlGateway.port }}"
+                  sleep 60
+                done
+              fi
+          env:
+            - name: FLINK_CONF_DIR
+              value: "/opt/flink/conf"
+            {{- with .Values.sqlGateway.env }}
+            {{- toYaml . | nindent 12 }}
+            {{- end }}
+          ports:
+            - name: rest
+              containerPort: {{ .Values.sqlGateway.port }}
+              protocol: TCP
+          volumeMounts:
+            {{- with .Values.sqlGateway.volumeMounts }}
+            {{- toYaml . | nindent 12 }}
+            {{- end }}
+            - name: sql-gateway-config
+              mountPath: /opt/flink/conf/sql-gateway-defaults.yaml
+              subPath: sql-gateway-defaults.yaml
+              readOnly: true
+          resources:
+            {{- toYaml .Values.sqlGateway.resources | nindent 12 }}
+          {{- with .Values.sqlGateway.livenessProbe }}
+          livenessProbe:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          {{- with .Values.sqlGateway.readinessProbe }}
+          readinessProbe:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+      volumes:
+        {{- with .Values.sqlGateway.volumes }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+        - name: sql-gateway-config
+          configMap:
+            name: sql-gateway-config
+{{- end }}
@@ -0,0 +1,96 @@
+{{- if .Values.sqlGateway.enabled }}
+---
+# Hadoop Configuration ConfigMap for SQL Gateway
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: hadoop-config-{{ .Values.sessionCluster.name }}
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.sqlGateway.labels" . | nindent 4 }}
+data:
+  core-site.xml: |
+    <?xml version="1.0" encoding="UTF-8"?>
+    <configuration>
+      <property>
+        <name>fs.defaultFS</name>
+        <value>hdfs://localhost:9000</value>
+      </property>
+      <property>
+        <name>hadoop.tmp.dir</name>
+        <value>/tmp/hadoop</value>
+      </property>
+      
+      <!-- S3A Configuration -->
+      <property>
+        <name>fs.s3a.impl</name>
+        <value>org.apache.hadoop.fs.s3a.S3AFileSystem</value>
+      </property>
+      <property>
+        <name>fs.s3a.aws.credentials.provider</name>
+        <value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value>
+      </property>
+      <property>
+        <name>fs.s3a.access.key</name>
+        <value>${AWS_ACCESS_KEY_ID}</value>
+      </property>
+      <property>
+        <name>fs.s3a.secret.key</name>
+        <value>${AWS_SECRET_ACCESS_KEY}</value>
+      </property>
+      <property>
+        <name>fs.s3a.endpoint</name>
+        <value>${S3_ENDPOINT}</value>
+      </property>
+      <property>
+        <name>fs.s3a.path.style.access</name>
+        <value>true</value>
+      </property>
+      <property>
+        <name>fs.s3a.connection.ssl.enabled</name>
+        <value>false</value>
+      </property>
+      <property>
+        <name>fs.s3a.fast.upload</name>
+        <value>true</value>
+      </property>
+      <property>
+        <name>fs.s3a.multipart.size</name>
+        <value>67108864</value>
+      </property>
+      <property>
+        <name>fs.s3a.multipart.threshold</name>
+        <value>134217728</value>
+      </property>
+    </configuration>
+  
+  hdfs-site.xml: |
+    <?xml version="1.0" encoding="UTF-8"?>
+    <configuration>
+      <property>
+        <name>dfs.replication</name>
+        <value>1</value>
+      </property>
+      <property>
+        <name>dfs.namenode.name.dir</name>
+        <value>/tmp/hadoop/namenode</value>
+      </property>
+      <property>
+        <name>dfs.datanode.data.dir</name>
+        <value>/tmp/hadoop/datanode</value>
+      </property>
+    </configuration>
+  
+  yarn-site.xml: |
+    <?xml version="1.0" encoding="UTF-8"?>
+    <configuration>
+      <property>
+        <name>yarn.nodemanager.aux-services</name>
+        <value>mapreduce_shuffle</value>
+      </property>
+      <property>
+        <name>yarn.resourcemanager.hostname</name>
+        <value>localhost</value>
+      </property>
+    </configuration>
+{{- end }}
@@ -0,0 +1,49 @@
+{{- if .Values.rbac.create }}
+---
+# ServiceAccount for Flink SQL Gateway
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: {{ include "flink-sql-gateway.serviceAccountName" . }}
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.labels" . | nindent 4 }}
+  {{- with .Values.global.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+---
+# ClusterRole for Flink SQL Gateway
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: {{ include "flink-sql-gateway.fullname" . }}
+  labels:
+    {{- include "flink-sql-gateway.labels" . | nindent 4 }}
+rules:
+  - apiGroups: [""]
+    resources: ["pods", "services", "endpoints", "persistentvolumeclaims", "events", "configmaps", "secrets"]
+    verbs: ["*"]
+  - apiGroups: ["apps"]
+    resources: ["deployments", "replicasets"]
+    verbs: ["*"]
+  - apiGroups: ["flink.apache.org"]
+    resources: ["flinkdeployments", "flinksessionjobs"]
+    verbs: ["*"]
+---
+# ClusterRoleBinding for Flink SQL Gateway
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: {{ include "flink-sql-gateway.fullname" . }}
+  labels:
+    {{- include "flink-sql-gateway.labels" . | nindent 4 }}
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: {{ include "flink-sql-gateway.fullname" . }}
+subjects:
+  - kind: ServiceAccount
+    name: {{ include "flink-sql-gateway.serviceAccountName" . }}
+    namespace: {{ include "flink-sql-gateway.namespace" . }}
+{{- end }}
@@ -0,0 +1,71 @@
+{{- if .Values.sqlGateway.enabled }}
+---
+# Service for SQL Gateway
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ .Values.sqlGateway.name }}
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.sqlGateway.labels" . | nindent 4 }}
+  {{- with .Values.global.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  type: {{ .Values.services.sqlGateway.type }}
+  ports:
+    - name: rest
+      port: {{ .Values.services.sqlGateway.port }}
+      targetPort: {{ .Values.services.sqlGateway.targetPort }}
+      protocol: TCP
+  selector:
+    app: {{ .Values.sqlGateway.name }}
+{{- end }}
+
+{{- if .Values.ingress.enabled }}
+---
+# Ingress for SQL Gateway
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: {{ include "flink-sql-gateway.fullname" . }}
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.labels" . | nindent 4 }}
+  {{- with .Values.ingress.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  {{- if .Values.ingress.className }}
+  ingressClassName: {{ .Values.ingress.className }}
+  {{- end }}
+  {{- if .Values.ingress.tls }}
+  tls:
+    {{- range .Values.ingress.tls }}
+    - hosts:
+        {{- range .hosts }}
+        - {{ . | quote }}
+        {{- end }}
+      secretName: {{ .secretName }}
+    {{- end }}
+  {{- end }}
+  rules:
+    {{- range .Values.ingress.hosts }}
+    - host: {{ .host | quote }}
+      http:
+        paths:
+          {{- range .paths }}
+          - path: {{ .path }}
+            {{- if .pathType }}
+            pathType: {{ .pathType }}
+            {{- end }}
+            backend:
+              service:
+                name: {{ $.Values.sqlGateway.name }}
+                port:
+                  number: {{ $.Values.services.sqlGateway.port }}
+          {{- end }}
+    {{- end }}
+{{- end }}
@@ -0,0 +1,92 @@
+{{- if .Values.sqlClient.enabled }}
+---
+# SQL Client Deployment for interactive SQL queries
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .Values.sqlClient.name }}
+  namespace: {{ include "flink-sql-gateway.namespace" . }}
+  labels:
+    {{- include "flink-sql-gateway.sqlClient.labels" . | nindent 4 }}
+  {{- with .Values.global.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: {{ .Values.sqlClient.name }}
+  template:
+    metadata:
+      labels:
+        app: {{ .Values.sqlClient.name }}
+        {{- include "flink-sql-gateway.selectorLabels" . | nindent 8 }}
+    spec:
+      serviceAccountName: {{ include "flink-sql-gateway.serviceAccountName" . }}
+      {{- with .Values.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.podSecurityContext }}
+      securityContext:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      containers:
+        - name: sql-client
+          image: {{ include "flink-sql-gateway.image" . }}
+          imagePullPolicy: {{ .Values.global.image.pullPolicy }}
+          {{- with .Values.securityContext }}
+          securityContext:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          command: ["/bin/bash"]
+          args:
+            - -c
+            - |
+              echo "Flink SQL Client Ready"
+              echo "Connect to SQL Gateway: {{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}"
+              echo "Usage: /opt/flink/bin/sql-client.sh gateway --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}"
+              
+              # Wait for SQL Gateway to be ready
+              echo "Waiting for SQL Gateway..."
+              until curl -f http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}/v1/info; do
+                echo "Waiting for SQL Gateway..."
+                sleep 10
+              done
+              
+              echo "SQL Gateway is ready!"
+              echo "You can now connect using:"
+              echo "/opt/flink/bin/sql-client.sh gateway --endpoint http://{{ .Values.sqlGateway.name }}:{{ .Values.sqlGateway.port }}"
+              
+              # Keep container running
+              tail -f /dev/null
+          {{- with .Values.sqlClient.env }}
+          env:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          volumeMounts:
+            {{- with .Values.sqlClient.volumeMounts }}
+            {{- toYaml . | nindent 12 }}
+            {{- end }}
+            - name: sql-scripts
+              mountPath: /opt/sql-scripts
+              readOnly: true
+          resources:
+            {{- toYaml .Values.sqlClient.resources | nindent 12 }}
+      volumes:
+        {{- with .Values.sqlClient.volumes }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+        - name: sql-scripts
+          configMap:
+            name: sql-test-scripts
+{{- end }}
@@ -0,0 +1,219 @@
+# Default values for flink-sql-gateway.
+# This is a YAML-formatted file.
+# Declare variables to be passed into your templates.
+
+# Global settings
+global:
+  namespace: flink-sql-test
+  image:
+    repository: paasup/flink-sql-gateway
+    tag: 1.20-kafka
+    pullPolicy: IfNotPresent
+  labels:
+    app: flink-sql-gateway
+    version: v1.20.0
+  annotations: {}
+
+# Flink Session Cluster configuration
+sessionCluster:
+  enabled: true
+  name: flink-session-cluster
+  flinkVersion: v1_20
+  jobManager:
+    replicas: 1
+    resources:
+      memory: 1024m
+      cpu: 0.5
+  taskManager:
+    replicas: 1
+    resources:
+      memory: 2048m
+      cpu: 1
+  flinkConfiguration:
+    taskmanager.numberOfTaskSlots: "2"
+    parallelism.default: "1"
+    execution.checkpointing.storage.fs.path: file:///tmp/flink-checkpoints
+    execution.savepoint.path: file:///tmp/flink-savepoints
+    execution.checkpointing.interval: 60s
+    table.exec.source.idle-timeout: 30s
+    table.exec.resource.default-parallelism: "1"
+  # Optional environment variables for the session cluster
+  env: []
+  # Example:
+  # env:
+  #   - name: KAFKA_BOOTSTRAP_SERVERS
+  #     value: "kafka-cluster:9092"
+  
+  # Optional volume mounts for the session cluster
+  volumeMounts: []
+  # Example:
+  # volumeMounts:
+  #   - name: truststore-certs
+  #     mountPath: /opt/flink/certs
+  #     readOnly: true
+  
+  # Optional volumes for the session cluster
+  volumes: []
+  # Example:
+  # volumes:
+  #   - name: truststore-certs
+  #     secret:
+  #       secretName: truststore-secret
+
+# SQL Gateway configuration
+sqlGateway:
+  enabled: true
+  name: flink-sql-gateway
+  replicas: 1
+  port: 8083
+  resources:
+    requests:
+      memory: 512Mi
+      cpu: 0.25
+    limits:
+      memory: 1Gi
+      cpu: 0.5
+  livenessProbe:
+    httpGet:
+      path: /v1/info
+      port: 8083
+    initialDelaySeconds: 60
+    periodSeconds: 30
+  readinessProbe:
+    httpGet:
+      path: /v1/info
+      port: 8083
+    initialDelaySeconds: 30
+    periodSeconds: 10
+  config:
+    endpoint:
+      rest:
+        address: 0.0.0.0
+        port: 8083
+        bindAddress: 0.0.0.0
+    session:
+      maxNum: 1000
+      idleTimeout: 600000  # 10 minutes
+      checkInterval: 60000  # 1 minute
+      planCache:
+        enabled: true
+        maxSize: 100
+        ttl: 3600000  # 1 hour
+
+  # Optional Flink configuration for SQL Gateway
+  flinkConfiguration: {}
+  # Example:
+  # flinkConfiguration:
+  #   env.java.opts: "-Djavax.net.ssl.trustStore=/opt/flink/certs/ca.p12 -Djavax.net.ssl.trustStoreType=PKCS12 -Djavax.net.ssl.trustStorePassword=${TRUSTSTORE_PASSWORD}"
+        
+  # Optional environment variables for the SQL Gateway
+  env: []
+  # Example:
+  # env:
+  #   - name: KAFKA_BOOTSTRAP_SERVERS
+  #     value: "kafka-cluster:9092"
+  
+  # Optional volume mounts for the SQL Gateway
+  volumeMounts: []
+  # Example:
+  # volumeMounts:
+  #   - name: truststore-certs
+  #     mountPath: /opt/flink/certs
+  #     readOnly: true
+  
+  # Optional volumes for the SQL Gateway
+  volumes: []
+  # Example:
+  # volumes:
+  #   - name: truststore-certs
+  #     secret:
+  #       secretName: truststore-secret
+
+# SQL Client configuration (optional - for interactive SQL queries)
+sqlClient:
+  enabled: false  # Set to true if you need interactive SQL client
+  name: flink-sql-client
+  resources:
+    requests:
+      memory: 512Mi
+      cpu: 0.25
+    limits:
+      memory: 1Gi
+      cpu: 0.5
+  # Optional environment variables for the SQL Client
+  env: []
+  # Example:
+  # env:
+  #   - name: KAFKA_BOOTSTRAP_SERVERS
+  #     value: "kafka-cluster:9092"
+  
+  # Optional volume mounts for the SQL Client
+  volumeMounts: []
+  # Example:
+  # volumeMounts:
+  #   - name: truststore-certs
+  #     mountPath: /opt/flink/certs
+  #     readOnly: true
+  
+  # Optional volumes for the SQL Client
+  volumes: []
+  # Example:
+  # volumes:
+  #   - name: truststore-certs
+  #     secret:
+  #       secretName: truststore-secret
+
+# Service configuration
+services:
+  sqlGateway:
+    type: ClusterIP
+    port: 8083
+    targetPort: 8083
+  sessionCluster:
+    type: ClusterIP
+    port: 8081
+    targetPort: 8081
+
+# RBAC configuration
+rbac:
+  create: true
+  serviceAccountName: flink-sql-gateway
+
+# Security configuration
+security: {}
+# Example security configurations can be added here if needed
+# security:
+#   runAsUser: 1000
+#   runAsGroup: 1000
+#   fsGroup: 1000
+
+# Monitoring configuration
+monitoring:
+  enabled: false
+  prometheus:
+    enabled: false
+  grafana:
+    enabled: false
+
+# Ingress configuration
+ingress:
+  enabled: false
+  className: ""
+  annotations: {}
+  hosts:
+    - host: flink-sql-gateway.local
+      paths:
+        - path: /
+          pathType: Prefix
+  tls: []
+
+# Node selector and tolerations
+nodeSelector: {}
+tolerations: []
+affinity: {}
+
+# Pod security context
+podSecurityContext: {}
+
+# Container security context
+securityContext: {}