feat: support multiple operator in single namespace

Dockerfile

@@ -1,4 +1,4 @@
-FROM public.ecr.aws/docker/library/golang:1.23.4-bookworm AS build
+FROM public.ecr.aws/docker/library/golang:1.24.1-bookworm AS build
 
 ARG upx_version=4.2.4
 

Dockerfile.flink

@@ -1,32 +1,74 @@
-FROM public.ecr.aws/docker/library/flink:1.20.1-scala_2.12-java17
+ARG FLINK_VERSION=2.3.0
+ARG SCALA_VERSION=2.12
+ARG JAVA_VERSION=21
+
+FROM flink:${FLINK_VERSION}-scala_${SCALA_VERSION}-java${JAVA_VERSION}
 
 # Set working directory
 WORKDIR /opt/flink
 
 # Set environment variables for Flink mini-cluster
-ENV FLINK_HOME /opt/flink
+ENV FLINK_HOME=/opt/flink
 ENV PATH=$FLINK_HOME/bin:$PATH
 
 # Expose necessary ports for the Flink UI (JobManager) and job manager
 EXPOSE 8081 6123
 
 COPY ./start-cluster.sh /opt/flink/bin/start-cluster.sh
-RUN chmod +x /opt/flink/bin/start-cluster.sh
 
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka/3.4.0-1.20/flink-connector-kafka-3.4.0-1.20.jar -P /opt/flink/lib/
+# ---- Flink-version-dependent connectors ----
+
+# Kafka connector
+ARG KAFKA_CONNECTOR_VERSION=5.0.0-2.2
+RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-connector-kafka/${KAFKA_CONNECTOR_VERSION}/flink-connector-kafka-${KAFKA_CONNECTOR_VERSION}.jar -P /opt/flink/lib/
+
+# Kafka clients (version-independent)
 RUN wget -q https://repo1.maven.org/maven2/org/apache/kafka/kafka-clients/3.9.0/kafka-clients-3.9.0.jar -P /opt/flink/lib/
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-avro/1.20.1/flink-avro-1.20.1.jar -P /opt/flink/lib/
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-avro-confluent-registry/1.20.1/flink-avro-confluent-registry-1.20.1.jar -P /opt/flink/lib/
+
+# Avro format
+ARG AVRO_VERSION=2.2.1
+RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-avro/${AVRO_VERSION}/flink-avro-${AVRO_VERSION}.jar -P /opt/flink/lib/
+
+# Avro Confluent Registry
+ARG AVRO_REGISTRY_VERSION=2.2.1
+RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-avro-confluent-registry/${AVRO_REGISTRY_VERSION}/flink-avro-confluent-registry-${AVRO_REGISTRY_VERSION}.jar -P /opt/flink/lib/
+
+# ---- Version-independent connectors ----
+
+# ClickHouse SQL connector
 RUN wget -q https://repo1.maven.org/maven2/name/nkonev/flink/flink-sql-connector-clickhouse/1.17.1-8/flink-sql-connector-clickhouse-1.17.1-8.jar -P /opt/flink/lib/
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-postgres-cdc/3.2.1/flink-sql-connector-postgres-cdc-3.2.1.jar -P /opt/flink/lib/
+
+# PostgreSQL CDC
+ARG PG_CDC_VERSION=3.6.0-2.2
+RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-postgres-cdc/${PG_CDC_VERSION}/flink-sql-connector-postgres-cdc-${PG_CDC_VERSION}.jar -P /opt/flink/lib/
+
+# Legacy Avro
 RUN wget -q https://repo1.maven.org/maven2/org/apache/avro/avro/1.8.2/avro-1.8.2.jar -P /opt/flink/lib/
+
+# misc libs
 RUN wget -q https://repo1.maven.org/maven2/net/objecthunter/exp4j/0.4.5/exp4j-0.4.5.jar -P /opt/flink/lib/
 RUN wget -q https://jdbc.postgresql.org/download/postgresql-42.7.4.jar -P /opt/flink/lib/
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-sql-jdbc-driver/1.20.1/flink-sql-jdbc-driver-1.20.1.jar -P /opt/flink/lib/
+
+# JDBC driver
+ARG JDBC_DRIVER_VERSION=2.2.1
+RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-sql-jdbc-driver/${JDBC_DRIVER_VERSION}/flink-sql-jdbc-driver-${JDBC_DRIVER_VERSION}.jar -P /opt/flink/lib/
+
+# Legacy Scala JDBC connector (no 2.x version exists)
 RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-jdbc_2.12/1.10.3/flink-jdbc_2.12-1.10.3.jar -P /opt/flink/lib/
