feat(jar): add OCI registry pull for JAR artifacts with backward compat

- Add jarRef and jarPullSecret fields to FlinkJob CRD (jarUri/basicAuth deprecated)
- OCI pull via go-containerregistry with dual auth (K8s pull secret + env vars)
- Media type validation on pulled layers
- Atomic status patches with runningJarRef/runningJarDigest tracking
- NeedsUpgrade/RunningRef/RunningRefPatchData domain helpers
- README with usage guide, pushing JARs, and GitHub Actions CI/CD workflow
- CONTEXT.md domain glossary

.gitignore

@@ -2,4 +2,5 @@ db
 *.log
 tmp
 *.jar
-.env
+.env
+/operator

.vscode/settings.json

@@ -14,7 +14,7 @@
     "nindent",
     "reactivex",
     "repsert",
-    "rxgo",
+    "taskmanager",
     "tolerations"
   ]
 }

CONTEXT.md

@@ -0,0 +1,41 @@
+# Flink Kube Operator
+
+Kubernetes operator that manages the lifecycle of Apache Flink jobs via CRDs.
+
+## Language
+
+**FlinkJob**:
+A Kubernetes Custom Resource representing a desired Flink job. Contains spec (desired state) and status (observed state).
+_Avoid_: Job CR, CRD instance
+
+**ManagedJob**:
+Runtime wrapper that drives a FlinkJob's lifecycle through its state machine (init → run → savepoint → upgrade).
+_Avoid_: Job controller, job runner
+
+**jarRef**:
+Full OCI reference identifying a JAR artifact in a registry (e.g. `registry.io/org/app:1.0.0`).
+_Avoid_: jarUri, jar URL, artifact URL
+
+**jarDigest**:
+SHA256 content hash of the JAR layer in the OCI manifest. Immutable identifier for the artifact bytes.
+_Avoid_: content hash, checksum
+
+**pullSecret**:
+A `kubernetes.io/dockerconfigjson` Kubernetes Secret containing registry credentials, referenced by name in the FlinkJob spec.
+_Avoid_: imagePullSecret, registry secret
+
+**JarFile**:
+Ephemeral artifact handle — a downloaded/pulled JAR on local disk awaiting upload to Flink, then deleted.
+_Avoid_: temp jar, local jar
+
+**LifeCycleStatus**:
+Operator-managed lifecycle state of a FlinkJob (HEALTHY, RESTORING, GRACEFULLY_PAUSED, FAILED, etc.).
+_Avoid_: operator status, lifecycle state
+
+**JobStatus**:
+Flink-reported job state (RUNNING, FAILED, FINISHED, etc.). Mirrored from the Flink REST API.
+_Avoid_: flink status, job state
+
+**Savepoint**:
+A consistent snapshot of a Flink job's state, used for restart and upgrade.
+_Avoid_: checkpoint, snapshot

Dockerfile

@@ -1,8 +1,8 @@
-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
 
-RUN apt-get update && apt-get install -y --no-install-recommends xz-utils && \
+RUN apt-get update && apt-get install -y --no-install-recommends xz-utils ca-certificates && \
   curl -Ls https://github.com/upx/upx/releases/download/v${upx_version}/upx-${upx_version}-amd64_linux.tar.xz -o - | tar xvJf - -C /tmp && \
   cp /tmp/upx-${upx_version}-amd64_linux/upx /usr/local/bin/ && \
   chmod +x /usr/local/bin/upx && \
@@ -27,6 +27,7 @@ FROM public.ecr.aws/docker/library/busybox:1.37.0 AS final
 
 
 COPY --from=build /flink-kube-operator /flink-kube-operator
+COPY --from=build /etc/ssl/certs /etc/ssl/certs
 
 EXPOSE 8083
 

Dockerfile.flink

@@ -1,30 +1,74 @@
-FROM public.ecr.aws/docker/library/flink:1.20.0-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.0/flink-avro-1.20.0.jar -P /opt/flink/lib/
-RUN wget -q https://repo1.maven.org/maven2/org/apache/flink/flink-avro-confluent-registry/1.20.0/flink-avro-confluent-registry-1.20.0.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.0/flink-sql-jdbc-driver-1.20.0.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/
+
+# 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"]

README.md

