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

@@ -3,3 +3,4 @@ db
 tmp
 *.jar
 .env
+/operator

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

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

@@ -24,7 +24,6 @@ spec:
               type: object
               required:
                 - key
-                - jarUri
               properties:
                 key:
                   type: string
@@ -37,12 +36,21 @@ spec:
                   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:
                   description: "Program arguments. Accepts a list of strings (--key, value, ...) or a map of key-value pairs."
                   x-kubernetes-preserve-unknown-fields: true
@@ -98,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

@@ -8,9 +8,8 @@ spec:
   name: "Word Count Example"
   entryClass: "tech.logicamp.logiline.FacilityEnrichment"
   parallelism: 1
-  jarUri: "https://git.logicamp.tech/api/packages/logiline/generic/facility-enrichment/1.0.0/facility-enrichment.jar"
+  jarRef: "lcr.logicamp.tech/logiline/facility-enrichment:1.0.0"
-  jarURIBasicAuthUsername: logiline-actrunner
+  jarPullSecret: "lcr-registry-creds"
-  jarURIBasicAuthPassword: daeweeb7ohpaiw3oojiCoong
   # args accepts two formats:
   #   Map (recommended):
   args:

go.mod

@@ -5,6 +5,7 @@ go 1.23.2
 require (
 	github.com/danielgtaylor/huma/v2 v2.27.0
 	github.com/gofiber/fiber/v2 v2.52.6
+	github.com/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
@@ -14,6 +15,10 @@ require (
 
 require (
 	github.com/andybalholm/brotli v1.1.1 // 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/go-openapi/jsonpointer v0.21.0 // indirect
@@ -26,13 +31,19 @@ require (
 	github.com/klauspost/compress v1.17.11 // 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/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/vbatts/tar-split v0.11.6 // indirect
 	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // 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 (
@@ -53,7 +64,7 @@ require (
 	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
@@ -61,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

go.sum

@@ -1,5 +1,7 @@
 github.com/andybalholm/brotli v1.1.1 h1:PR2pgnyFznKEugtsUo0xLdDop5SKXd5Qf5ysW+7XdTA=
 github.com/andybalholm/brotli v1.1.1/go.mod h1:05ib4cKhjx3OQYUY22hTVd34Bc8upXjOLL2rKwwZBoA=
+github.com/containerd/stargz-snapshotter/estargz v0.16.3 h1:7evrXtoh1mSbGj/pfRccTampEyKpjpOnS3CyiV1Ebr8=
+github.com/containerd/stargz-snapshotter/estargz v0.16.3/go.mod h1:uyr4BfYfOj3G9WBVE8cOlQmXAbPN9VEQpBBeJIuOipU=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/danielgtaylor/huma/v2 v2.27.0 h1:yxgJ8GqYqKeXw/EnQ4ZNc2NBpmn49AlhxL2+ksSXjUI=
 github.com/danielgtaylor/huma/v2 v2.27.0/go.mod h1:NbSFXRoOMh3BVmiLJQ9EbUpnPas7D9BeOxF/pZBAGa0=
@@ -7,6 +9,12 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/docker/cli v27.5.0+incompatible h1:aMphQkcGtpHixwwhAXJT1rrK/detk2JIvDaFkLctbGM=
+github.com/docker/cli v27.5.0+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
+github.com/docker/distribution v2.8.3+incompatible h1:AtKxIZ36LoNK51+Z6RpzLpddBirtxJnzDrHLEKxTAYk=
+github.com/docker/distribution v2.8.3+incompatible/go.mod h1:J2gT2udsDAN96Uj4KfcMRqY0/ypR+oyYUYmja8H+y+w=
+github.com/docker/docker-credential-helpers v0.8.2 h1:bX3YxiGzFP5sOXWc3bTPEXdEaZSeVMrFgOr3T+zrFAo=
+github.com/docker/docker-credential-helpers v0.8.2/go.mod h1:P3ci7E3lwkZg6XiHdRKft1KckHiO9a2rNtyFbZ/ry9M=
 github.com/emicklei/go-restful/v3 v3.11.0 h1:rAQeMHw1c7zTmncogyy8VvRZwtkmkZ4FxERmMY4rD+g=
 github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/evanphx/json-patch/v5 v5.9.0 h1:kcBlZQbplgElYIlo/n1hJbls2z/1awpXxpRi0/FOJfg=
@@ -38,6 +46,8 @@ github.com/google/gnostic-models v0.6.9/go.mod h1:CiWsm0s6BSQd1hRn8/QmxqB6BesYcb
 github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-containerregistry v0.20.3 h1:oNx7IdTI936V8CQRveCjaxOiegWwvM7kqkbXTpyiovI=
+github.com/google/go-containerregistry v0.20.3/go.mod h1:w00pIgBRDVUDFM6bq+Qx8lwNWK+cxgCuX1vd3PIBDNI=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0=
 github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
@@ -73,6 +83,8 @@ github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWE
 github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
 github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
 github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
+github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG+4E0Y=
+github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -84,6 +96,10 @@ github.com/onsi/ginkgo/v2 v2.19.0 h1:9Cnnf7UHo57Hy3k6/m5k3dRfGTMXGvxhHFvkDTCTpvA
 github.com/onsi/ginkgo/v2 v2.19.0/go.mod h1:rlwLi9PilAFJ8jCg9UE1QP6VBpd6/xj3SRC0d6TU0To=
 github.com/onsi/gomega v1.33.1 h1:dsYjIxxSR755MDmKVsaFQTE22ChNBcuuTWgkUDSubOk=
 github.com/onsi/gomega v1.33.1/go.mod h1:U4R44UsT+9eLIaYRB2a5qajjtQYn0hauxvRm16AVYg0=
+github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
+github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
+github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
+github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -96,12 +112,15 @@ github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/samber/lo v1.47.0 h1:z7RynLwP5nbyRscyvcD043DWYoOcYRv3mV8lBeqOCLc=
 github.com/samber/lo v1.47.0/go.mod h1:RmDH9Ct32Qy3gduHQuKJ3gW1fMHAnE/fAzQuf6He5cU=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
@@ -113,6 +132,8 @@ github.com/valyala/fasthttp v1.58.0 h1:GGB2dWxSbEprU9j0iMJHgdKYJVDyjrOwF9RE59PbR
 github.com/valyala/fasthttp v1.58.0/go.mod h1:SYXvHHaFp7QZHGKSHmoMipInhrI5StHrhDTYVEjK/Kw=
 github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVSA8=
 github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
+github.com/vbatts/tar-split v0.11.6 h1:4SjTW5+PU11n6fZenf2IPoV8/tz3AaYHMWjf23envGs=
+github.com/vbatts/tar-split v0.11.6/go.mod h1:dqKNtesIOr2j2Qv3W/cHjnvk9I8+G7oAkFDFN6TCBEI=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
 github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU=
@@ -140,12 +161,17 @@ golang.org/x/net v0.31.0 h1:68CPQngjLL0r2AlUKiSxtQFKvzRVbnzLwMUn5SzcLHo=
 golang.org/x/net v0.31.0/go.mod h1:P4fl1q7dY2hnZFxEk4pPSkDHF+QqjitcnDjUQyMM+pM=
 golang.org/x/oauth2 v0.21.0 h1:tsimM75w1tF/uws5rbeHzIWxEqElMehnc+iW793zsZs=
 golang.org/x/oauth2 v0.21.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
+golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
@@ -164,12 +190,15 @@ golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roY
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
 golang.org/x/tools v0.24.0 h1:J1shsA93PJUEVaUSaay7UXAyE8aimq3GW0pjlolpa24=
 golang.org/x/tools v0.24.0/go.mod h1:YhNqVBIfWHdzvTLs0d8LCuMhkKUgSUKldakyV7W/WDQ=
+golang.org/x/tools v0.29.0 h1:Xx0h3TtM9rzQpQuR4dKLrdglAmCEN5Oi+P74JdhdzXE=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
 google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+google.golang.org/protobuf v1.36.3 h1:82DV7MYdb8anAVi3qge1wSnMDrnKK7ebr+I0hHRN1BU=
+google.golang.org/protobuf v1.36.3/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

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/v1alpha1/flink_job.go

@@ -64,12 +64,18 @@ type FlinkJobSpec struct {
 	Name             string          `json:"name"`
 	FlinkCluster     string          `json:"flinkCluster"`
 	Parallelism      int             `json:"parallelism"`
-	JarURI                  string          `json:"jarUri"`
+	JarRef           string          `json:"jarRef,omitempty"`
-	JarURIBasicAuthUsername *string         `json:"jarURIBasicAuthUsername"`
+	JarPullSecret    string          `json:"jarPullSecret,omitempty"`
-	JarURIBasicAuthPassword *string         `json:"jarURIBasicAuthPassword"`
 	SavepointInterval metaV1.Duration `json:"savepointInterval"`
 	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 {
@@ -85,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
@@ -140,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/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/jar/jar.go

@@ -2,31 +2,51 @@ package jar
 
 import (
 	"crypto/rand"
-	"encoding/base64"
 	"encoding/hex"
 	"errors"
+	"fmt"
 	"io"
 	"net/http"
-	"net/http/cookiejar"
 	"os"
 	"strings"
 
 	"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"
 	"go.uber.org/zap"
 )
 
 type JarFile struct {
-	uri               string
+	ref               string
 	filePath          string
+	digest            string
 	basicAuthUsername *string
 	basicAuthPassword *string
+	auth              authn.Authenticator
 }
 
-func NewJarFile(URI string, basicAuthUsername *string, basicAuthPassword *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,
 	}
@@ -38,10 +58,10 @@ func NewJarFile(URI string, basicAuthUsername *string, basicAuthPassword *string
 }
 
 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]
@@ -52,6 +72,78 @@ 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 {
 	randBytes := make([]byte, 16)
 	rand.Read(randBytes)
@@ -66,31 +158,15 @@ func (jarFile *JarFile) Download() error {
 
 	var resp *http.Response
 	if jarFile.basicAuthPassword != nil && jarFile.basicAuthUsername != nil {
-
+		req, err := http.NewRequest("GET", jarFile.ref, nil)
-		basicAuth := func(username, password string) string {
-			auth := username + ":" + password
-			return base64.StdEncoding.EncodeToString([]byte(auth))
-		}
-
-		redirectPolicyFunc := func(req *http.Request, via []*http.Request) error {
-			req.Header.Add("Authorization", "Basic "+basicAuth(*jarFile.basicAuthUsername, *jarFile.basicAuthPassword))
-			return nil
-		}
-
-		client := &http.Client{
-			Jar:           &cookiejar.Jar{},
-			CheckRedirect: redirectPolicyFunc,
-		}
-
-		req, err := http.NewRequest("GET", jarFile.uri, nil)
 		if err != nil {
 			jarFile.delete()
 			return err
 		}
-		req.Header.Add("Authorization", "Basic "+basicAuth(*jarFile.basicAuthUsername, *jarFile.basicAuthPassword))
+		req.SetBasicAuth(*jarFile.basicAuthUsername, *jarFile.basicAuthPassword)
-		resp, err = client.Do(req)
+		resp, err = http.DefaultClient.Do(req)
 	} else {
-		resp, err = http.Get(jarFile.uri)
+		resp, err = http.Get(jarFile.ref)
 	}
 	if err != nil {
 		jarFile.delete()
@@ -98,9 +174,7 @@ func (jarFile *JarFile) Download() error {
 		return err
 	}
 	if resp.StatusCode > 299 {
-		respBody := []byte{}
+		err = fmt.Errorf("download failed: status %s", resp.Status)
-		resp.Body.Read(respBody)
-		err = errors.New(string(respBody) + " status:" + resp.Status)
 		pkg.Logger.Error("error in downloading jar", zap.Error(err))
 		return err
 	}

internal/managed_job/cycle.go

@@ -10,8 +10,6 @@ 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 == "" || job.def.Status.JobStatus == v1alpha1.JobStatusFinished) {
 		job.Run(false)
@@ -26,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()
 		}
 
@@ -42,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] 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, job.def.Spec.JarURIBasicAuthUsername, job.def.Spec.JarURIBasicAuthPassword)
+	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
+	patchData := map[string]interface{}{
-	job.crd.Patch(job.def.UID, 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/run.go

@@ -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
-	job.crd.Patch(job.def.UID, map[string]interface{}{
+	statusPatch := 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,
-		},
+	}
+
+	refKey, refVal := job.def.Spec.RunningRefPatchData()
+	statusPatch[refKey] = refVal
+
+	job.crd.Patch(job.def.UID, map[string]interface{}{
+		"status": statusPatch,
 	})
 
 	return nil

internal/managed_job/upgrade.go

@@ -7,10 +7,13 @@ import (
 )
 
 func (job *ManagedJob) upgrade() {
+	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("currentRef", effectiveRef),
-		zap.String("prevJarURI", *job.def.Status.RunningJarURI),
+		zap.String("prevRef", prevRef),
 	)
 	job.def.Status.JarId = nil
 	job.crd.Patch(job.def.UID, map[string]interface{}{
@@ -25,9 +28,8 @@ func (job *ManagedJob) upgrade() {
 	}
 	pkg.Logger.Info("[managed-job] [upgrade] restoring...",
 		zap.String("jobName", job.def.GetName()),
-		zap.String("currentJarURI", job.def.Spec.JarURI),
+		zap.String("currentRef", effectiveRef),
-		zap.String("prevJarURI", *job.def.Status.RunningJarURI),
+		zap.String("prevRef", prevRef),
-		zap.Error(err),
 	)
 
 	err = job.Run(true)