feat: support multiple operator in single namespace

feat: update flink to 2.3
fix(helm): wrong savepoint path config when storage-type is filesystem
2026-07-23 13:51:33 +03:30 · 2026-07-22 10:48:53 +03:30 · 2025-07-18 18:12:21 +03:30 · 2025-05-17 14:35:22 +03:30 · 2025-05-17 14:02:01 +03:30 · 2025-05-17 13:02:24 +03:30
35 changed files with 396 additions and 81 deletions
@@ -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

@@ -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"]
@@ -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:
@@ -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
@@ -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
@@ -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=
@@ -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"
@@ -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 }}
@@ -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 }}
@@ -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
@@ -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 }}
@@ -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 }}
@@ -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
@@ -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:
@@ -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:
@@ -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 }}

@@ -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,40 +118,50 @@ 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
-    resources: 
+    resources:
      limits:
        cpu: 3
        memory: 4Gi
      requests:
        cpu: 1
-        memory: 2Gi
+        memory: 2Gi
@@ -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"
@@ -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"`
@@ -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"

@@ -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
 	}
@@ -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()
@@ -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))
@@ -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"
Author	SHA1	Message	Date
mohamad	7fb18cd45f	feat: support multiple operator in single namespace	2026-07-23 13:51:33 +03:30
mohamad	494d32c565	feat: update flink to 2.3	2026-07-22 10:48:53 +03:30
mohamad	5ca1c28b33	fix(helm): wrong savepoint path config when storage-type is filesystem	2025-07-18 18:12:21 +03:30
mohamad	d73292ac54	fix: resolve missing task manager statefulset savepoint pvc mount	2025-05-17 14:35:22 +03:30
mohamad	f0df5ff937	fix: wrong fieldPath in task-manager statefulset spec.hostname	2025-05-17 14:02:01 +03:30
mohamad	83c4b5ded2	feat(helm): add filesystem savepoint storage mode	2025-05-17 13:02:24 +03:30
mohamad	89647f3b5b	fix(helm): add flink taskmanager host env to task manager	2025-04-15 12:08:17 +03:30
mohamad	dedbe00fba	fix(helm): wrong checkpoint path flink properties	2025-04-13 10:38:15 +03:30
mohamad	62c340bc64	feat(helm): add filesystem checkpoint storage mode	2025-04-13 10:00:32 +03:30
mohamad	44ff3627fc	feat(helm): add flink properties variable to values	2025-04-12 23:14:52 +03:30
mohamad	392004d99a	ci(docker): add zstd dependency jar to flink docker file	2025-04-12 23:07:53 +03:30
mohamad	22c7d712f4	feat: update flink http client library	2025-04-07 13:20:39 +03:30
mohamad	2dd625ec7c	feat: update flink http client library	2025-04-07 11:28:33 +03:30
mohamad	c991215a9d	Merge branch 'main' of https://git.logicamp.tech/Logicamp/flink-kube-operator	2025-04-06 08:48:54 +03:30
mohamad	1c32bfbbe0	chore: create index and chart package	2025-04-06 01:53:33 +03:30
mohamad	f210090dff	Merge branch 'feature/new-helm-structure' into HEAD	2025-04-06 01:49:21 +03:30
mohamad	556d9ff6af	fix: wong update status in some situations	2025-03-05 11:40:22 +03:30