@@ -1,6 +1,203 @@
-Installation:
+# Flink Kube Operator
+
+A Kubernetes operator for managing Apache Flink jobs via Custom Resource Definitions.
+
+## Installation
 
 ```bash
 helm repo add lc-flink-operator https://git.logicamp.tech/Logicamp/flink-kube-operator/raw/branch/main/helm/
 helm install flink-kube-operator lc-flink-operator/flink-kube-operator
-```
+```
+
+## FlinkJob CRD
+
+Define Flink jobs as Kubernetes resources:
+
+```yaml
+apiVersion: flink.logicamp.dev/v1alpha1
+kind: FlinkJob
+metadata:
+  name: my-flink-job
+spec:
+  key: word-count
+  name: "Word Count Example"
+  entryClass: "com.example.WordCount"
+  parallelism: 1
+  jarRef: "registry.example.com/myorg/word-count:1.0.0"
+  jarPullSecret: "registry-creds"
+  args:
+    kafka-host: kafka:9092
+    kafka-group-id: my-group
+```
+
+### Spec Fields
+
+| Field | Type | Required | Description |
+|-------|------|----------|-------------|
+| `key` | string | yes | Unique identifier for the job |
+| `name` | string | yes | Human-readable job name |
+| `entryClass` | string | yes | Flink job entry class |
+| `parallelism` | int | yes | Job parallelism |
+| `jarRef` | string | yes | OCI reference to the JAR artifact (e.g. `registry.io/org/app:tag`) |
+| `jarPullSecret` | string | no | Name of a `kubernetes.io/dockerconfigjson` Secret for registry auth |
+| `args` | map/list | no | Program arguments (map or list format) |
+| `savepointInterval` | duration | no | Interval between automatic savepoints |
+
+### Deprecated Fields (backward compat)
+
+| Field | Replaced by |
+|-------|-------------|
+| `jarUri` | `jarRef` |
+| `jarURIBasicAuthUsername` | `jarPullSecret` |
+| `jarURIBasicAuthPassword` | `jarPullSecret` |
+
+If both `jarRef` and `jarUri` are set, `jarRef` takes precedence.
+
+## Registry Authentication
+
+### Kubernetes (pull secret)
+
+Create a `kubernetes.io/dockerconfigjson` Secret and reference it via `jarPullSecret`:
+
+```bash
+kubectl create secret docker-registry registry-creds \
+  --docker-server=registry.example.com \
+  --docker-username=user \
+  --docker-password=pass
+```
+
+```yaml
+spec:
+  jarRef: "registry.example.com/myorg/my-job:1.0.0"
+  jarPullSecret: "registry-creds"
+```
+
+### Docker Compose / local dev (environment variables)
+
+Set registry credentials as environment variables on the operator container:
+
+```yaml
+# docker-compose.yaml
+services:
+  operator:
+    image: flink-kube-operator:latest
+    environment:
+      REGISTRY_SERVER: "registry.example.com"
+      REGISTRY_USERNAME: "user"
+      REGISTRY_PASSWORD: "pass"
+```
+
+### Credential Resolution Order
+
+1. `spec.jarPullSecret` — reads the K8s Secret
+2. `REGISTRY_USERNAME` + `REGISTRY_PASSWORD` env vars — fallback (optionally scoped by `REGISTRY_SERVER`)
+3. Anonymous — public registries
+
+## Pushing JARs to an OCI Registry
+
+Flink job JARs are stored as OCI artifacts with media type `application/java-archive`. Use one of the methods below to push your JAR.
+
+### Using crane (recommended)
+
+[crane](https://github.com/google/go-containerregistry/tree/main/cmd/crane) is a tool for interacting with OCI registries.
+
+```bash
+# Install crane
+go install github.com/google/go-containerregistry/cmd/crane@latest
+
+# Log in to your registry
+crane auth login registry.example.com -u user -p pass
+
+# Push a JAR as an OCI artifact
+crane append \
+  --file=my-job.jar \
+  --new_tag=registry.example.com/myorg/my-job:1.0.0 \
+  --base=scratch
+```
+
+### Using oras
+
+[oras](https://oras.land/) is a tool for pushing OCI artifacts.
+
+```bash
+# Install oras
+# https://oras.land/docs/installation
+
+# Log in to your registry
+oras login registry.example.com -u user -p pass
+
+# Push a JAR with the correct media type
+oras push registry.example.com/myorg/my-job:1.0.0 \
+  --artifact-type application/java-archive \
+  my-job.jar:application/java-archive
+```
+
+### Using Docker (buildkit)
+
+```bash
+# Create a minimal Dockerfile
+cat > Dockerfile.jar <<'EOF'
+FROM scratch
+COPY my-job.jar /my-job.jar
+EOF
+
+# Build and push
+docker build -f Dockerfile.jar -t registry.example.com/myorg/my-job:1.0.0 .
+docker push registry.example.com/myorg/my-job:1.0.0
+```
+
+### GitHub Actions (CI/CD)
+
+Use [crane](https://github.com/google/go-containerregistry/tree/main/cmd/crane) in your GitHub Actions workflow to push JARs as OCI artifacts. `docker/build-push-action` is not suitable — it builds container images, not arbitrary OCI artifacts.
+
+```yaml
+name: Publish JAR
+
+on:
+  push:
+    tags:
+      - 'v*'
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME: ${{ github.repository }}/my-job
+
+jobs:
+  publish:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Log in to registry
+        uses: docker/login-action@v3
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Set up crane
+        uses: imjasonh/setup-crane@v0.4
+
+      - name: Build JAR
+        run: mvn package -DskipTests
+
+      - name: Push JAR to OCI registry
+        run: |
+          crane append \
+            --base=scratch \
+            --new_tag=${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.ref_name }} \
+            --new_layer=target/my-job.jar \
+            --media_type=application/java-archive
+```
+
+For GitLab CI, Bitbucket Pipelines, or other CI systems, use the same `crane append` command after authenticating to your registry.
+
+### Versioning Strategy
+
+- Use semantic version tags (`:1.0.0`, `:1.1.0`) for explicit version control
+- Changing the tag in `spec.jarRef` triggers a job upgrade (pause → re-upload → resume from savepoint)
+- The operator stores the SHA256 digest in `status.runningJarDigest` after each pull

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
@@ -24,22 +24,36 @@ spec:
               type: object
               required:
                 - key
-                - jarUri
               properties:
                 key:
                   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:
                   type: integer
+                jarRef:
+                  type: string
+                  description: "OCI reference to the JAR artifact (e.g. registry.io/org/app:tag). Takes precedence over jarUri."
+                jarPullSecret:
+                  type: string
+                  description: "Name of a kubernetes.io/dockerconfigjson Secret for registry authentication."
                 jarUri:
                   type: string
+                  description: "Deprecated. Use jarRef instead. HTTP(S) URL to the JAR file."
+                jarURIBasicAuthUsername:
+                  type: string
+                  description: "Deprecated. Use jarPullSecret instead."
+                jarURIBasicAuthPassword:
+                  type: string
+                  description: "Deprecated. Use jarPullSecret instead."
                 args:
-                  type: array
-                  items:
-                    type: string
+                  description: "Program arguments. Accepts a list of strings (--key, value, ...) or a map of key-value pairs."
+                  x-kubernetes-preserve-unknown-fields: true
                 savepointInterval:
                   type: string
                   format: duration
@@ -92,8 +106,15 @@ spec:
                 lastRestoredSavepointRestoredDate:
                   type: string
                   format: time
+                runningJarDigest:
+                  type: string
+                  description: "SHA256 digest of the currently running JAR layer."
+                runningJarRef:
+                  type: string
+                  description: "OCI reference of the currently running JAR."
                 runningJarURI:
                   type: string
+                  description: "Deprecated. HTTP URI of the currently running JAR."
                 pauseSavepointTriggerId:
                   type: string
                 restoredCount:

example-job.yaml

@@ -1,15 +1,26 @@
 # flink-job-instance.yaml
-apiVersion: flink.logicamp.tech/v1alpha1
+apiVersion: flink.logicamp.dev/v1alpha1
 kind: FlinkJob
 metadata:
   name: my-flink-job
-  namespace: default
 spec:
   key: word-count
   name: "Word Count Example"
-  entryClass: "org.apache.flink.examples.java.wordcount.WordCount"
-  parallelism: 2
-  jarUri: "http://192.168.7.7:8080/product-enrichment-processor.jar"
+  entryClass: "tech.logicamp.logiline.FacilityEnrichment"
+  parallelism: 1
+  jarRef: "lcr.logicamp.tech/logiline/facility-enrichment:1.0.0"
+  jarPullSecret: "lcr-registry-creds"
+  # args accepts two formats:
+  #   Map (recommended):
+  args:
+    kafka-host: kafka.pg-release:9092
+    kafka-group-id: batch-sales-config-dedup-processor
+  #   List (also valid):
+  #   args:
+  #     - "--kafka-host"
+  #     - kafka.pg-release:9092
+  #     - "--kafka-group-id"
+  #     - batch-sales-config-dedup-processor
   flinkConfiguration:
-    taskmanager.numberOfTaskSlots: "2"
-    parallelism.default: "2"
+    taskmanager.numberOfTaskSlots: "1"
+    parallelism.default: "1"

go.mod

@@ -3,8 +3,10 @@ module flink-kube-operator
 go 1.23.2
 
 require (
-	github.com/logi-camp/go-flink-client v0.2.0
-	github.com/matoous/go-nanoid/v2 v2.1.0
+	github.com/danielgtaylor/huma/v2 v2.27.0
+	github.com/gofiber/fiber/v2 v2.52.6
+	github.com/google/go-containerregistry v0.20.3
+	github.com/logi-camp/go-flink-client v0.2.2
 	github.com/samber/lo v1.47.0
 	go.uber.org/zap v1.27.0
 	k8s.io/apimachinery v0.31.3
@@ -12,48 +14,40 @@ require (
 )
 
 require (
-	github.com/Jeffail/gabs/v2 v2.7.0 // indirect
 	github.com/andybalholm/brotli v1.1.1 // indirect
-	github.com/danielgtaylor/casing v1.0.0 // indirect
-	github.com/danielgtaylor/huma/v2 v2.27.0 // indirect
+	github.com/containerd/stargz-snapshotter/estargz v0.16.3 // indirect
+	github.com/docker/cli v27.5.0+incompatible // indirect
+	github.com/docker/distribution v2.8.3+incompatible // indirect
+	github.com/docker/docker-credential-helpers v0.8.2 // indirect
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch/v5 v5.9.0 // indirect
-	github.com/fatih/color v1.18.0 // indirect
-	github.com/fatih/structs v1.1.0 // indirect
-	github.com/go-chi/chi v4.1.2+incompatible // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.23.0 // indirect
-	github.com/goccy/go-yaml v1.15.13 // indirect
-	github.com/gofiber/fiber/v2 v2.52.6 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/gnostic-models v0.6.9 // indirect
 	github.com/google/uuid v1.6.0 // indirect
-	github.com/graphql-go/graphql v0.8.1 // indirect
-	github.com/graphql-go/handler v0.2.4 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/klauspost/compress v1.17.11 // indirect
-	github.com/koron-go/gqlcost v0.3.1 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-runewidth v0.0.16 // indirect
+	github.com/mitchellh/go-homedir v1.1.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.0 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
-	github.com/tent/http-link-go v0.0.0-20130702225549-ac974c61c2f9 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/valyala/fasthttp v1.58.0 // indirect
 	github.com/valyala/tcplisten v1.0.0 // indirect
-	github.com/xeipuuv/gojsonpointer v0.0.0-20190905194746-02993c407bfb // indirect
-	github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415 // indirect
-	github.com/xeipuuv/gojsonschema v1.2.0 // indirect
+	github.com/vbatts/tar-split v0.11.6 // indirect
 	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect
-	golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
-	google.golang.org/protobuf v1.35.1 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	google.golang.org/protobuf v1.36.3 // indirect
 )
 
 require (
-	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/emirpasic/gods v1.12.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
@@ -66,16 +60,11 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
-	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/reactivex/rxgo/v2 v2.5.0
 	github.com/spf13/pflag v1.0.5 // indirect
-	github.com/stretchr/objx v0.5.2 // indirect
-	github.com/stretchr/testify v1.9.0 // indirect
-	github.com/teivah/onecontext v0.0.0-20200513185103-40f981bfd775 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.31.0 // indirect
-	golang.org/x/oauth2 v0.21.0 // indirect
+	golang.org/x/oauth2 v0.25.0 // indirect
 	golang.org/x/sys v0.29.0 // indirect
 	golang.org/x/term v0.26.0 // indirect
 	golang.org/x/text v0.20.0 // indirect
@@ -83,7 +72,7 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/api v0.31.3 // indirect
+	k8s.io/api v0.31.3
 	k8s.io/klog/v2 v2.130.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20241127205056-99599406b04f // indirect
 	k8s.io/utils v0.0.0-20241104163129-6fe5fd82f078 // indirect

helm/chart/Chart.lock

@@ -0,0 +1,6 @@
+dependencies:
+- name: minio
+  repository: https://charts.bitnami.com/bitnami
+  version: 16.0.2
+digest: sha256:9a822e9c5a4eee1b6515c143150c1dd6f84ceb080a7be4573e09396c5916f7d3
+generated: "2025-04-04T14:42:09.771390014+03:30"

helm/chart/Chart.yaml

@@ -2,5 +2,5 @@ apiVersion: v2
 name: flink-kube-operator
 description: Helm chart for flink kube operator
 type: application
-version: 0.1.8
-appVersion: "0.1.0"
+version: 1.3.0
+appVersion: "0.1.1"

helm/chart/templates/NOTES.txt

@@ -17,6 +17,6 @@
 {{- else if contains "ClusterIP" .Values.service.type }}
   export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "flink-kube-operator.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}")
   export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}")
-  echo "Visit http://127.0.0.1:8080 to use your application"
-  kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT
+  echo "Visit http://127.0.0.1:8081 to use your application"
+  kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8081:$CONTAINER_PORT
 {{- end }}

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

@@ -0,0 +1,51 @@
+{{- define "flink.env" -}}
+- name: JOB_MANAGER_RPC_ADDRESS
+  value: "localhost"
+- name: NAMESPACE
+  value: {{ .Release.Namespace }}
+- name: FLINK_PROPERTIES
+  value: |
+    jobmanager.rpc.address: {{ .Release.Name }}-flink-job-manager
+    jobmanager.memory.process.size: {{ .Values.flink.jobManager.processMemory }}
+    taskmanager.memory.process.size: {{ .Values.flink.taskManager.processMemory }}
+    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.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
+    {{- 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/data.pvc.yaml

@@ -1,10 +0,0 @@
-apiVersion: v1
-kind: PersistentVolumeClaim
-metadata:
-  name: {{ .Values.flink.state.data.pvcName }}
-spec:
-  accessModes:
-    - ReadWriteOnce
-  resources:
-    requests:
-      storage: {{ .Values.flink.state.data.size }}  # Use size defined in values.yaml

helm/chart/templates/flink/deploy.yaml

@@ -1,152 +0,0 @@
-{{- define "flink.env" -}}
-- name: JOB_MANAGER_RPC_ADDRESS
-  value: "localhost"
-- name: NAMESPACE
-  value: {{ .Release.Namespace }}
-- name: FLINK_PROPERTIES
-  value: |
-    jobmanager.rpc.address: localhost
-    jobmanager.memory.process.size: 2048m
-    taskmanager.memory.process.size: 2048m
-    taskmanager.data.port: 6125
-    taskmanager.numberOfTaskSlots: {{ .Values.flink.taskManager.numberOfTaskSlots }}
-    parallelism.default: {{ .Values.flink.parallelism.default }}
-    state.backend: {{ .Values.flink.state.backend }}
-    rest.port: 8081
-    rootLogger.level = DEBUG
-    rootLogger.appenderRef.console.ref = ConsoleAppender
-    high-availability.type: kubernetes
-    kubernetes.namespace: {{ .Release.Namespace }}
-    kubernetes.cluster-id: cluster-one
-    web.upload.dir: {{ .Values.flink.state.data.dir }}/web-upload
-    state.checkpoints.dir: file://{{ .Values.flink.state.data.dir }}/checkpoints
-    state.backend.rocksdb.localdir: file://{{ .Values.flink.state.data.dir }}/rocksdb
-    high-availability.storageDir: file://{{ .Values.flink.state.ha.dir }}
-    state.savepoints.dir: file://{{ .Values.flink.state.savepoints.dir }}
-    state.backend.incremental: false
-{{- end }}
-
-{{- define "flink.volumeMounts" -}}
-- name: flink-data
-  mountPath: {{ .Values.flink.state.data.dir }}/data
-- name: flink-data
-  mountPath: {{ .Values.flink.state.data.dir }}/rocksdb
-  subPath: rocksdb
-- name: flink-data
-  mountPath: {{ .Values.flink.state.data.dir }}/checkpoints
-  subPath: checkpoints
-- name: flink-data
-  mountPath: {{ .Values.flink.state.data.dir }}/web-upload
-  subPath: web-upload
-- name: flink-ha
-  mountPath: {{ .Values.flink.state.ha.dir }}
-- name: flink-savepoints
-  mountPath: {{ .Values.flink.state.savepoints.dir }}
-{{- end }}
-
-{{- define "flink.volumes" -}}
-- name: flink-data
-  persistentVolumeClaim:
-    claimName: {{ .Values.flink.state.data.pvcName }}
-- name: flink-savepoints
-  persistentVolumeClaim:
-    claimName: {{ .Values.flink.state.savepoints.pvcName }}
-- name: flink-ha
-  persistentVolumeClaim:
-    claimName: {{ .Values.flink.state.ha.pvcName }}
-{{- end }}
-
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: {{ .Release.Name }}-flink
-  labels:
-    app.kubernetes.io/name: {{ .Release.Name }}-flink
-    app.kubernetes.io/instance: {{ .Release.Name }}
-spec:
-  replicas: 1
-  strategy:
-    type: Recreate
-  selector:
-    matchLabels:
-      app.kubernetes.io/name: {{ .Release.Name }}-flink
-      app.kubernetes.io/instance: {{ .Release.Name }}
-  template:
-    metadata:
-      labels:
-        app.kubernetes.io/name: {{ .Release.Name }}-flink
-        app.kubernetes.io/instance: {{ .Release.Name }}
-    spec:
-      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}
-      initContainers:
-        - name: volume-mount-hack
-          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
-          runAsUser: 0
-          command: ["sh", "-c", "chown -R flink {{ .Values.flink.state.data.dir }}/data {{ .Values.flink.state.data.dir }}/rocksdb {{ .Values.flink.state.data.dir }}/checkpoints {{ .Values.flink.state.data.dir }}/web-upload {{ .Values.flink.state.ha.dir }} {{ .Values.flink.state.savepoints.dir }}"]
-          volumeMounts:
-            {{- include "flink.volumeMounts" . | nindent 12 }}
-      containers:
-        - name: jobmanager
-          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
-          imagePullPolicy: Always
-          args: ["jobmanager"]
-          ports:
-            - containerPort: 6123  # JobManager RPC port
-              name: rpc
-            - containerPort: 6124  # JobManager blob server port
-              name: blob
-            - containerPort: 6125  # JobManager queryable state port
-              name: query
-            - containerPort: 8081  # JobManager Web UI port
-              name: ui
-          env:
-            {{- include "flink.env" . | nindent 12 }}
-          volumeMounts:
-            {{- include "flink.volumeMounts" . | nindent 12 }}
-
-        - name: taskmanager
-          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
-          imagePullPolicy: Always
-          args: ["taskmanager"]
-          ports:
-            - containerPort: 6121  # TaskManager data port
-              name: data
-            - containerPort: 6122  # TaskManager RPC port
-              name: rpc
-          env:
-            {{- include "flink.env" . | nindent 12 }}
-          volumeMounts:
-            {{- include "flink.volumeMounts" . | nindent 12 }}
-        - name: operator
-          securityContext:
-            {{- toYaml .Values.securityContext | nindent 12 }}
-          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
-          imagePullPolicy: {{ .Values.image.pullPolicy }}
-          ports:
-            - name: http
-              containerPort: {{ .Values.service.port }}
-              protocol: TCP
-          env:
-            - name: FLINK_API_URL
-              value: localhost:8081
-            - name: SAVEPOINT_PATH
-              value: file://{{ .Values.flink.state.savepoints.dir }}
-          resources:
-            {{- toYaml .Values.resources | nindent 12 }}
-          volumeMounts:
-            {{- include "flink.volumeMounts" . | nindent 12 }}
-      volumes:
-        {{- include "flink.volumes" . | nindent 8 }}
-
-      {{- with .Values.nodeSelector }}
-      nodeSelector:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
-      {{- with .Values.affinity }}
-      affinity:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}
-      {{- with .Values.tolerations }}
-      tolerations:
-        {{- toYaml . | nindent 8 }}
-      {{- end }}

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

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

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

@@ -0,0 +1,103 @@
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .Release.Name }}-flink-job-manager
+  labels:
+    app: {{ .Release.Name }}-flink-operator
+    component: {{ .Release.Name }}-flink-job-manager
+spec:
+  replicas: 1
+  strategy:
+    type: Recreate
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-flink-operator
+      component: {{ .Release.Name }}-flink-job-manager
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-flink-operator
+        component: {{ .Release.Name }}-flink-job-manager
+    spec:
+      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}
+      initContainers:
+        - name: volume-mount-hack
+          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
+          securityContext:
+            runAsUser: 0
+          command: ["sh", "-c", "chown -R flink {{ .Values.flink.state.ha.dir }}"]
+          volumeMounts:
+            - name: flink-ha
+              mountPath: {{ .Values.flink.state.ha.dir }}
+      containers:
+        - name: jobmanager
+          image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
+          imagePullPolicy: Always
+          args: ["jobmanager"]
+          ports:
+            - containerPort: 6123  # JobManager RPC port
+              name: rpc
+            - containerPort: 6124  # JobManager blob server port
+              name: blob
+            - containerPort: 6125  # JobManager queryable state port
+              name: query
+            - containerPort: 8081  # JobManager Web UI port
+              name: ui
+          env:
+            {{- include "flink.env" . | nindent 12 }}
+            - name: POD_IP
+              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://{{ .Release.Name }}-minio:9000"
+            - name: AWS_ACCESS_KEY_ID
+              valueFrom:
+                secretKeyRef:
+                  name: {{ .Release.Name }}-minio
+                  key: root-user
+            - name: AWS_SECRET_ACCESS_KEY
+              valueFrom:
+                secretKeyRef:
+                  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 }}-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 }}
+      {{- end }}
+      {{- with .Values.affinity }}
+      affinity:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
+      {{- with .Values.tolerations }}
+      tolerations:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}

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

@@ -0,0 +1,28 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ .Release.Name }}-flink-job-manager
+  labels:
+    app.kubernetes.io/name: {{ .Release.Name }}-flink-job-manager
+    app.kubernetes.io/instance: {{ .Release.Name }}
+spec:
+  ports:
+    - name: flink-web-ui
+      port: 8081
+      targetPort: 8081
+    - name: rpc
+      port: 6123
+      targetPort: 6123
+    - name: blob
+      port: 6124
+      targetPort: 6124
+    - name: query
+      port: 6125
+      targetPort: 6125
+    - name: operator
+      port: 3000
+      targetPort: 3000
+  selector:
+    app: {{ .Release.Name }}-flink-operator
+    component: {{ .Release.Name }}-flink-job-manager
+  type: ClusterIP  # Change to LoadBalancer if you want external access

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/savepoint.pvc.yaml

@@ -1,10 +0,0 @@
-apiVersion: v1
-kind: PersistentVolumeClaim
-metadata:
-  name: {{ .Values.flink.state.savepoints.pvcName }}
-spec:
-  accessModes:
-    - ReadWriteOnce
-  resources:
-    requests:
-      storage: {{ .Values.flink.state.savepoints.size }}  # Use size defined in values.yaml

helm/chart/templates/flink/service.yaml

@@ -1,19 +0,0 @@
-apiVersion: v1
-kind: Service
-metadata:
-  name: flink
-  labels:
-    app.kubernetes.io/name: {{ .Release.Name }}-flink
-    app.kubernetes.io/instance: {{ .Release.Name }}
-spec:
-  ports:
-    - port: 8081
-      name: flink-web-ui
-      targetPort: 8081
-    - port: 3000
-      name: operator
-      targetPort: 3000
-  selector:
-    app.kubernetes.io/name: {{ .Release.Name }}-flink
-    app.kubernetes.io/instance: {{ .Release.Name }}
-  type: ClusterIP  # Change to LoadBalancer if you want external access

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

@@ -0,0 +1,80 @@
+
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: {{ .Release.Name }}-flink-task-manager
+  labels:
+    app: {{ .Release.Name }}-flink-operator
+    component: taskmanager
+spec:
+  serviceName: {{ .Release.Name }}-flink-task-manager
+  replicas: {{ .Values.flink.taskManager.replicas }}
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-flink-operator
+      component: {{ .Release.Name }}-flink-task-manager
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-flink-operator
+        component: {{ .Release.Name }}-flink-task-manager
+    spec:
+      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}
+      containers:
+      - name: task-manager
+        image: {{ .Values.flink.image.repository }}:{{ .Values.flink.image.tag }}
+        imagePullPolicy: Always
+        args: ["taskmanager"]
+        env:
+        {{- include "flink.env" . | nindent 8 }}
+        - name: POD_IP
+          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://{{ .Release.Name }}-minio:9000"
+        - name: AWS_ACCESS_KEY_ID
+          valueFrom:
+            secretKeyRef:
+              name: {{ .Release.Name }}-minio
+              key: root-user
+        - name: AWS_SECRET_ACCESS_KEY
+          valueFrom:
+            secretKeyRef:
+              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
+    spec:
+      accessModes: [ ReadWriteOnce ]
+      resources:
+        requests:
+          storage: {{ .Values.flink.taskManager.storage.rocksDb.size }}

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/service.yaml

@@ -1,9 +1,10 @@
 apiVersion: v1
 kind: Service
 metadata:
-  name: {{ include "flink-kube-operator.fullname" . }}
+  name: {{ .Release.Name }}-flink-operator
   labels:
-    {{- include "flink-kube-operator.labels" . | nindent 4 }}
+    app: {{ .Release.Name }}-flink-operator
+    component: {{ .Release.Name }}-flink-operator
 spec:
   type: {{ .Values.service.type }}
   ports:
@@ -12,4 +13,5 @@ spec:
       protocol: TCP
       name: http
   selector:
-    {{- include "flink-kube-operator.selectorLabels" . | nindent 4 }}
+    app: {{ .Release.Name }}-flink-operator
+    component: {{ .Release.Name }}-flink-operator

helm/chart/templates/operator/statefulset.yaml

@@ -0,0 +1,72 @@
+
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: {{ .Release.Name }}-flink-operator
+  labels:
+    app: {{ .Release.Name }}-flink-operator
+    component: {{ .Release.Name }}-flink-operator
+spec:
+  serviceName: {{ .Release.Name }}-flink-operator
+  replicas: 1
+  selector:
+    matchLabels:
+      app: {{ .Release.Name }}-flink-operator
+      component: {{ .Release.Name }}-flink-operator
+  template:
+    metadata:
+      labels:
+        app: {{ .Release.Name }}-flink-operator
+        component: {{ .Release.Name }}-flink-operator
+    spec:
+      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}
+      initContainers:
+        - name: wait-for-jobmanager
+          image: curlimages/curl:8.5.0  # Lightweight curl image
+          command:
+            - sh
+            - -c
+            - |
+              echo "Waiting for Flink JobManager to be ready..."
+              until curl -sSf "http://{{ .Release.Name }}-flink-job-manager:8081/taskmanagers"; do
+                echo "JobManager not ready yet - retrying in 5s..."
+                sleep 5
+              done
+              echo "JobManager is ready!"
+      containers:
+        - name: operator
+          securityContext:
+            {{- toYaml .Values.securityContext | nindent 12 }}
+          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
+          imagePullPolicy: {{ .Values.image.pullPolicy }}
+          ports:
+            - name: http
+              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: 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
+              valueFrom:
+                secretKeyRef:
+                  name: {{ .Release.Name }}-minio
+                  key: root-user
+            - name: AWS_SECRET_ACCESS_KEY
+              valueFrom:
+                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.1
 
 # 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,27 +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.0-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
+    default: 1 # Default parallelism for Flink jobs
 
   state:
-    backend: rocksdb  # Use RocksDB for state backend
-    savepoints:
-      dir: "/opt/flink/savepoints"  # Directory to store savepoints
-      pvcName: flink-savepoints-pvc  # PVC for savepoints persistence
-      size: 10Gi  # PVC size for savepoints storage
-    data:
-      dir: "/opt/flink/data"  # Directory to store checkpoints/web-upload/rocksdb
-      pvcName: flink-data-pvc  # PVC for checkpoints/web-upload/rocksdb
-      size: 10Gi  # PVC size for checkpoints/web-upload/rocksdb
+    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: 100  # 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 # Number of task manager replicas
+    storage:
+      rocksDb:
+        size: 4Gi
+    resources:
+      limits:
+        cpu: 3
+        memory: 4Gi
+      requests:
+        cpu: 1
+        memory: 2Gi

helm/index.yaml

@@ -1,9 +1,155 @@
 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: "2024-12-25T01:46:01.931022489+03:30"
+    created: "2025-04-04T13:50:27.971040367+03:30"
+    description: Helm chart for flink kube operator
+    digest: 00acef7878bcf372d036fabaac400913097d678087a756102b54a28428197bdf
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-0.1.14.tgz
+    version: 0.1.14
+  - apiVersion: v2
+    appVersion: 0.1.0
+    created: "2025-03-04T23:13:19.698003661+03:30"
+    description: Helm chart for flink kube operator
+    digest: d104b9242362415a7b920e4e2af975730e208ff73db17b8d2afd11ea8b78b4a2
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-0.1.13.tgz
+    version: 0.1.13
+  - apiVersion: v2
+    appVersion: 0.1.0
+    created: "2025-03-04T23:13:19.697555829+03:30"
+    description: Helm chart for flink kube operator
+    digest: f58802990389ecde00a49a442f6e83a007e281e972d07f2979657d2763fe94ba
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-0.1.12.tgz
+    version: 0.1.12
+  - apiVersion: v2
+    appVersion: 0.1.0
+    created: "2025-03-04T18:04:35.491747333+03:30"
+    description: Helm chart for flink kube operator
+    digest: 0daa98c63b443018c2072a2d7448c972faff2274fb04433c613532b408cd3ab1
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-0.1.11.tgz
+    version: 0.1.11
+  - apiVersion: v2
+    appVersion: 0.1.0
+    created: "2025-03-04T18:04:35.490697387+03:30"
+    description: Helm chart for flink kube operator
+    digest: e091256eeb8640b61443cbe4781426ef493737ab0ac1145e568426bb2c1ed3ba
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-0.1.10.tgz
+    version: 0.1.10
+  - apiVersion: v2
+    appVersion: 0.1.0
+    created: "2025-03-04T18:04:35.495842696+03:30"
+    description: Helm chart for flink kube operator
+    digest: abc08853c65ba36ff3485f182555522408e150f2508d4cac672d588972ddca3c
+    name: flink-kube-operator
+    type: application
+    urls:
+    - flink-kube-operator-0.1.9.tgz
+    version: 0.1.9
+  - apiVersion: v2
+    appVersion: 0.1.0
+    created: "2025-03-04T18:04:35.495392608+03:30"
     description: Helm chart for flink kube operator
     digest: 3986a0a2348db1e17a1524eb0d87eabf6d64050d4007c5b393f723393cc4b675
     name: flink-kube-operator
@@ -13,7 +159,7 @@ entries:
     version: 0.1.8
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.930596114+03:30"
+    created: "2025-03-04T18:04:35.494948853+03:30"
     description: Helm chart for flink kube operator
     digest: 1bbeb92ecd10e36fa7d742a61cced0d842139ada0cfeff6fa1b0cf8718189235
     name: flink-kube-operator
@@ -23,7 +169,7 @@ entries:
     version: 0.1.7
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.930060136+03:30"
+    created: "2025-03-04T18:04:35.49450822+03:30"
     description: Helm chart for flink kube operator
     digest: 4031f4a79e65f6c5e60b6ebf9dd7e2a663b1fb6f893056ad81ca33660f94406e
     name: flink-kube-operator
@@ -33,7 +179,7 @@ entries:
     version: 0.1.6
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.929536043+03:30"
+    created: "2025-03-04T18:04:35.494040193+03:30"
     description: Helm chart for flink kube operator
     digest: 22ed155c8538ca5e7dc26863304eb9f76b09c454edbf709a891d7ccc440f35f6
     name: flink-kube-operator
@@ -43,7 +189,7 @@ entries:
     version: 0.1.5
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.929001953+03:30"
+    created: "2025-03-04T18:04:35.493584927+03:30"
     description: Helm chart for flink kube operator
     digest: b548a64ef89bbcd12d92fefffd1fd37758e8fccda02aecd97c7519a08f10fa4a
     name: flink-kube-operator
@@ -53,7 +199,7 @@ entries:
     version: 0.1.4
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.92842706+03:30"
+    created: "2025-03-04T18:04:35.493138547+03:30"
     description: Helm chart for flink kube operator
     digest: 05a9664f574e2d5d1cca764efb6481ad21b9176663b907973a8ef5264f15a91f
     name: flink-kube-operator
@@ -63,7 +209,7 @@ entries:
     version: 0.1.3
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.927900408+03:30"
+    created: "2025-03-04T18:04:35.492696005+03:30"
     description: Helm chart for flink kube operator
     digest: 89345b1a9a79aa18b646705aeb8cfdc547629600cb8a00708a3f64d188f296f2
     name: flink-kube-operator
@@ -73,7 +219,7 @@ entries:
     version: 0.1.2
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.927362234+03:30"
+    created: "2025-03-04T18:04:35.490170385+03:30"
     description: Helm chart for flink kube operator
     digest: 1d2af9af6b9889cc2962d627946464766f1b65b05629073b7fffb9a98cd957e2
     name: flink-kube-operator
@@ -83,7 +229,7 @@ entries:
     version: 0.1.1
   - apiVersion: v2
     appVersion: 0.1.0
-    created: "2024-12-25T01:46:01.926717732+03:30"
+    created: "2025-03-04T18:04:35.489734651+03:30"
     description: Helm chart for flink kube operator
     digest: 0890d955904e6a3b2155c086a933b27e45266d896fb69eaad0e811dea40414da
     name: flink-kube-operator
@@ -91,4 +237,4 @@ entries:
     urls:
     - flink-kube-operator-0.1.0.tgz
     version: 0.1.0
-generated: "2024-12-25T01:46:01.925995262+03:30"
+generated: "2025-07-18T18:09:46.244672127+03:30"

internal/crd/auth.go

@@ -0,0 +1,83 @@
+package crd
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/google/go-containerregistry/pkg/authn"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+)
+
+type dockerConfigJSON struct {
+	Auths map[string]authn.AuthConfig `json:"auths"`
+}
+
+// ResolveAuth returns an OCI authenticator for the given registry.
+// It checks in order: pull secret from K8s, then environment variables.
+// Returns authn.Anonymous if no credentials are found.
+func (crd *Crd) ResolveAuth(pullSecretName string, server string) (authn.Authenticator, error) {
+	if pullSecretName != "" {
+		return crd.authFromPullSecret(pullSecretName, server)
+	}
+	return authFromEnv(server), nil
+}
+
+func (crd *Crd) authFromPullSecret(secretName string, server string) (authn.Authenticator, error) {
+	ns := os.Getenv("NAMESPACE")
+	secret := &corev1.Secret{}
+	err := crd.runtimeClient.Get(context.Background(), types.NamespacedName{
+		Namespace: ns,
+		Name:      secretName,
+	}, secret)
+	if err != nil {
+		return nil, fmt.Errorf("reading pull secret %q: %w", secretName, err)
+	}
+
+	if secret.Type != corev1.SecretTypeDockerConfigJson {
+		return nil, fmt.Errorf("secret %q is not of type kubernetes.io/dockerconfigjson", secretName)
+	}
+
+	dockerConfigJSONBytes, ok := secret.Data[corev1.DockerConfigJsonKey]
+	if !ok {
+		return nil, fmt.Errorf("secret %q missing .dockerconfigjson key", secretName)
+	}
+
+	var cfg dockerConfigJSON
+	if err := json.Unmarshal(dockerConfigJSONBytes, &cfg); err != nil {
+		return nil, fmt.Errorf("parsing docker config from secret %q: %w", secretName, err)
+	}
+
+	if server != "" {
+		if entry, ok := cfg.Auths[server]; ok {
+			return authn.FromConfig(entry), nil
+		}
+	}
+
+	for _, entry := range cfg.Auths {
+		return authn.FromConfig(entry), nil
+	}
+
+	return authn.Anonymous, nil
+}
+
+func authFromEnv(server string) authn.Authenticator {
+	username := os.Getenv("REGISTRY_USERNAME")
+	password := os.Getenv("REGISTRY_PASSWORD")
+	envServer := os.Getenv("REGISTRY_SERVER")
+
+	if username == "" || password == "" {
+		return authn.Anonymous
+	}
+
+	if envServer != "" && server != "" && envServer != server {
+		return authn.Anonymous
+	}
+
+	return authn.FromConfig(authn.AuthConfig{
+		Username: username,
+		Password: password,
+	})
+}

internal/crd/finalizer.go

@@ -5,7 +5,6 @@ import (
 	"flink-kube-operator/internal/crd/v1alpha1"
 	"flink-kube-operator/pkg"
 
-	"github.com/reactivex/rxgo/v2"
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -13,28 +12,26 @@ import (
 
 var FinalizerChannel chan (types.UID) = make(chan (types.UID))
 
-func (crd Crd) manageFinalizer(jobEventObservable rxgo.Observable) {
+func (crd Crd) manageFinalizer(jobEventChannel chan FlinkJobCrdEvent) {
 
 	finalizerName := "flink-operator.logicamp.tech/finalizer"
-	for j := range jobEventObservable.Observe() {
+	for jobEvent := range jobEventChannel {
+		pkg.Logger.Debug("[crd] [manage-finalizer] main loop", zap.String("name", jobEvent.Job.Name))
 		go func() {
-
-			jobEvent := j.V.(*FlinkJobCrdEvent)
-
 			if jobEvent.Job.GetDeletionTimestamp() != nil {
 				// Resource is being deleted
 				if controllerutil.ContainsFinalizer(jobEvent.Job, finalizerName) {
 					// Perform cleanup
 					pkg.Logger.Debug("[finalizer] stopping managed job", zap.String("name", jobEvent.Job.GetName()))
 					if err := crd.cleanupResources(jobEvent.Job); err != nil {
-						pkg.Logger.Error("[crd] [manage-finalizer] cleanup failed", zap.Error(err))
+						pkg.Logger.Info("[crd] [manage-finalizer] cleanup failed", zap.Error(err))
 						return
 					}
 
 					// Remove finalizer
 					controllerutil.RemoveFinalizer(jobEvent.Job, finalizerName)
 					if err := crd.runtimeClient.Update(context.Background(), jobEvent.Job); err != nil {
-						pkg.Logger.Error("[crd] [manage-finalizer] failed to remove finalizer", zap.Error(err))
+						pkg.Logger.Info("[crd] [manage-finalizer] failed to remove finalizer", zap.Error(err))
 						return
 					}
 					pkg.Logger.Debug("[crd] [manage-finalizer] job removed", zap.String("name", jobEvent.Job.GetName()))
@@ -49,7 +46,7 @@ func (crd Crd) manageFinalizer(jobEventObservable rxgo.Observable) {
 				pkg.Logger.Debug("[finalizer] adding job")
 				// Update the resource to add the finalizer
 				if err := crd.runtimeClient.Update(context.Background(), jobEvent.Job); err != nil {
-					pkg.Logger.Error("[finalizer] failed to add", zap.Error(err))
+					pkg.Logger.Info("[finalizer] failed to add", zap.Error(err))
 					return
 				}
 			}

internal/crd/new.go

@@ -2,6 +2,7 @@ package crd
 
 import (
 	"flink-kube-operator/internal/crd/v1alpha1"
+	"os"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
@@ -12,7 +13,7 @@ import (
 )
 
 type Crd struct {
-	client        dynamic.NamespaceableResourceInterface
+	client        dynamic.ResourceInterface
 	runtimeClient client.Client
 }
 
@@ -32,12 +33,12 @@ func New() *Crd {
 		panic(err)
 	}
 
-	shema := runtime.NewScheme()
-	v1alpha1.AddKnownTypes(shema)
+	scheme := runtime.NewScheme()
+	v1alpha1.AddKnownTypes(scheme)
 	// Get FlinkJob resource interface
-	flinkJobClient := dynamicClient.Resource(v1alpha1.FlinkJobGVR)
+	flinkJobClient := dynamicClient.Resource(v1alpha1.FlinkJobGVR).Namespace(os.Getenv("NAMESPACE"))
 	runtimeClient, err := client.New(config.GetConfigOrDie(), client.Options{
-		Scheme: shema,
+		Scheme: scheme,
 	})
 	if err != nil {
 		panic(err)
@@ -47,11 +48,13 @@ func New() *Crd {
 		runtimeClient: runtimeClient,
 	}
 
-	// Watch for FlinkJob creation
-	jobEventObservable := crd.watchFlinkJobs()
+	jobEventCh := make(chan FlinkJobCrdEvent)
 
 	// add finalizer to new resources
-	go crd.manageFinalizer(jobEventObservable)
+	go crd.manageFinalizer(jobEventCh)
+
+	// Watch for FlinkJob creation
+	crd.watchFlinkJobs(jobEventCh)
 
 	return &crd
 }

internal/crd/patch.go

@@ -23,7 +23,6 @@ func (crd *Crd) Patch(jobUid types.UID, patchData map[string]interface{}) error
 
 	// Patch the status sub-resource
 	unstructuredJob, err := crd.client.
-		Namespace(job.GetNamespace()).
 		Patch(
 			context.Background(),
 			job.GetName(),

internal/crd/v1alpha1/flink_job.go

@@ -1,7 +1,10 @@
 package v1alpha1
 
 import (
+	"encoding/json"
 	"errors"
+	"sort"
+	"strings"
 	"time"
 
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -9,14 +12,70 @@ import (
 
 //go:generate go run sigs.k8s.io/controller-tools/cmd/controller-gen object paths=$GOFILE
 
+// Args is a list of program arguments passed to the Flink job.
+// It accepts two formats in JSON/YAML:
+//
+//	List:  ["--kafka-host", "broker:9092", "--kafka-group-id", "my-group"]
+//	Map:   {kafka-host: "broker:9092", kafka-group-id: "my-group"}
+//
+// Map keys are automatically prefixed with "--" and expanded into alternating
+// key/value pairs. The internal representation is always a flat []string.
+type Args []string
+
+// UnmarshalJSON implements dual-format deserialization.
+func (a *Args) UnmarshalJSON(data []byte) error {
+	// Try map format first: {"key": "value", ...}
+	var m map[string]string
+	if err := json.Unmarshal(data, &m); err == nil {
+		*a = make(Args, 0, len(m)*2)
+		// Sort keys for deterministic ordering.
+		keys := make([]string, 0, len(m))
+		for k := range m {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		for _, k := range keys {
+			*a = append(*a, "--"+k, m[k])
+		}
+		return nil
+	}
+
+	// Fall back to list format: ["--key", "value", ...]
+	var s []string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return err
+	}
+	*a = Args(s)
+	return nil
+}
+
+// MarshalJSON serializes Args in the compact map format.
+func (a Args) MarshalJSON() ([]byte, error) {
+	m := make(map[string]string, len(a)/2)
+	for i := 0; i < len(a)-1; i += 2 {
+		key := strings.TrimPrefix(a[i], "--")
+		m[key] = a[i+1]
+	}
+	return json.Marshal(m)
+}
+
 type FlinkJobSpec struct {
-	Key               string          `json:"key"`
-	Name              string          `json:"name"`
-	Parallelism       int             `json:"parallelism"`
-	JarURI            string          `json:"jarUri"`
+	Key              string          `json:"key"`
+	Name             string          `json:"name"`
+	FlinkCluster     string          `json:"flinkCluster"`
+	Parallelism      int             `json:"parallelism"`
+	JarRef           string          `json:"jarRef,omitempty"`
+	JarPullSecret    string          `json:"jarPullSecret,omitempty"`
 	SavepointInterval metaV1.Duration `json:"savepointInterval"`
-	EntryClass        string          `json:"entryClass"`
-	Args              []string        `json:"args"`
+	EntryClass       string          `json:"entryClass"`
+	Args             Args            `json:"args,omitempty"`
+
+	// Deprecated: use JarRef instead. Kept for backward compatibility.
+	JarURI string `json:"jarUri,omitempty"`
+	// Deprecated: use JarPullSecret instead. Kept for backward compatibility.
+	JarURIBasicAuthUsername *string `json:"jarURIBasicAuthUsername,omitempty"`
+	// Deprecated: use JarPullSecret instead. Kept for backward compatibility.
+	JarURIBasicAuthPassword *string `json:"jarURIBasicAuthPassword,omitempty"`
 }
 
 type FlinkJobStatus struct {
@@ -32,7 +91,10 @@ type FlinkJobStatus struct {
 	LastRestoredSavepointDate         *time.Time      `json:"lastRestoredSavepointDate,omitempty"`
 	LastRestoredSavepointRestoredDate *time.Time      `json:"lastRestoredSavepointRestoredDate,omitempty"`
 	RestoredCount                     int             `json:"restoredCount,omitempty"`
-	RunningJarURI                     *string         `json:"runningJarURI"`
+	RunningJarDigest                  *string         `json:"runningJarDigest,omitempty"`
+	RunningJarRef                     *string         `json:"runningJarRef,omitempty"`
+	// Deprecated: kept for backward compatibility during migration.
+	RunningJarURI *string `json:"runningJarURI,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
@@ -87,3 +149,48 @@ const (
 	LifeCycleStatusHealthy             LifeCycleStatus = "HEALTHY"
 	LifeCycleStatusFailed              LifeCycleStatus = "FAILED"
 )