+
+# JDBC connector (latest: 3.3.0-1.20, no 2.x version yet)
+ARG JDBC_CONNECTOR_VERSION=3.3.0-1.20
+RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-connector-jdbc/${JDBC_CONNECTOR_VERSION}/flink-connector-jdbc-${JDBC_CONNECTOR_VERSION}.jar -P /opt/flink/lib/
+
+# misc libs
 RUN wget -q https://repo1.maven.org/maven2/com/aventrix/jnanoid/jnanoid/2.0.0/jnanoid-2.0.0.jar -P /opt/flink/lib/
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-connector-jdbc/3.2.0-1.19/flink-connector-jdbc-3.2.0-1.19.jar -P /opt/flink/lib/ 
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-s3-fs-presto/1.20.1/flink-s3-fs-presto-1.20.1.jar -P /opt/flink/lib/ 
+
+# S3 filesystem
+ARG S3_FS_VERSION=2.2.1
+RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-s3-fs-presto/${S3_FS_VERSION}/flink-s3-fs-presto-${S3_FS_VERSION}.jar -P /opt/flink/lib/
+
+# Compression
+RUN wget -q https://repo1.maven.org/maven2/com/github/luben/zstd-jni/1.5.7-2/zstd-jni-1.5.7-2.jar -P /opt/flink/lib/
 
 # Command to start Flink JobManager and TaskManager in a mini-cluster setup
 CMD ["bin/start-cluster.sh"]

crds.yaml

@@ -2,9 +2,9 @@
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
-  name: flink-jobs.flink.logicamp.tech
+  name: flink-jobs.flink.logicamp.dev
 spec:
-  group: flink.logicamp.tech
+  group: flink.logicamp.dev
   names:
     kind: FlinkJob
     plural: flink-jobs
@@ -30,6 +30,9 @@ spec:
                   type: string
                 name:
                   type: string
+                flinkCluster:
+                  type: string
+                  description: "Identifier of the Flink cluster that manages this job. Defaults to the OPERATOR_ID (usually the Helm release name) if not set."
                 entryClass:
                   type: string
                 parallelism:

example-job.yaml

@@ -1,5 +1,5 @@
 # flink-job-instance.yaml
-apiVersion: flink.logicamp.tech/v1alpha1
+apiVersion: flink.logicamp.dev/v1alpha1
 kind: FlinkJob
 metadata:
   name: my-flink-job

go.mod

