Add chart flink-sql-gateway

2026-01-23 08:17:33 +09:00
parent 47213d260e
commit bb69f903f4
11 changed files with 1382 additions and 0 deletions
@@ -0,0 +1,22 @@
+apiVersion: v2
+name: flink-sql-gateway
+description: A Helm chart for Flink SQL Gateway with Kafka integration
+type: application
+version: 0.1.0
+appVersion: "2.0.1"
+keywords:
+  - flink
+  - sql
+  - gateway
+  - kafka
+  - streaming
+home: https://github.com/your-org/flink-sql-gateway-helm
+sources:
+  - https://github.com/your-org/flink-sql-gateway-helm
+maintainers:
+  - name: Flink SQL Gateway Team
+    email: team@example.com
+dependencies: []
+annotations:
+  category: Analytics
+  licenses: Apache-2.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,53 @@
+# 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"
+  volumeMounts: 
+    - name: kafka-certs
+      mountPath: /opt/flink/certs
+      readOnly: true
+  volumes:
+    - name: kafka-certs
+      secret:
+        secretName: kafka-ca-cert
+
+# 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
+
+sqlClient:
+   enabled: true
+# 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,159 @@
+{{- 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-conf.yaml with remote cluster configuration
+              cat > /opt/flink/conf/config.yaml << EOF
+              # Remote Flink Cluster Configuration
+              rest.address: {{ .Values.sessionCluster.name }}-rest
+              rest.port: 8081
+              execution.target: remote
+              
+              # 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
+              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,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,83 @@
+{{- if .Values.sqlClient.enabled }}
+---
+# SQL Client Pod for interactive SQL queries
+apiVersion: v1
+kind: Pod
+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:
+  serviceAccountName: {{ include "flink-sql-gateway.serviceAccountName" . }}
+  {{- with .Values.nodeSelector }}
+  nodeSelector:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+  {{- with .Values.affinity }}
+  affinity:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+  {{- with .Values.tolerations }}
+  tolerations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+  {{- with .Values.podSecurityContext }}
+  securityContext:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+  containers:
+    - name: sql-client
+      image: {{ include "flink-sql-gateway.image" . }}
+      imagePullPolicy: {{ .Values.global.image.pullPolicy }}
+      {{- with .Values.securityContext }}
+      securityContext:
+        {{- toYaml . | nindent 8 }}
+      {{- 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 8 }}
+      {{- end }}
+      volumeMounts:
+        {{- with .Values.sqlClient.volumeMounts }}
+        {{- toYaml . | nindent 8 }}
+        {{- end }}
+        - name: sql-scripts
+          mountPath: /opt/sql-scripts
+          readOnly: true
+      resources:
+        {{- toYaml .Values.sqlClient.resources | nindent 8 }}
+  volumes:
+    {{- with .Values.sqlClient.volumes }}
+    {{- toYaml . | nindent 4 }}
+    {{- end }}
+    - name: sql-scripts
+      configMap:
+        name: sql-test-scripts
+  restartPolicy: Always
+{{- end }}
@@ -0,0 +1,212 @@
+# 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: kafka-certs
+  #     mountPath: /opt/flink/certs
+  #     readOnly: true
+  
+  # Optional volumes for the session cluster
+  volumes: []
+  # Example:
+  # volumes:
+  #   - name: kafka-certs
+  #     secret:
+  #       secretName: kafka-cluster-ca-cert
+
+# 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 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: kafka-certs
+  #     mountPath: /opt/flink/certs
+  #     readOnly: true
+  
+  # Optional volumes for the SQL Gateway
+  volumes: []
+  # Example:
+  # volumes:
+  #   - name: kafka-certs
+  #     secret:
+  #       secretName: kafka-cluster-ca-cert
+
+# 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: kafka-certs
+  #     mountPath: /opt/flink/certs
+  #     readOnly: true
+  
+  # Optional volumes for the SQL Client
+  volumes: []
+  # Example:
+  # volumes:
+  #   - name: kafka-certs
+  #     secret:
+  #       secretName: kafka-cluster-ca-cert
+
+# 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: {}