+
+// EffectiveJarRef returns the OCI reference, preferring JarRef over the deprecated JarURI.
+func (s *FlinkJobSpec) EffectiveJarRef() string {
+	if s.JarRef != "" {
+		return s.JarRef
+	}
+	return s.JarURI
+}
+
+// IsOCISource returns true if the job uses an OCI reference (jarRef) rather than an HTTP URL (jarUri).
+func (s *FlinkJobSpec) IsOCISource() bool {
+	return s.JarRef != ""
+}
+
+// RunningRef returns the currently tracked running jar reference,
+// preferring RunningJarRef (OCI) over the deprecated RunningJarURI.
+func (st *FlinkJobStatus) RunningRef() string {
+	if st.RunningJarRef != nil {
+		return *st.RunningJarRef
+	}
+	if st.RunningJarURI != nil {
+		return *st.RunningJarURI
+	}
+	return ""
+}
+
+// NeedsUpgrade returns true if the spec's jar reference differs from
+// what is currently running, indicating an upgrade is needed.
+func (s *FlinkJobSpec) NeedsUpgrade(st FlinkJobStatus) bool {
+	running := st.RunningRef()
+	if running == "" {
+		return false
+	}
+	return s.EffectiveJarRef() != running
+}
+
+// RunningRefPatchKey returns the status patch key and value for tracking
+// which jar reference is currently running.
+func (s *FlinkJobSpec) RunningRefPatchData() (key string, value interface{}) {
+	if s.IsOCISource() {
+		ref := s.EffectiveJarRef()
+		return "runningJarRef", &ref
+	}
+	return "runningJarURI", s.JarURI
+}

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/crd/v1alpha1/zz_generated.deepcopy.go