@@ -5,7 +5,7 @@ go 1.23.2
 require (
 	github.com/danielgtaylor/huma/v2 v2.27.0
 	github.com/gofiber/fiber/v2 v2.52.6
-	github.com/logi-camp/go-flink-client v0.2.0
+	github.com/logi-camp/go-flink-client v0.2.1
 	github.com/samber/lo v1.47.0
 	go.uber.org/zap v1.27.0
 	k8s.io/apimachinery v0.31.3

go.sum

@@ -62,8 +62,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/logi-camp/go-flink-client v0.2.0 h1:PIyfJq7FjW28bnvemReCicIuQD7JzVgJDk2xPTZUS2s=
-github.com/logi-camp/go-flink-client v0.2.0/go.mod h1:A79abedX6wGQI0FoICdZI7SRoGHj15QwMwWowgsKYFI=
+github.com/logi-camp/go-flink-client v0.2.1 h1:STfKamFm9+2SxxfZO3ysdFsb5MViQdThB4UHbnkUOE8=
+github.com/logi-camp/go-flink-client v0.2.1/go.mod h1:A79abedX6wGQI0FoICdZI7SRoGHj15QwMwWowgsKYFI=
 github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
 github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=

helm/chart/Chart.yaml

@@ -2,9 +2,5 @@ apiVersion: v2
 name: flink-kube-operator
 description: Helm chart for flink kube operator
 type: application
-version: 0.1.14
-appVersion: "0.1.0"
-dependencies:
-  - name: minio
-    repository: https://charts.bitnami.com/bitnami
-    version: 16.0.2 
+version: 1.3.0
+appVersion: "0.1.1"

helm/chart/templates/flink/checkpoint-pvc.yaml

@@ -0,0 +1,12 @@
+{{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {{ .Release.Name }}-flink-checkpoint-pvc
+spec:
+  accessModes:
+    - ReadWriteMany
+  resources:
+    requests:
+      storage: {{ .Values.flink.state.checkpoint.size }}
+{{- end }}

helm/chart/templates/flink/config.yaml

@@ -11,21 +11,41 @@
     taskmanager.data.port: 6125
     taskmanager.numberOfTaskSlots: {{ .Values.flink.taskManager.numberOfTaskSlots }}
     parallelism.default: {{ .Values.flink.parallelism.default }}
+    {{- if semverCompare ">=2.0.0" .Values.flink.version }}
+    state.backend.type: {{ .Values.flink.state.backend }}
+    {{- else }}
     state.backend: {{ .Values.flink.state.backend }}
+    {{- end }}
     rest.port: 8081
     rootLogger.level = DEBUG
     rootLogger.appenderRef.console.ref = ConsoleAppender
     high-availability.type: kubernetes
     kubernetes.namespace: {{ .Release.Namespace }}
     kubernetes.cluster-id: {{ .Values.clusterId | default (print .Release.Name "-cluster") }}
-    execution.checkpointing.interval: {{ .Values.flink.checkpoint.interval }}
-    execution.checkpointing.mode: {{ .Values.flink.checkpoint.mode }}
-    state.checkpoints.dir: s3://flink/checkpoints/
+    execution.checkpointing.interval: {{ .Values.flink.state.checkpoint.interval }}
+    execution.checkpointing.mode: {{ .Values.flink.state.checkpoint.mode }}
+    {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+    state.checkpoints.dir: file:///opt/flink/checkpoints/
+    {{- else if eq .Values.flink.state.checkpoint.storageType "s3" }}
+    state.checkpoints.dir: s3://{{ .Release.Name }}-flink/checkpoints/
+    {{- end }}
+    {{- if semverCompare ">=2.0.0" .Values.flink.version }}
+    state.backend.rocksdb.local-dir: /opt/flink/rocksdb
+    {{- else }}
     state.backend.rocksdb.localdir: /opt/flink/rocksdb
+    {{- end }}
     high-availability.storageDir: /opt/flink/ha
-    state.savepoints.dir: s3://flink/savepoints/
+    {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+    state.savepoints.dir: file:///opt/flink/savepoints/
+    {{- else if eq .Values.flink.state.savepoint.storageType "s3" }}
+    state.savepoints.dir: s3://{{ .Release.Name }}-flink/savepoints/
+    {{- end }}
     state.backend.incremental: {{ .Values.flink.state.incremental }}
     rest.profiling.enabled: true
+    {{- if or (eq .Values.flink.state.checkpoint.storageType "s3") (eq .Values.flink.state.savepoint.storageType "s3") }}
     s3.endpoint: http://{{ .Release.Name }}-minio:9000
     s3.path.style.access: true
+    {{- end }}
+    {{- toYaml .Values.flink.properties | default "" | nindent 4 }}
+    
 {{- end }}

helm/chart/templates/flink/ha.pvc.yaml

@@ -1,7 +1,7 @@
 apiVersion: v1
 kind: PersistentVolumeClaim
 metadata:
-  name: {{ .Release.Name }}-{{ .Values.flink.state.ha.pvcName }}
+  name: {{ .Release.Name }}-flink-ha-pvc
 spec:
   accessModes:
     - ReadWriteOnce

helm/chart/templates/flink/job-manager-deploy.yaml

@@ -24,7 +24,8 @@ spec:
       initContainers:
         - name: volume-mount-hack
           image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
-          runAsUser: 0
+          securityContext:
+            runAsUser: 0
           command: ["sh", "-c", "chown -R flink {{ .Values.flink.state.ha.dir }}"]
           volumeMounts:
             - name: flink-ha
@@ -49,27 +50,45 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: status.podIP
+            {{- if or (eq .Values.flink.state.checkpoint.storageType "s3") (eq .Values.flink.state.savepoint.storageType "s3") }}
             - name: S3_ENDPOINT
-              value: "http://minio-service:9000"
+              value: "http://{{ .Release.Name }}-minio:9000"
             - name: AWS_ACCESS_KEY_ID
               valueFrom:
                 secretKeyRef:
-                  name: {{ .Release.Name }}-flink-secrets
-                  key: minio_access_key
+                  name: {{ .Release.Name }}-minio
+                  key: root-user
             - name: AWS_SECRET_ACCESS_KEY
               valueFrom:
                 secretKeyRef:
-                  name: {{ .Release.Name }}-flink-secrets
-                  key: minio_secret_key
+                  name: {{ .Release.Name }}-minio
+                  key: root-password
+            {{- end }}
           volumeMounts:
             - name: flink-ha
               mountPath: {{ .Values.flink.state.ha.dir }}
-
+            {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+            - name: flink-checkpoint
+              mountPath: /opt/flink/checkpoints
+            {{- end }}
+            {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+            - name: flink-savepoint
+              mountPath: /opt/flink/savepoints
+            {{- end }}
       volumes:
         - name: flink-ha
           persistentVolumeClaim:
-            claimName: {{ .Release.Name }}-{{ .Values.flink.state.ha.pvcName }}
-
+            claimName: {{ .Release.Name }}-flink-ha-pvc
+        {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+        - name: flink-checkpoint
+          persistentVolumeClaim:
+            claimName: {{ .Release.Name }}-flink-checkpoint-pvc
+        {{- end }}
+        {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+        - name: flink-savepoint
+          persistentVolumeClaim:
+            claimName: {{ .Release.Name }}-flink-savepoint-pvc
+        {{- end }}
       {{- with .Values.nodeSelector }}
       nodeSelector:
         {{- toYaml . | nindent 8 }}

helm/chart/templates/flink/savepoint-pvc.yaml

@@ -0,0 +1,12 @@
+{{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: {{ .Release.Name }}-flink-savepoint-pvc
+spec:
+  accessModes:
+    - ReadWriteMany
+  resources:
+    requests:
+      storage: {{ .Values.flink.state.savepoint.size }}
+{{- end }}

helm/chart/templates/flink/task-manager-statefulset.yaml

@@ -31,23 +31,45 @@ spec:
           valueFrom:
             fieldRef:
               fieldPath: status.podIP
+        {{- if or (eq .Values.flink.state.checkpoint.storageType "s3") (eq .Values.flink.state.savepoint.storageType "s3") }}
         - name: S3_ENDPOINT
-          value: "http://minio-service:9000"
+          value: "http://{{ .Release.Name }}-minio:9000"
         - name: AWS_ACCESS_KEY_ID
           valueFrom:
             secretKeyRef:
-              name: {{ .Release.Name }}-flink-secrets
-              key: minio_access_key
+              name: {{ .Release.Name }}-minio
+              key: root-user
         - name: AWS_SECRET_ACCESS_KEY
           valueFrom:
             secretKeyRef:
-              name: {{ .Release.Name }}-flink-secrets
-              key: minio_secret_key
+              name: {{ .Release.Name }}-minio
+              key: root-password
+        {{- end }}
         volumeMounts:
         - name: rocksdb-storage
           mountPath: /opt/flink/rocksdb
+        {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+        - name: flink-checkpoint
+          mountPath: /opt/flink/checkpoints
+        {{- end }}
+        {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+        - name: flink-savepoint
+          mountPath: /opt/flink/savepoints
+        {{- end }}
         resources:
           {{- toYaml .Values.flink.taskManager.resources | nindent 10 }}
+      volumes:
+        {{- if eq .Values.flink.state.checkpoint.storageType "filesystem" }}
+        - name: flink-checkpoint
+          persistentVolumeClaim:
+            claimName: {{ .Release.Name }}-flink-checkpoint-pvc
+        {{- end }}
+        {{- if eq .Values.flink.state.savepoint.storageType "filesystem" }}
+        - name: flink-savepoint
+          persistentVolumeClaim:
+            claimName: {{ .Release.Name }}-flink-savepoint-pvc
+        {{- end }}
+      
   volumeClaimTemplates:
   - metadata:
       name: rocksdb-storage

helm/chart/templates/operator/hpa.yaml

@@ -8,8 +8,8 @@ metadata:
 spec:
   scaleTargetRef:
     apiVersion: apps/v1
-    kind: Deployment
-    name: {{ include "flink-kube-operator.fullname" . }}
+    kind: StatefulSet
+    name: {{ .Release.Name }}-flink-operator
   minReplicas: {{ .Values.autoscaling.minReplicas }}
   maxReplicas: {{ .Values.autoscaling.maxReplicas }}
   metrics:

helm/chart/templates/operator/role.yaml

@@ -6,7 +6,7 @@ metadata:
     {{- include "flink-kube-operator.labels" . | nindent 4 }}
 rules:
   - apiGroups:
-      - flink.logicamp.tech  # API group of the FlinkJob CRD
+      - flink.logicamp.dev  # API group of the FlinkJob CRD
     resources:
       - flink-jobs           # The plural name of your custom resource
     verbs:

helm/chart/templates/operator/statefulset.yaml

@@ -44,12 +44,15 @@ spec:
               containerPort: {{ .Values.service.port }}
               protocol: TCP
           env:
+            - name: OPERATOR_ID
+              value: {{ .Values.operatorId | default .Release.Name }}
             - name: FLINK_API_URL
               value: {{ .Release.Name }}-flink-job-manager:8081
-            - name: SAVEPOINT_PATH
-              value: s3://flink/savepoints/
             - name: NAMESPACE
               value: "{{ .Release.Namespace }}"
+            {{- if eq .Values.flink.state.savepoint.storageType "s3" }}
+            - name: SAVEPOINT_PATH
+              value: s3://{{ .Release.Name }}-flink/savepoints/
             - name: S3_ENDPOINT
               value: "http://{{ .Release.Name }}-minio:9000"
             - name: AWS_ACCESS_KEY_ID
@@ -62,5 +65,8 @@ spec:
                 secretKeyRef:
                   name: {{ .Release.Name }}-minio
                   key: root-password
-
+            {{- else }}
+            - name: SAVEPOINT_PATH
+              value: /opt/flink/savepoints/
+            {{- end }}
 

helm/chart/values.yaml

@@ -7,11 +7,11 @@ replicaCount: 1
 
 # This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/
 image:
-  repository: lcr.logicamp.tech/library/flink-kube-operator
+  repository: logicampdev/flink-kube-operator
   # This sets the pull policy for images.
   pullPolicy: Always
   # Overrides the image tag whose default is the chart appVersion.
-  tag: latest
+  tag: v1.3.0
 
 # This is for the secretes for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/
 imagePullSecrets: []
@@ -38,8 +38,7 @@ podAnnotations: {}
 # For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
 podLabels: {}
 
-podSecurityContext: {}
-  # fsGroup: 2000
+podSecurityContext: {} # fsGroup: 2000
 
 securityContext: {}
   # capabilities:
@@ -64,10 +63,10 @@ ingress:
     # kubernetes.io/ingress.class: nginx
     # kubernetes.io/tls-acme: "true"
   hosts:
-    - host: chart-example.local
-      paths:
-        - path: /
-          pathType: ImplementationSpecific
+  - host: chart-example.local
+    paths:
+    - path: /
+      pathType: ImplementationSpecific
   tls: []
   #  - secretName: chart-example-tls
   #    hosts:
@@ -106,6 +105,10 @@ autoscaling:
 config:
   flinkApiUrl: flink:8081
 
+# Identifier for this operator instance. Used to partition FlinkJob CRs via
+# label selector. Defaults to the Helm release name if left empty.
+# Set explicitly when you need stable names across upgrades.
+operatorId: ""
 
 nodeSelector: {}
 
@@ -115,33 +118,43 @@ affinity: {}
 
 # Global values for the Flink deployment
 flink:
+  # Flink major.minor version (semver string). Used by config template to emit
+  # the correct configuration key names for the target Flink version.
+  # Override to "1.20" for backward compatibility with Flink 1.20.x.
+  version: "2.3"
   image:
-    repository: lcr.logicamp.tech/library/flink
-    tag: 1.20.1-scala_2.12-java17-minicluster
+    repository: logicampdev/flink-kube-operator-flink
+    tag: 2.3.0-java21-v1.3.0
 
   parallelism:
-    default: 1  # Default parallelism for Flink jobs
-  
-  checkpoint:
-    interval: 5min
-    mode: EXACTLY_ONCE
+    default: 1 # Default parallelism for Flink jobs
 
   state:
-    backend: rocksdb  # Use RocksDB for state backend
+    backend: rocksdb # Use RocksDB for state backend
     incremental: true
     ha:
-      dir: "/opt/flink/ha"  # Directory to store ha data
+      dir: "/opt/flink/ha" # Directory to store ha data
       pvcName: flink-ha-pvc # PVC for ha
       size: 10Gi # PVC size for ha
+    checkpoint:
+      storageType: s3 # s3 / filesystem
+      interval: 5min
+      mode: EXACTLY_ONCE
+      size: 8Gi
+    savepoint:
+      storageType: s3
+      size: 8Gi
 
   jobManager:
     processMemory: 4096m # Size of job manager process memory
 
+  properties:
+    jobmanager.rpc.timeout: 300s
 
   taskManager:
-    numberOfTaskSlots: 12  # Number of task slots for TaskManager
+    numberOfTaskSlots: 12 # Number of task slots for task manager
     processMemory: 4096m # Size of task manager process memory
-    replicas: 1
+    replicas: 1 # Number of task manager replicas
     storage:
       rocksDb:
         size: 4Gi

helm/index.yaml

@@ -1,6 +1,92 @@
 apiVersion: v1
 entries:
   flink-kube-operator:
+  - apiVersion: v2
+    appVersion: 0.1.1
+    created: "2025-07-18T18:09:46.27166563+03:30"
+    description: Helm chart for flink kube operator
+    digest: 597f2c07884bb5411dcc6e1a9cdf7672977858efe30273a46fb6525eb6013091
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-1.2.3.tgz
+    version: 1.2.3
+  - apiVersion: v2
+    appVersion: 0.1.1
+    created: "2025-05-17T14:34:55.317942453+03:30"
+    description: Helm chart for flink kube operator
+    digest: 422a34dc173ebe29adccd46d7ef94505cc022ff20ccbfb85ac3e6e201cba476c
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-1.2.2.tgz
+    version: 1.2.2
+  - apiVersion: v2
+    appVersion: 0.1.1
+    created: "2025-05-17T14:01:29.891695937+03:30"
+    description: Helm chart for flink kube operator
+    digest: 404ed2c28ff43b630b44c1215be5369417a1b9b2747ae24e2963a6b81813e7dc
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-1.2.1.tgz
+    version: 1.2.1
+  - apiVersion: v2
+    appVersion: 0.1.1
+    created: "2025-05-17T12:47:25.848097207+03:30"
+    dependencies:
+    - name: minio
+      repository: https://charts.bitnami.com/bitnami
+      version: 16.0.2
+    description: Helm chart for flink kube operator
+    digest: 3458b9be97d2a4bcf8574706e44ea9f7fdeb11e83058a615566e6e094a51b920
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-1.2.0.tgz
+    version: 1.2.0
+  - apiVersion: v2
+    appVersion: 0.1.1
+    created: "2025-04-15T12:06:59.425538953+03:30"
+    dependencies:
+    - name: minio
+      repository: https://charts.bitnami.com/bitnami
+      version: 16.0.2
+    description: Helm chart for flink kube operator
+    digest: 2b307a113476eebb34f58308bf1d4d0d36ca5e08fe6541f369a1c231ae0a71be
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-1.1.2.tgz
+    version: 1.1.2
+  - apiVersion: v2
+    appVersion: 0.1.1
+    created: "2025-04-12T23:13:39.394371646+03:30"
+    dependencies:
+    - name: minio
+      repository: https://charts.bitnami.com/bitnami
+      version: 16.0.2
+    description: Helm chart for flink kube operator
+    digest: 14b08b443b4118cee4c279f62b498bc040b4a3e7ebafa8e195606e3d9b21810a
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-1.0.1.tgz
+    version: 1.0.1
+  - apiVersion: v2
+    appVersion: 0.1.1
+    created: "2025-04-06T01:52:09.478716316+03:30"
+    dependencies:
+    - name: minio
+      repository: https://charts.bitnami.com/bitnami
+      version: 16.0.2
+    description: Helm chart for flink kube operator
+    digest: e177bc2f11987f4add27c09e521476eabaa456df1b9621321200b58f3a330813
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-1.0.0.tgz
+    version: 1.0.0
   - apiVersion: v2
     appVersion: 0.1.0
     created: "2025-04-04T13:50:27.971040367+03:30"
@@ -151,4 +237,4 @@ entries:
     urls:
     - flink-kube-operator-0.1.0.tgz
     version: 0.1.0
-generated: "2025-04-04T13:50:27.967565847+03:30"
+generated: "2025-07-18T18:09:46.244672127+03:30"

internal/crd/v1alpha1/flink_job.go

@@ -12,6 +12,7 @@ import (
 type FlinkJobSpec struct {
 	Key                     string          `json:"key"`
 	Name                    string          `json:"name"`
+	FlinkCluster            string          `json:"flinkCluster"`
 	Parallelism             int             `json:"parallelism"`
 	JarURI                  string          `json:"jarUri"`
 	JarURIBasicAuthUsername *string         `json:"jarURIBasicAuthUsername"`

internal/crd/v1alpha1/register.go

@@ -6,7 +6,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
-const GroupName = "flink.logicamp.tech"
+const GroupName = "flink.logicamp.dev"
 const GroupVersion = "v1alpha1"
 const ResourceName = "flink-jobs"
 

internal/managed_job/cycle.go

@@ -13,7 +13,7 @@ func (job *ManagedJob) Cycle() {
 	// pkg.Logger.Debug("[managed-job] [new] check cycle", zap.String("jobName", job.def.GetName()))
 
 	// Init job
-	if job.def.Status.LifeCycleStatus == "" && job.def.Status.JobStatus == "" {
+	if job.def.Status.LifeCycleStatus == "" && (job.def.Status.JobStatus == "" || job.def.Status.JobStatus == v1alpha1.JobStatusFinished) {
 		job.Run(false)
 		return
 	}

internal/managed_job/manager.go

@@ -3,6 +3,7 @@ package managed_job
 import (
 	"flink-kube-operator/internal/crd"
 	"flink-kube-operator/internal/crd/v1alpha1"
+	"os"
 	"time"
 
 	"flink-kube-operator/pkg"
@@ -73,6 +74,8 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 	}
 	pkg.Logger.Debug("[manager] [cycle] overviews", zap.Any("overviews", jobManagerJobOverviews))
 
+	operatorId := os.Getenv("OPERATOR_ID")
+
 	// Loop over job definitions as Kubernetes CRD
 	for _, uid := range crd.GetAllJobKeys() {
 		if lo.Contains(mgr.processingJobsIds, uid) {
@@ -82,6 +85,27 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 
 		// Get job definition from Kubernetes CRD
 		def := crd.GetJob(uid)
+
+		// Auto-claim: if flinkCluster is empty, set it to this operator's ID
+		if def.Spec.FlinkCluster == "" {
+			pkg.Logger.Info("[manager] auto-claiming unowned job", zap.String("name", def.GetName()))
+			crdInstance.Patch(def.GetUID(), map[string]interface{}{
+				"spec": map[string]interface{}{
+					"flinkCluster": operatorId,
+				},
+			})
+			def.Spec.FlinkCluster = operatorId
+		}
+
+		// Skip CRs that belong to another operator
+		if def.Spec.FlinkCluster != operatorId {
+			pkg.Logger.Debug("[manager] skipping job owned by another operator",
+				zap.String("name", def.GetName()),
+				zap.String("flinkCluster", def.Spec.FlinkCluster),
+				zap.String("operatorId", operatorId),
+			)
+			continue
+		}
 		mgr.processingJobsIds = append(mgr.processingJobsIds, uid)
 
 		// Check if job exists in manager managed jobs
@@ -117,14 +141,15 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 				"status": patchStatusObj,
 			})
 		} else {
-			patchStatusObj := map[string]interface{}{
-				"jobStatus":       "",
-				"lifeCycleStatus": string(v1alpha1.LifeCycleStatusFailed),
-			}
+			// TODO handle job not found status
+			// patchStatusObj := map[string]interface{}{
+			// 	"jobStatus":       "",
+			// 	"lifeCycleStatus": string(v1alpha1.LifeCycleStatusFailed),
+			// }
 
-			crdInstance.Patch(uid, map[string]interface{}{
-				"status": patchStatusObj,
-			})
+			// crdInstance.Patch(uid, map[string]interface{}{
+			// 	"status": patchStatusObj,
+			// })
 		}
 
 		managedJob.Cycle()

internal/managed_job/run.go

@@ -43,7 +43,7 @@ func (job *ManagedJob) Run(restoreMode bool) error {
 				EntryClass:            job.def.Spec.EntryClass,
 				SavepointPath:         savepointPath,
 				Parallelism:           job.def.Spec.Parallelism,
-				ProgramArg:            job.def.Spec.Args,
+				ProgramArgsList:       job.def.Spec.Args,
 			})
 			if err == nil {
 				pkg.Logger.Info("[managed-job] [run] jar successfully ran", zap.Any("run-jar-resp", runJarResp))

scripts/build-flink-images.sh

@@ -0,0 +1,58 @@
+#!/bin/bash
+set -euo pipefail
+
+REGISTRY="${REGISTRY:-lcr.logicamp.tech/library/flink}"
+
+echo "=== Building Flink 2.3.0 image (default) ==="
+docker build \
+  --build-arg FLINK_VERSION=2.3.0 \
+  --build-arg JAVA_VERSION=21 \
+  --build-arg KAFKA_CONNECTOR_VERSION=5.0.0-2.2 \
+  --build-arg AVRO_VERSION=2.2.1 \
+  --build-arg AVRO_REGISTRY_VERSION=2.2.1 \
+  --build-arg PG_CDC_VERSION=3.6.0-2.2 \
+  --build-arg JDBC_DRIVER_VERSION=2.2.1 \
+  --build-arg JDBC_CONNECTOR_VERSION=3.3.0-1.20 \
+  --build-arg S3_FS_VERSION=2.2.1 \
+  -f Dockerfile.flink \
+  -t "${REGISTRY}:2.3.0-scala_2.12-java21-minicluster" \
+  .
+
+echo ""
+echo "=== Building Flink 2.2.1 image ==="
+docker build \
+  --build-arg FLINK_VERSION=2.2.1 \
+  --build-arg JAVA_VERSION=17 \
+  --build-arg KAFKA_CONNECTOR_VERSION=5.0.0-2.2 \
+  --build-arg AVRO_VERSION=2.2.1 \
+  --build-arg AVRO_REGISTRY_VERSION=2.2.1 \
+  --build-arg PG_CDC_VERSION=3.6.0-2.2 \
+  --build-arg JDBC_DRIVER_VERSION=2.2.1 \
+  --build-arg JDBC_CONNECTOR_VERSION=3.3.0-1.20 \
+  --build-arg S3_FS_VERSION=2.2.1 \
+  -f Dockerfile.flink \
+  -t "${REGISTRY}:2.2.1-scala_2.12-java17-minicluster" \
+  .
+
+echo ""
+echo "=== Building Flink 1.20.1 image (backward compat) ==="
+docker build \
+  --build-arg FLINK_VERSION=1.20.1 \
+  --build-arg JAVA_VERSION=17 \
+  --build-arg KAFKA_CONNECTOR_VERSION=3.4.0-1.20 \
+  --build-arg AVRO_VERSION=1.20.1 \
+  --build-arg AVRO_REGISTRY_VERSION=1.20.1 \
+  --build-arg PG_CDC_VERSION=3.2.1 \
+  --build-arg JDBC_DRIVER_VERSION=1.20.1 \
+  --build-arg JDBC_CONNECTOR_VERSION=3.2.0-1.19 \
+  --build-arg S3_FS_VERSION=1.20.1 \
+  -f Dockerfile.flink \
+  -t "${REGISTRY}:1.20.1-scala_2.12-java17-minicluster" \
+  .
+
+echo ""
+echo "=== Done ==="
+echo "Images built:"
+echo "  ${REGISTRY}:2.3.0-scala_2.12-java21-minicluster"
+echo "  ${REGISTRY}:2.2.1-scala_2.12-java17-minicluster"
+echo "  ${REGISTRY}:1.20.1-scala_2.12-java17-minicluster"