@@ -5,6 +5,8 @@
 package v1alpha1
 
 import (
+	"time"
+
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -13,7 +15,8 @@ func (in *FlinkJob) DeepCopyInto(out *FlinkJob) {
 	*out = *in
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
-	out.Spec = in.Spec
+	in.Spec.DeepCopyInto(&out.Spec)
+	in.Status.DeepCopyInto(&out.Status)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FlinkJob.
@@ -34,6 +37,111 @@ func (in *FlinkJob) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *FlinkJobSpec) DeepCopyInto(out *FlinkJobSpec) {
+	*out = *in
+	if in.JarURIBasicAuthUsername != nil {
+		in, out := &in.JarURIBasicAuthUsername, &out.JarURIBasicAuthUsername
+		*out = new(string)
+		**out = **in
+	}
+	if in.JarURIBasicAuthPassword != nil {
+		in, out := &in.JarURIBasicAuthPassword, &out.JarURIBasicAuthPassword
+		*out = new(string)
+		**out = **in
+	}
+	if in.Args != nil {
+		in, out := &in.Args, &out.Args
+		*out = make(Args, len(*in))
+		copy(*out, *in)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FlinkJobSpec.
+func (in *FlinkJobSpec) DeepCopy() *FlinkJobSpec {
+	if in == nil {
+		return nil
+	}
+	out := new(FlinkJobSpec)
+	in.DeepCopyInto(out)
+	return out
+}
+
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *FlinkJobStatus) DeepCopyInto(out *FlinkJobStatus) {
+	*out = *in
+	if in.LastSavepointPath != nil {
+		in, out := &in.LastSavepointPath, &out.LastSavepointPath
+		*out = new(string)
+		**out = **in
+	}
+	if in.JarId != nil {
+		in, out := &in.JarId, &out.JarId
+		*out = new(string)
+		**out = **in
+	}
+	if in.JobId != nil {
+		in, out := &in.JobId, &out.JobId
+		*out = new(string)
+		**out = **in
+	}
+	if in.Error != nil {
+		in, out := &in.Error, &out.Error
+		*out = new(string)
+		**out = **in
+	}
+	if in.SavepointTriggerId != nil {
+		in, out := &in.SavepointTriggerId, &out.SavepointTriggerId
+		*out = new(string)
+		**out = **in
+	}
+	if in.PauseSavepointTriggerId != nil {
+		in, out := &in.PauseSavepointTriggerId, &out.PauseSavepointTriggerId
+		*out = new(string)
+		**out = **in
+	}
+	if in.LastSavepointDate != nil {
+		in, out := &in.LastSavepointDate, &out.LastSavepointDate
+		*out = new(time.Time)
+		**out = **in
+	}
+	if in.LastRestoredSavepointDate != nil {
+		in, out := &in.LastRestoredSavepointDate, &out.LastRestoredSavepointDate
+		*out = new(time.Time)
+		**out = **in
+	}
+	if in.LastRestoredSavepointRestoredDate != nil {
+		in, out := &in.LastRestoredSavepointRestoredDate, &out.LastRestoredSavepointRestoredDate
+		*out = new(time.Time)
+		**out = **in
+	}
+	if in.RunningJarDigest != nil {
+		in, out := &in.RunningJarDigest, &out.RunningJarDigest
+		*out = new(string)
+		**out = **in
+	}
+	if in.RunningJarRef != nil {
+		in, out := &in.RunningJarRef, &out.RunningJarRef
+		*out = new(string)
+		**out = **in
+	}
+	if in.RunningJarURI != nil {
+		in, out := &in.RunningJarURI, &out.RunningJarURI
+		*out = new(string)
+		**out = **in
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new FlinkJobStatus.
+func (in *FlinkJobStatus) DeepCopy() *FlinkJobStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(FlinkJobStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *FlinkJobList) DeepCopyInto(out *FlinkJobList) {
 	*out = *in

internal/crd/watch.go

@@ -7,7 +7,6 @@ import (
 
 	"flink-kube-operator/pkg"
 
-	"github.com/reactivex/rxgo/v2"
 	"go.uber.org/zap"
 	metaV1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -15,53 +14,57 @@ import (
 	"k8s.io/apimachinery/pkg/watch"
 )
 
-func (crd Crd) watchFlinkJobs() rxgo.Observable {
-
-	ch := make(chan rxgo.Item)
+func (crd Crd) watchFlinkJobs(ch chan FlinkJobCrdEvent) {
 
 	go func() {
-		pkg.Logger.Debug("[crd] starting watch")
-		watcher, err := crd.client.Namespace(os.Getenv("NAMESPACE")).Watch(context.Background(), metaV1.ListOptions{})
-		if err != nil {
-			panic(err)
-		}
-		defer watcher.Stop()
-		for event := range watcher.ResultChan() {
-			unstructuredJob := event.Object.(*unstructured.Unstructured)
-			unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
+		for {
+			pkg.Logger.Debug("[crd] starting watch")
+			watcher, err := crd.client.Watch(context.Background(), metaV1.ListOptions{})
 			if err != nil {
-				pkg.Logger.Error("cannot create unstructured map", zap.Error(err))
-				continue
+				panic(err)
 			}
-			job := &v1alpha1.FlinkJob{}
+			namespace := os.Getenv("NAMESPACE")
+			pkg.Logger.Debug("[crd] [watch]", zap.String("namespace", namespace))
+			for event := range watcher.ResultChan() {
+				unstructuredJob := event.Object.(*unstructured.Unstructured)
+				unstructuredMap, _, err := unstructured.NestedMap(unstructuredJob.Object)
+				if err != nil {
+					pkg.Logger.Error("[crd] [watch]cannot create unstructured map", zap.Error(err))
+					continue
+				}
+				job := &v1alpha1.FlinkJob{}
 
-			err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
-			if err != nil {
-				pkg.Logger.Error("cannot convert unstructured to structured", zap.Error(err))
-				continue
-			}
-
-			ch <- rxgo.Item{
-				V: &FlinkJobCrdEvent{
-					EventType: event.Type,
-					Job:       job,
-				},
-			}
-			switch event.Type {
-			case watch.Bookmark:
-			case watch.Modified:
-				//pkg.Logger.Info("[crd] [watch] flink job modified", zap.String("jobName", job.GetName()))
-				crd.repsert(job)
-			case watch.Added:
-				//pkg.Logger.Info("[crd] [watch] new flink job created")
-				crd.repsert(job)
-			case watch.Deleted:
-				crd.remove(job.UID)
+				err = runtime.DefaultUnstructuredConverter.FromUnstructured(unstructuredMap, job)
+				if err != nil {
+					pkg.Logger.Error("[crd] [watch]cannot convert unstructured to structured", zap.Error(err))
+					continue
+				}
+				if job.Namespace != namespace {
+					continue
+				}
 
+				go func() {
+					ch <- FlinkJobCrdEvent{
+						EventType: event.Type,
+						Job:       job,
+					}
+				}()
+				pkg.Logger.Debug("[crd] [watch] change in", zap.String("name", job.Name), zap.String("operation", string(event.Type)))
+				switch event.Type {
+				case watch.Bookmark:
+				case watch.Modified:
+					//pkg.Logger.Info("[crd] [watch] flink job modified", zap.String("jobName", job.GetName()))
+					crd.repsert(job)
+				case watch.Added:
+					//pkg.Logger.Info("[crd] [watch] new flink job created")
+					crd.repsert(job)
+				case watch.Deleted:
+					crd.remove(job.UID)
+				}
 			}
+			defer watcher.Stop()
+			pkg.Logger.Warn("[crd] [watch] Watcher stopped, restarting...")
 		}
-
 	}()
 
-	return rxgo.FromChannel(ch)
 }

internal/jar/jar.go

@@ -1,7 +1,10 @@
 package jar
 
 import (
+	"crypto/rand"
+	"encoding/hex"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
 	"os"
@@ -9,19 +12,43 @@ import (
 
 	"flink-kube-operator/pkg"
 
+	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/google/go-containerregistry/pkg/name"
+	"github.com/google/go-containerregistry/pkg/v1/remote"
+	"github.com/google/go-containerregistry/pkg/v1/types"
 	api "github.com/logi-camp/go-flink-client"
-	gonanoid "github.com/matoous/go-nanoid/v2"
 	"go.uber.org/zap"
 )
 
 type JarFile struct {
-	uri      string
-	filePath string
+	ref               string
+	filePath          string
+	digest            string
+	basicAuthUsername *string
+	basicAuthPassword *string
+	auth              authn.Authenticator
 }
 
-func NewJarFile(URI string) (*JarFile, error) {
+// NewJarFile creates a JarFile from an OCI reference, pulling the artifact.
+// auth is the OCI authenticator (from pull secret or env vars). Use authn.Anonymous for public registries.
+func NewJarFile(ref string, auth authn.Authenticator) (*JarFile, error) {
 	jarFile := &JarFile{
-		uri: URI,
+		ref:  ref,
+		auth: auth,
+	}
+	err := jarFile.Pull()
+	if err != nil {
+		return nil, err
+	}
+	return jarFile, nil
+}
+
+// NewJarFileFromURI creates a JarFile from an HTTP URI (deprecated path, kept for backward compat).
+func NewJarFileFromURI(uri string, basicAuthUsername *string, basicAuthPassword *string) (*JarFile, error) {
+	jarFile := &JarFile{
+		ref:               uri,
+		basicAuthUsername: basicAuthUsername,
+		basicAuthPassword: basicAuthPassword,
 	}
 	err := jarFile.Download()
 	if err != nil {
@@ -31,10 +58,10 @@ func NewJarFile(URI string) (*JarFile, error) {
 }
 
 func (jarFile *JarFile) Upload(flinkClient *api.Client) (fileName string, err error) {
-
 	resp, err := flinkClient.UploadJar(jarFile.filePath)
 	if err != nil {
-		pkg.Logger.Error("[main] error uploading jar", zap.Error(err))
+		pkg.Logger.Error("[jar] error uploading jar", zap.Error(err))
+		return "", err
 	}
 	filePathParts := strings.Split(resp.FileName, "/")
 	fileName = filePathParts[len(filePathParts)-1]
@@ -45,8 +72,82 @@ func (jarFile *JarFile) Upload(flinkClient *api.Client) (fileName string, err er
 	return
 }
 
+func (jarFile *JarFile) Digest() string {
+	return jarFile.digest
+}
+
+func (jarFile *JarFile) Pull() error {
+	randBytes := make([]byte, 16)
+	rand.Read(randBytes)
+	fileName := hex.EncodeToString(randBytes)
+	jarFile.filePath = "/tmp/" + fileName + ".jar"
+
+	ref, err := name.ParseReference(jarFile.ref)
+	if err != nil {
+		return fmt.Errorf("parsing OCI reference %q: %w", jarFile.ref, err)
+	}
+
+	img, err := remote.Image(ref, remote.WithAuth(jarFile.auth))
+	if err != nil {
+		return fmt.Errorf("fetching OCI image %q: %w", jarFile.ref, err)
+	}
+
+	d, err := img.Digest()
+	if err != nil {
+		return fmt.Errorf("getting image digest: %w", err)
+	}
+	jarFile.digest = d.String()
+
+	layers, err := img.Layers()
+	if err != nil {
+		return fmt.Errorf("getting image layers: %w", err)
+	}
+	if len(layers) == 0 {
+		return fmt.Errorf("OCI image %q has no layers", jarFile.ref)
+	}
+
+	layer := layers[0]
+	mediaType, err := layer.MediaType()
+	if err != nil {
+		return fmt.Errorf("getting layer media type: %w", err)
+	}
+	if mediaType != types.OCILayer && mediaType != types.OCIUncompressedLayer && mediaType != types.DockerLayer && mediaType != types.DockerUncompressedLayer && string(mediaType) != "application/java-archive" {
+		pkg.Logger.Warn("[jar] unexpected layer media type, expected application/java-archive or standard OCI layer",
+			zap.String("got", string(mediaType)),
+			zap.String("ref", jarFile.ref),
+		)
+	}
+
+	blob, err := layer.Compressed()
+	if err != nil {
+		return fmt.Errorf("reading layer blob: %w", err)
+	}
+	defer blob.Close()
+
+	out, err := os.Create(jarFile.filePath)
+	if err != nil {
+		return fmt.Errorf("creating temp file: %w", err)
+	}
+	defer out.Close()
+
+	_, err = io.Copy(out, blob)
+	if err != nil {
+		jarFile.delete()
+		return fmt.Errorf("writing layer to disk: %w", err)
+	}
+
+	pkg.Logger.Info("[jar] pulled OCI artifact",
+		zap.String("ref", jarFile.ref),
+		zap.String("digest", jarFile.digest),
+		zap.String("path", jarFile.filePath),
+	)
+	return nil
+}
+
 func (jarFile *JarFile) Download() error {
-	fileName, _ := gonanoid.New()
+	randBytes := make([]byte, 16)
+	rand.Read(randBytes)
+	fileName := hex.EncodeToString(randBytes)
 	jarFile.filePath = "/tmp/" + fileName + ".jar"
 	out, err := os.Create(jarFile.filePath)
 	if err != nil {
@@ -54,9 +155,27 @@ func (jarFile *JarFile) Download() error {
 	}
 
 	defer out.Close()
-	resp, err := http.Get(jarFile.uri)
-	if err != nil || resp.StatusCode > 299 {
+
+	var resp *http.Response
+	if jarFile.basicAuthPassword != nil && jarFile.basicAuthUsername != nil {
+		req, err := http.NewRequest("GET", jarFile.ref, nil)
+		if err != nil {
+			jarFile.delete()
+			return err
+		}
+		req.SetBasicAuth(*jarFile.basicAuthUsername, *jarFile.basicAuthPassword)
+		resp, err = http.DefaultClient.Do(req)
+	} else {
+		resp, err = http.Get(jarFile.ref)
+	}
+	if err != nil {
 		jarFile.delete()
+		pkg.Logger.Error("error in downloading jar", zap.Error(err))
+		return err
+	}
+	if resp.StatusCode > 299 {
+		err = fmt.Errorf("download failed: status %s", resp.Status)
+		pkg.Logger.Error("error in downloading jar", zap.Error(err))
 		return err
 	}
 

internal/managed_job/cycle.go

@@ -10,19 +10,12 @@ import (
 )
 
 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
 	}
 
-	if job.def.Status.JobStatus == v1alpha1.JobStatusFinished && job.def.Status.LifeCycleStatus == v1alpha1.LifeCycleStatusGracefullyPaused {
-		job.Run(true)
-		return
-	}
-
 	if job.def.Status.JobStatus == v1alpha1.JobStatusRunning {
 		if (job.def.Spec.SavepointInterval.Duration != 0) && ((job.def.Status.LastSavepointDate == nil) || time.Now().Add(-job.def.Spec.SavepointInterval.Duration).After(*job.def.Status.LastSavepointDate)) {
 			if job.def.Status.SavepointTriggerId == nil {
@@ -31,7 +24,8 @@ func (job *ManagedJob) Cycle() {
 				job.trackSavepoint()
 			}
 		}
-		if job.def.Status.RunningJarURI != nil && job.def.Spec.JarURI != *job.def.Status.RunningJarURI {
+
+		if job.def.Spec.NeedsUpgrade(job.def.Status) {
 			job.upgrade()
 		}
 
@@ -47,14 +41,6 @@ func (job *ManagedJob) Cycle() {
 		job.crd.SetJobStatus(job.def.UID, job.def.Status)
 		return
 	}
-	// if job.def.Status.JobStatus == v1alpha1.JobStatusFailed && job.def.Status.LastSavepointPath != nil {
-	// 	//job.restore()
-	// 	return
-	// }
-	// if job.def.Status.JobStatus == v1alpha1.JobStatusFailed && job.def.Status.LastSavepointPath == nil {
-	// 	//job.restore()
-	// 	return
-	// }
 
-	pkg.Logger.Warn("[managed-job] [cycle]", zap.String("unhanded job status", string(job.def.Status.JobStatus)))
+	pkg.Logger.Warn("[managed-job] [cycle] unhandled job status", zap.String("name", job.def.Name), zap.String("status", string(job.def.Status.JobStatus)), zap.String("namespace", job.def.Namespace))
 }

internal/managed_job/jar.go

@@ -4,41 +4,90 @@ import (
 	"flink-kube-operator/internal/jar"
 	"flink-kube-operator/pkg"
 
+	"github.com/google/go-containerregistry/pkg/authn"
+	"github.com/google/go-containerregistry/pkg/name"
 	"go.uber.org/zap"
 )
 
-// upload jar file and set the jarId for later usages
+// upload pulls/downloads the jar and uploads it to Flink, then patches the CRD status.
 func (job *ManagedJob) upload() error {
-	jarFile, err := jar.NewJarFile(job.def.Spec.JarURI)
+	spec := job.def.Spec
+	var jarFile *jar.JarFile
+	var err error
+
+	if spec.IsOCISource() {
+		auth, authErr := job.resolveAuth()
+		if authErr != nil {
+			pkg.Logger.Error("[managed-job] [upload] auth resolution failed", zap.Error(authErr))
+			return authErr
+		}
+		jarFile, err = jar.NewJarFile(spec.EffectiveJarRef(), auth)
+	} else {
+		jarFile, err = jar.NewJarFileFromURI(spec.JarURI, spec.JarURIBasicAuthUsername, spec.JarURIBasicAuthPassword)
+	}
+
 	if err != nil {
-		pkg.Logger.Debug("[manage-job] [upload] error on download jar", zap.Error(err))
+		pkg.Logger.Debug("[managed-job] [upload] error on pull/download jar", zap.Error(err))
 		return err
 	}
+
 	jarId, err := jarFile.Upload(job.client)
 	if err != nil {
-		pkg.Logger.Debug("[manage-job] [upload] error on upload jar", zap.Error(err))
+		pkg.Logger.Debug("[managed-job] [upload] error on upload jar", zap.Error(err))
 		return err
 	}
-	pkg.Logger.Info("[manage-job] [upload] uploaded", zap.Any("upload-jar-resp", jarId))
+	pkg.Logger.Info("[managed-job] [upload] uploaded", zap.String("jarId", jarId))
 
-	job.def.Status.JarId = &jarId
-	job.crd.Patch(job.def.UID, map[string]interface{}{
+	patchData := map[string]interface{}{
 		"status": map[string]interface{}{
-			"jarId": job.def.Status.JarId,
+			"jarId": &jarId,
 		},
-	})
+	}
+
+	if spec.IsOCISource() {
+		digest := jarFile.Digest()
+		patchData["status"].(map[string]interface{})["runningJarDigest"] = &digest
+	}
+
+	job.crd.Patch(job.def.UID, patchData)
 	return nil
 }
 
+func (job *ManagedJob) resolveAuth() (authn.Authenticator, error) {
+	s := job.def.Spec
+	server := ""
+	ref := s.EffectiveJarRef()
+	if reg, err := extractRegistry(ref); err == nil {
+		server = reg
+	}
+	return job.crd.ResolveAuth(s.JarPullSecret, server)
+}
+
+func extractRegistry(ref string) (string, error) {
+	r, err := name.NewRegistry(ref)
+	if err != nil {
+		t, err2 := name.NewTag(ref)
+		if err2 != nil {
+			return "", err
+		}
+		return t.RegistryStr(), nil
+	}
+	return r.RegistryStr(), nil
+}
+
 func (job *ManagedJob) RemoveJar() {
 	if job.def.Status.JarId != nil {
 		err := job.client.DeleteJar(*job.def.Status.JarId)
-		pkg.Logger.Error("[managed-job] [jar]", zap.Error(err))
+		if err != nil {
+			pkg.Logger.Error("[managed-job] [jar] delete jar error", zap.Error(err))
+		}
 		err = job.crd.Patch(job.def.UID, map[string]interface{}{
 			"status": map[string]interface{}{
 				"jarId": nil,
 			},
 		})
-		pkg.Logger.Error("[managed-job] [jar]", zap.Error(err))
+		if err != nil {
+			pkg.Logger.Error("[managed-job] [jar] patch jarId error", zap.Error(err))
+		}
 	}
 }

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"
@@ -50,8 +51,11 @@ func NewManager(client *api.Client, crdInstance *crd.Crd) Manager {
 	go func() {
 		for event := range crd.FinalizerChannel {
 			manager := mgr.GetJob(event)
-			manager.Stop()
-			delete(mgr.managedJobs, event)
+			if manager != nil {
+				err := manager.Stop()
+				pkg.Logger.Info("[finalizer]", zap.Error(err))
+				delete(mgr.managedJobs, event)
+			}
 		}
 	}()
 
@@ -68,11 +72,12 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 			},
 		})
 	}
-	//pkg.Logger.Debug("[manager] [cycle] overviews", zap.Any("overviews", jobsOverviews))
+	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() {
-		// pkg.Logger.Debug("mgr.processingJobsIds", zap.Any("processingJobIds", mgr.processingJobsIds))
 		if lo.Contains(mgr.processingJobsIds, uid) {
 			pkg.Logger.Warn("[manager] already in process", zap.Any("uid", uid))
 			continue
@@ -80,11 +85,32 @@ 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
-		managedJob, ok := mgr.managedJobs[uid]
-		if ok {
+		managedJob, jobFound := mgr.managedJobs[uid]
+		if jobFound {
 			managedJob.Update(def)
 		} else {
 			// Add job to manager managed job
@@ -94,14 +120,14 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 
 		}
 
-		jobManagerJobOverview, ok := lo.Find(jobManagerJobOverviews.Jobs, func(job api.JobOverview) bool {
+		jobManagerJobOverview, jobFound := lo.Find(jobManagerJobOverviews.Jobs, func(job api.JobOverview) bool {
 			jobId := managedJob.GetJobId()
 			if jobId != nil {
 				return job.ID == *jobId
 			}
 			return false
 		})
-		if ok {
+		if jobFound {
 			// pkg.Logger.Debug("[manager] read status from flink", zap.String("name", jobManagerJobOverview.Name), zap.String("state", jobManagerJobOverview.State))
 			patchStatusObj := map[string]interface{}{
 				"jobStatus": v1alpha1.JobStatus(jobManagerJobOverview.State),
@@ -114,6 +140,16 @@ func (mgr *Manager) cycle(client *api.Client, crdInstance *crd.Crd) {
 			crdInstance.Patch(uid, map[string]interface{}{
 				"status": patchStatusObj,
 			})
+		} else {
+			// TODO handle job not found status
+			// patchStatusObj := map[string]interface{}{
+			// 	"jobStatus":       "",
+			// 	"lifeCycleStatus": string(v1alpha1.LifeCycleStatusFailed),
+			// }
+
+			// crdInstance.Patch(uid, map[string]interface{}{
+			// 	"status": patchStatusObj,
+			// })
 		}
 
 		managedJob.Cycle()
@@ -129,6 +165,7 @@ func (mgr *Manager) GetJobs() map[types.UID]ManagedJob {
 	return mgr.managedJobs
 }
 
-func (mgr *Manager) GetJob(id types.UID) ManagedJob {
-	return mgr.managedJobs[id]
+func (mgr *Manager) GetJob(id types.UID) *ManagedJob {
+	job := mgr.managedJobs[id]
+	return &job
 }

internal/managed_job/pause.go

@@ -33,7 +33,7 @@ func (job *ManagedJob) Pause() error {
 		if savepointPath != "" {
 			job.def.Status.LastSavepointPath = &savepointPath
 			job.def.Status.PauseSavepointTriggerId = nil
-			job.def.Status.JobStatus = ""
+			job.def.Status.JobStatus = "FINISHED"
 			job.def.Status.LastSavepointPath = &savepointPath
 			lastSavepointDate := time.Now()
 			job.def.Status.LastSavepointDate = &lastSavepointDate

internal/managed_job/remove.go

@@ -1,5 +1,11 @@
 package managed_job
 
+import "errors"
+
 func (job *ManagedJob) Stop() error {
-	return job.client.StopJob(*job.def.Status.JobId)
+	if job.def.Status.JobId != nil {
+		return job.client.StopJob(*job.def.Status.JobId)
+	} else {
+		return errors.New("job Id not found")
+	}
 }

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))
@@ -83,17 +83,21 @@ func (job *ManagedJob) Run(restoreMode bool) error {
 	// job.def.Status.JobId = &runJarResp.JobId
 	// job.def.Status.JobStatus = v1alpha1.JobStatusCreating
 	// job.def.Status.Error = nil
+	statusPatch := map[string]interface{}{
+		"jobId":                             jobId,
+		"jobStatus":                         v1alpha1.JobStatusCreating,
+		"lifeCycleStatus":                   v1alpha1.LifeCycleStatusRestoring,
+		"lastRestoredSavepointDate":         job.def.Status.LastSavepointDate,
+		"restoredCount":                     job.def.Status.RestoredCount + 1,
+		"lastRestoredSavepointRestoredDate": time.Now().Format(time.RFC3339),
+		"error":                             nil,
+	}
+
+	refKey, refVal := job.def.Spec.RunningRefPatchData()
+	statusPatch[refKey] = refVal
+
 	job.crd.Patch(job.def.UID, map[string]interface{}{
-		"status": map[string]interface{}{
-			"jobId":                             jobId,
-			"runningJarURI":                     job.def.Spec.JarURI,
-			"jobStatus":                         v1alpha1.JobStatusCreating,
-			"lifeCycleStatus":                   v1alpha1.LifeCycleStatusRestoring,
-			"lastRestoredSavepointDate":         job.def.Status.LastSavepointDate,
-			"restoredCount":                     job.def.Status.RestoredCount + 1,
-			"lastRestoredSavepointRestoredDate": time.Now().Format(time.RFC3339),
-			"error":                             nil,
-		},
+		"status": statusPatch,
 	})
 
 	return nil

internal/managed_job/savepoint.go

@@ -17,14 +17,14 @@ func (job ManagedJob) createSavepoint() error {
 		pkg.Logger.Debug("[managed-job] [savepoint] no job id")
 		return v1alpha1.ErrNoJobId
 	}
-	pkg.Logger.Info("[managed-job] [savepoint] creating savepoint", zap.String("interval", job.def.Spec.SavepointInterval.String()))
+	pkg.Logger.Info("[managed-job] [savepoint] creating savepoint", zap.String("name", job.def.GetName()), zap.String("interval", job.def.Spec.SavepointInterval.String()))
 	resp, err := job.client.SavePoints(*job.def.Status.JobId, os.Getenv("SAVEPOINT_PATH"), false)
 	if err != nil {
 		pkg.Logger.Error("[managed-job] [savepoint] error in creating savepoint", zap.Error(err))
 		return err
 	}
 	pkg.Logger.Debug("[managed-job] [savepoint] savepoint created successfully", zap.String("trigger-id", resp.RequestID))
-
+	job.def.Status.SavepointTriggerId = &resp.RequestID
 	job.crd.Patch(job.def.UID, map[string]interface{}{
 		"status": map[string]interface{}{
 			"savepointTriggerId": resp.RequestID,
@@ -89,3 +89,7 @@ func (job ManagedJob) TriggerSavepoint() error {
 	err = job.trackSavepoint()
 	return err
 }
+
+func (job ManagedJob) GetLastSavepointPath() *string {
+	return job.def.Status.LastSavepointPath
+}

internal/managed_job/upgrade.go

@@ -7,10 +7,13 @@ import (
 )
 
 func (job *ManagedJob) upgrade() {
-	pkg.Logger.Info("[managed-job] [upgrade] pausing... ",
+	effectiveRef := job.def.Spec.EffectiveJarRef()
+	prevRef := job.def.Status.RunningRef()
+
+	pkg.Logger.Info("[managed-job] [upgrade] pausing...",
 		zap.String("jobName", job.def.GetName()),
-		zap.String("currentJarURI", job.def.Spec.JarURI),
-		zap.String("prevJarURI", *job.def.Status.RunningJarURI),
+		zap.String("currentRef", effectiveRef),
+		zap.String("prevRef", prevRef),
 	)
 	job.def.Status.JarId = nil
 	job.crd.Patch(job.def.UID, map[string]interface{}{
@@ -23,11 +26,10 @@ func (job *ManagedJob) upgrade() {
 		pkg.Logger.Error("[managed-job] [upgrade] error in pausing", zap.Error(err))
 		return
 	}
-	pkg.Logger.Info("[managed-job] [upgrade] restoring... ",
+	pkg.Logger.Info("[managed-job] [upgrade] restoring...",
 		zap.String("jobName", job.def.GetName()),
-		zap.String("currentJarURI", job.def.Spec.JarURI),
-		zap.String("prevJarURI", *job.def.Status.RunningJarURI),
-		zap.Error(err),
+		zap.String("currentRef", effectiveRef),
+		zap.String("prevRef", prevRef),
 	)
 
 	err = job.Run(true)

internal/rest/controller/crd.go

@@ -14,6 +14,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 
 	"github.com/danielgtaylor/huma/v2"
 	"go.uber.org/zap"
@@ -128,7 +129,14 @@ func TriggerSavepoint(ctx context.Context, req *JobTriggerSavepointReq) (*JobTri
 }
 
 func DownloadSavepoint(ctx context.Context, req *types.SavepointDownloadReq) (*huma.StreamResponse, error) {
-	folderPath := req.SavepointPath // Change this to your folder path
+	mgr := managed_job.GetManager()
+	job := mgr.GetJob(k8sTypes.UID(req.JobUID))
+	lastSavepointPath := job.GetLastSavepointPath()
+
+	if lastSavepointPath == nil {
+		return nil, huma.Error404NotFound("there is no savepoint path is registered for the job")
+	}
+	folderPath := strings.TrimLeft(*lastSavepointPath, "file:") // Change this to your folder path
 
 	// Create a temporary zip file
 	zipFilePath, err := filepath.Abs("./savepoint.zip")

internal/rest/types/savepoint.go

@@ -1,7 +1,7 @@
 package types
 
 type SavepointDownloadReq struct {
-	SavepointPath string `query:"savepoint-path"`
+	JobUID string `query:"jobUID"`
 }
 
 type SavepointDownloadResp struct {

pkg/logger.go

@@ -2,6 +2,8 @@ package pkg
 
 import (
 	"context"
+	"os"
+	"strconv"
 
 	"github.com/mattn/go-colorable"
 	"go.uber.org/zap"
@@ -82,8 +84,10 @@ func OverrideLoggerConfig(config LoggerConfig) {
 	Logger = createOrUpdateInstance(config)
 }
 
+var level, err = strconv.Atoi(os.Getenv("LOG_LEVEL"))
+
 var Logger = GetLogger(context.Background(), LoggerConfig{
-	Level:           zap.DebugLevel,
+	Level:           zapcore.Level(level),
 	Filename:        "./tmp/error.log",
 	MaxSize:         100,
 	MaxAge:          90,

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"