diff --git a/CHANGELOG.md b/CHANGELOG.md
index d6c700bc..c176b375 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- BREAKING: Add required CLI argument and env var to set the image repository used to construct final product image names: `IMAGE_REPOSITORY` (`--image-repository`), eg. `oci.example.org/my/namespace` ([#684]).
+
 ### Fixed
 
 - Application template merging should not ignore "*Overrides" fields ([#680]).
@@ -22,6 +26,7 @@ All notable changes to this project will be documented in this file.
 [#674]: https://github.com/stackabletech/spark-k8s-operator/pull/674
 [#679]: https://github.com/stackabletech/spark-k8s-operator/pull/679
 [#680]: https://github.com/stackabletech/spark-k8s-operator/pull/680
+[#684]: https://github.com/stackabletech/spark-k8s-operator/pull/684
 
 ## [26.3.0] - 2026-03-16
 
diff --git a/Cargo.lock b/Cargo.lock
index 81925f25..6af9820b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1517,7 +1517,7 @@ dependencies = [
 [[package]]
 name = "k8s-version"
 version = "0.1.3"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "darling",
  "regex",
@@ -2880,7 +2880,7 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
 [[package]]
 name = "stackable-certs"
 version = "0.4.0"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "const-oid",
  "ecdsa",
@@ -2903,8 +2903,8 @@ dependencies = [
 
 [[package]]
 name = "stackable-operator"
-version = "0.110.1"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+version = "0.111.1"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "base64",
  "clap",
@@ -2945,7 +2945,7 @@ dependencies = [
 [[package]]
 name = "stackable-operator-derive"
 version = "0.3.1"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "darling",
  "proc-macro2",
@@ -2956,7 +2956,7 @@ dependencies = [
 [[package]]
 name = "stackable-shared"
 version = "0.1.0"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "jiff",
  "k8s-openapi",
@@ -2998,7 +2998,7 @@ dependencies = [
 [[package]]
 name = "stackable-telemetry"
 version = "0.6.3"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "axum",
  "clap",
@@ -3021,9 +3021,10 @@ dependencies = [
 
 [[package]]
 name = "stackable-versioned"
-version = "0.9.0"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+version = "0.10.0"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
+ "kube",
  "schemars",
  "serde",
  "serde_json",
@@ -3034,8 +3035,8 @@ dependencies = [
 
 [[package]]
 name = "stackable-versioned-macros"
-version = "0.9.0"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+version = "0.10.0"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "convert_case",
  "convert_case_extras",
@@ -3053,7 +3054,7 @@ dependencies = [
 [[package]]
 name = "stackable-webhook"
 version = "0.9.1"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#96f42571ea185a3cd76fedde351fcabbeefcae16"
+source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#7a5f0c3fbcd091340214a23f0607fcd4b4fcc152"
 dependencies = [
  "arc-swap",
  "async-trait",
diff --git a/Cargo.nix b/Cargo.nix
index 675635de..b3bbcbf3 100644
--- a/Cargo.nix
+++ b/Cargo.nix
@@ -4822,7 +4822,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         libName = "k8s_version";
@@ -9470,7 +9470,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         libName = "stackable_certs";
@@ -9568,12 +9568,12 @@ rec {
       };
       "stackable-operator" = rec {
         crateName = "stackable-operator";
-        version = "0.110.1";
+        version = "0.111.1";
         edition = "2024";
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         libName = "stackable_operator";
@@ -9753,7 +9753,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         procMacro = true;
@@ -9788,7 +9788,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         libName = "stackable_shared";
@@ -9979,7 +9979,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         libName = "stackable_telemetry";
@@ -10084,12 +10084,12 @@ rec {
       };
       "stackable-versioned" = rec {
         crateName = "stackable-versioned";
-        version = "0.9.0";
+        version = "0.10.0";
         edition = "2024";
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         libName = "stackable_versioned";
@@ -10097,6 +10097,12 @@ rec {
           "Stackable GmbH <info@stackable.de>"
         ];
         dependencies = [
+          {
+            name = "kube";
+            packageId = "kube";
+            usesDefaultFeatures = false;
+            features = [ "client" "jsonpatch" "runtime" "derive" "admission" "rustls-tls" "ring" ];
+          }
           {
             name = "schemars";
             packageId = "schemars";
@@ -10128,12 +10134,12 @@ rec {
       };
       "stackable-versioned-macros" = rec {
         crateName = "stackable-versioned-macros";
-        version = "0.9.0";
+        version = "0.10.0";
         edition = "2024";
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         procMacro = true;
@@ -10201,7 +10207,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "96f42571ea185a3cd76fedde351fcabbeefcae16";
+          rev = "7a5f0c3fbcd091340214a23f0607fcd4b4fcc152";
           sha256 = "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by";
         };
         libName = "stackable_webhook";
diff --git a/Cargo.toml b/Cargo.toml
index 274d1352..8e405087 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -11,7 +11,7 @@ repository = "https://github.com/stackabletech/spark-k8s-operator"
 
 [workspace.dependencies]
 product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.8.0" }
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.110.1", features = ["webhook"] }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.111.1", features = ["webhook"] }
 
 anyhow = "1.0"
 built = { version = "0.8", features = ["chrono", "git2"] }
diff --git a/crate-hashes.json b/crate-hashes.json
index e19b553d..a6396ca0 100644
--- a/crate-hashes.json
+++ b/crate-hashes.json
@@ -1,12 +1,12 @@
 {
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#k8s-version@0.1.3": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-certs@0.4.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-operator-derive@0.3.1": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-operator@0.110.1": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-shared@0.1.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-telemetry@0.6.3": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-versioned-macros@0.9.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-versioned@0.9.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.110.1#stackable-webhook@0.9.1": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#k8s-version@0.1.3": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-certs@0.4.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-operator-derive@0.3.1": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-operator@0.111.1": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-shared@0.1.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-telemetry@0.6.3": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-versioned-macros@0.10.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-versioned@0.10.0": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
+  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.111.1#stackable-webhook@0.9.1": "0d58yvxvy8hbai12bjhcyvh4zw182j5dsfyqja4k2xc1vzjy29by",
   "git+https://github.com/stackabletech/product-config.git?tag=0.8.0#product-config@0.8.0": "1dz70kapm2wdqcr7ndyjji0lhsl98bsq95gnb2lw487wf6yr7987"
 }
\ No newline at end of file
diff --git a/extra/crds.yaml b/extra/crds.yaml
index 5f832882..04dc601e 100644
--- a/extra/crds.yaml
+++ b/extra/crds.yaml
@@ -2155,8 +2155,9 @@ spec:
                 properties:
                   custom:
                     description: |-
-                      Overwrite the docker image.
-                      Specify the full docker image name, e.g. `oci.stackable.tech/sdp/superset:1.4.1-stackable2.1.0`
+                      Provide a custom container image.
+
+                      Specify the full container image name, e.g. `oci.example.tech/namespace/superset:1.4.1-my-tag`
                     type: string
                   productVersion:
                     description: Version of the product, e.g. `1.4.1`.
@@ -2183,14 +2184,20 @@ spec:
                     nullable: true
                     type: array
                   repo:
-                    description: Name of the docker repo, e.g. `oci.stackable.tech/sdp`
+                    description: |-
+                      The repository on the container image registry where the container image is located, e.g.
+                      `oci.example.com/namespace`.
+
+                      If not specified, the operator will use the image registry provided via the operator
+                      environment options.
                     nullable: true
                     type: string
                   stackableVersion:
                     description: |-
                       Stackable version of the product, e.g. `23.4`, `23.4.1` or `0.0.0-dev`.
-                      If not specified, the operator will use its own version, e.g. `23.4.1`.
-                      When using a nightly operator or a pr version, it will use the nightly `0.0.0-dev` image.
+
+                      If not specified, the operator will use its own version, e.g. `23.4.1`. When using a nightly
+                      operator or a PR version, it will use the nightly `0.0.0-dev` image.
                     nullable: true
                     type: string
                 type: object
@@ -2289,8 +2296,9 @@ spec:
                 properties:
                   custom:
                     description: |-
-                      Overwrite the docker image.
-                      Specify the full docker image name, e.g. `oci.stackable.tech/sdp/superset:1.4.1-stackable2.1.0`
+                      Provide a custom container image.
+
+                      Specify the full container image name, e.g. `oci.example.tech/namespace/superset:1.4.1-my-tag`
                     type: string
                   productVersion:
                     description: Version of the product, e.g. `1.4.1`.
@@ -2317,14 +2325,20 @@ spec:
                     nullable: true
                     type: array
                   repo:
-                    description: Name of the docker repo, e.g. `oci.stackable.tech/sdp`
+                    description: |-
+                      The repository on the container image registry where the container image is located, e.g.
+                      `oci.example.com/namespace`.
+
+                      If not specified, the operator will use the image registry provided via the operator
+                      environment options.
                     nullable: true
                     type: string
                   stackableVersion:
                     description: |-
                       Stackable version of the product, e.g. `23.4`, `23.4.1` or `0.0.0-dev`.
-                      If not specified, the operator will use its own version, e.g. `23.4.1`.
-                      When using a nightly operator or a pr version, it will use the nightly `0.0.0-dev` image.
+
+                      If not specified, the operator will use its own version, e.g. `23.4.1`. When using a nightly
+                      operator or a PR version, it will use the nightly `0.0.0-dev` image.
                     nullable: true
                     type: string
                 type: object
@@ -4138,8 +4152,9 @@ spec:
                 properties:
                   custom:
                     description: |-
-                      Overwrite the docker image.
-                      Specify the full docker image name, e.g. `oci.stackable.tech/sdp/superset:1.4.1-stackable2.1.0`
+                      Provide a custom container image.
+
+                      Specify the full container image name, e.g. `oci.example.tech/namespace/superset:1.4.1-my-tag`
                     type: string
                   productVersion:
                     description: Version of the product, e.g. `1.4.1`.
@@ -4166,14 +4181,20 @@ spec:
                     nullable: true
                     type: array
                   repo:
-                    description: Name of the docker repo, e.g. `oci.stackable.tech/sdp`
+                    description: |-
+                      The repository on the container image registry where the container image is located, e.g.
+                      `oci.example.com/namespace`.
+
+                      If not specified, the operator will use the image registry provided via the operator
+                      environment options.
                     nullable: true
                     type: string
                   stackableVersion:
                     description: |-
                       Stackable version of the product, e.g. `23.4`, `23.4.1` or `0.0.0-dev`.
-                      If not specified, the operator will use its own version, e.g. `23.4.1`.
-                      When using a nightly operator or a pr version, it will use the nightly `0.0.0-dev` image.
+
+                      If not specified, the operator will use its own version, e.g. `23.4.1`. When using a nightly
+                      operator or a PR version, it will use the nightly `0.0.0-dev` image.
                     nullable: true
                     type: string
                 type: object
@@ -6758,8 +6779,9 @@ spec:
                 properties:
                   custom:
                     description: |-
-                      Overwrite the docker image.
-                      Specify the full docker image name, e.g. `oci.stackable.tech/sdp/superset:1.4.1-stackable2.1.0`
+                      Provide a custom container image.
+
+                      Specify the full container image name, e.g. `oci.example.tech/namespace/superset:1.4.1-my-tag`
                     type: string
                   productVersion:
                     description: Version of the product, e.g. `1.4.1`.
@@ -6786,14 +6808,20 @@ spec:
                     nullable: true
                     type: array
                   repo:
-                    description: Name of the docker repo, e.g. `oci.stackable.tech/sdp`
+                    description: |-
+                      The repository on the container image registry where the container image is located, e.g.
+                      `oci.example.com/namespace`.
+
+                      If not specified, the operator will use the image registry provided via the operator
+                      environment options.
                     nullable: true
                     type: string
                   stackableVersion:
                     description: |-
                       Stackable version of the product, e.g. `23.4`, `23.4.1` or `0.0.0-dev`.
-                      If not specified, the operator will use its own version, e.g. `23.4.1`.
-                      When using a nightly operator or a pr version, it will use the nightly `0.0.0-dev` image.
+
+                      If not specified, the operator will use its own version, e.g. `23.4.1`. When using a nightly
+                      operator or a PR version, it will use the nightly `0.0.0-dev` image.
                     nullable: true
                     type: string
                 type: object
diff --git a/rust/operator-binary/src/connect/controller.rs b/rust/operator-binary/src/connect/controller.rs
index 922f3f49..28f2732e 100644
--- a/rust/operator-binary/src/connect/controller.rs
+++ b/rust/operator-binary/src/connect/controller.rs
@@ -22,7 +22,7 @@ use super::crd::{CONNECT_APP_NAME, CONNECT_CONTROLLER_NAME, v1alpha1};
 use crate::{
     Ctx,
     connect::{common, crd::SparkConnectServerStatus, executor, s3, server, service},
-    crd::constants::{OPERATOR_NAME, SPARK_IMAGE_BASE_NAME},
+    crd::constants::{CONTAINER_IMAGE_BASE_NAME, OPERATOR_NAME},
 };
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
@@ -189,7 +189,11 @@ pub async fn reconcile(
     let resolved_product_image = scs
         .spec
         .image
-        .resolve(SPARK_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION)
+        .resolve(
+            CONTAINER_IMAGE_BASE_NAME,
+            &ctx.operator_environment.image_repository,
+            crate::built_info::PKG_VERSION,
+        )
         .context(ResolveProductImageSnafu)?;
 
     // Resolve any S3 connections early to fail fast if there are issues.
diff --git a/rust/operator-binary/src/connect/crd.rs b/rust/operator-binary/src/connect/crd.rs
index 39653482..b919f7c1 100644
--- a/rust/operator-binary/src/connect/crd.rs
+++ b/rust/operator-binary/src/connect/crd.rs
@@ -444,6 +444,7 @@ impl v1alpha1::ExecutorConfig {
 #[cfg(test)]
 mod tests {
     use indoc::indoc;
+    use stackable_operator::versioned::test_utils::RoundtripTestData;
 
     use super::*;
 
@@ -517,4 +518,48 @@ mod tests {
             serde_yaml::with::singleton_map_recursive::deserialize(deserializer)
                 .expect("Failed to deserialize SparkConnectServer with S3 connectors CR");
     }
+
+    impl RoundtripTestData for v1alpha1::SparkConnectServerSpec {
+        fn roundtrip_test_data() -> Vec<Self> {
+            stackable_operator::utils::yaml_from_str_singleton_map(indoc! {r#"
+              - image:
+                  productVersion: 4.1.1
+                  pullPolicy: IfNotPresent
+                connectors:
+                  s3buckets:
+                    - reference: ingest-bucket
+                  s3connection:
+                    reference: s3-connection
+                server:
+                  jvmArgumentOverrides:
+                    add:
+                      - -Dmy.custom.jvm.arg=customValue
+                  roleConfig:
+                    listenerClass: external-unstable
+                  config:
+                    logging:
+                      enableVectorAgent: false
+                      containers:
+                        spark:
+                          custom:
+                            configMap: spark-connect-log-config
+                  configOverrides:
+                    spark-defaults.conf:
+                      spark.jars.ivy: /tmp/ivy2
+                executor:
+                  configOverrides:
+                    spark-defaults.conf:
+                      spark.executor.instances: "1"
+                      spark.executor.memoryOverhead: 1m
+                  config:
+                    logging:
+                      enableVectorAgent: false
+                      containers:
+                        spark:
+                          custom:
+                            configMap: spark-connect-log-config
+            "#})
+            .expect("Failed to parse SparkConnectServerSpec YAML")
+        }
+    }
 }
diff --git a/rust/operator-binary/src/crd/constants.rs b/rust/operator-binary/src/crd/constants.rs
index 5d95ad3f..0b2aab32 100644
--- a/rust/operator-binary/src/crd/constants.rs
+++ b/rust/operator-binary/src/crd/constants.rs
@@ -79,7 +79,7 @@ pub const HISTORY_FULL_CONTROLLER_NAME: &str =
 pub const HISTORY_APP_NAME: &str = "spark-history";
 pub const HISTORY_ROLE_NAME: &str = "node";
 
-pub const SPARK_IMAGE_BASE_NAME: &str = "spark-k8s";
+pub const CONTAINER_IMAGE_BASE_NAME: &str = "spark-k8s";
 
 pub const SPARK_DEFAULTS_FILE_NAME: &str = "spark-defaults.conf";
 pub const SPARK_ENV_SH_FILE_NAME: &str = "spark-env.sh";
diff --git a/rust/operator-binary/src/crd/history.rs b/rust/operator-binary/src/crd/history.rs
index 24459bee..04e68b2b 100644
--- a/rust/operator-binary/src/crd/history.rs
+++ b/rust/operator-binary/src/crd/history.rs
@@ -475,7 +475,10 @@ fn default_listener_class() -> String {
 #[cfg(test)]
 mod test {
     use indoc::indoc;
-    use stackable_operator::{commons::tls_verification::TlsClientDetails, crd::s3};
+    use stackable_operator::{
+        commons::tls_verification::TlsClientDetails, crd::s3,
+        versioned::test_utils::RoundtripTestData,
+    };
 
     use super::*;
     use crate::crd::logdir::S3LogDir;
@@ -553,4 +556,61 @@ mod test {
             env_map.get("TEST_SPARK_HIST_VAR")
         );
     }
+
+    impl RoundtripTestData for v1alpha1::SparkHistoryServerSpec {
+        fn roundtrip_test_data() -> Vec<Self> {
+            stackable_operator::utils::yaml_from_str_singleton_map(indoc! {r#"
+              - image:
+                  productVersion: 3.5.8
+                  pullPolicy: IfNotPresent
+                logFileDirectory:
+                  s3:
+                    prefix: eventlogs/
+                    bucket:
+                      reference: spark-history-s3-bucket
+                sparkConf:
+                  test.sparkConf: "true"
+                nodes:
+                  roleConfig:
+                    listenerClass: external-unstable
+                  envOverrides:
+                    TEST_SPARK_HIST_VAR_ROLE: ROLE
+                    TEST_SPARK_HIST_VAR_FROM_RG: ROLE
+                  configOverrides:
+                    security.properties:
+                      test.securityProperties.role: role
+                      test.securityProperties.fromRg: role
+                    spark-env.sh:
+                      TEST_SPARK-ENV-SH_ROLE: ROLE
+                      TEST_SPARK-ENV-SH_FROM_RG: ROLE
+                  roleGroups:
+                    default:
+                      replicas: 1
+                      config:
+                        cleaner: true
+                      envOverrides:
+                        TEST_SPARK_HIST_VAR_FROM_RG: ROLEGROUP
+                        TEST_SPARK_HIST_VAR_RG: ROLEGROUP
+                      configOverrides:
+                        security.properties:
+                          test.securityProperties.fromRg: rolegroup
+                          test.securityProperties.rg: rolegroup
+                        spark-env.sh:
+                          TEST_SPARK-ENV-SH_FROM_RG: ROLEGROUP
+                          TEST_SPARK-ENV-SH_RG: ROLEGROUP
+                      podOverrides:
+                        spec:
+                          containers:
+                            - name: spark-history
+                              resources:
+                                requests:
+                                  cpu: 500m
+                                  memory: 512Mi
+                                limits:
+                                  cpu: 1500m
+                                  memory: 1024Mi
+            "#})
+            .expect("Failed to parse SparkHistoryServerSpec YAML")
+        }
+    }
 }
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
index ea9ba5d1..e2048dea 100644
--- a/rust/operator-binary/src/crd/mod.rs
+++ b/rust/operator-binary/src/crd/mod.rs
@@ -1251,6 +1251,7 @@ mod tests {
         },
         product_config_utils::ValidatedRoleConfigByPropertyKind,
         product_logging::spec::Logging,
+        versioned::test_utils::RoundtripTestData,
     };
 
     use super::*;
@@ -1718,7 +1719,7 @@ spec:
         let resolved_product_image = spark_application
             .spec
             .spark_image
-            .resolve("spark-k8s", "0.0.0-dev")
+            .resolve("spark-k8s", "oci.example.org", "0.0.0-dev")
             .expect("test: resolved product image is always valid");
 
         let product_config =
@@ -1821,4 +1822,154 @@ spec:
 
         assert_eq!(got, expected);
     }
+
+    impl RoundtripTestData for v1alpha1::SparkApplicationSpec {
+        fn roundtrip_test_data() -> Vec<Self> {
+            stackable_operator::utils::yaml_from_str_singleton_map(indoc! {r#"
+              - sparkImage:
+                  productVersion: 3.5.8
+                  pullPolicy: IfNotPresent
+                image: my-registry.example.com/my-spark-deps:1.0.0
+                mode: cluster
+                mainClass: org.apache.spark.examples.SparkPi
+                mainApplicationFile: /stackable/spark/examples/jars/spark-examples.jar
+                vectorAggregatorConfigMapName: vector-aggregator-discovery
+                args:
+                  - "--verbose"
+                  - "--input"
+                  - "/data/in"
+                sparkConf:
+                  spark.kubernetes.file.upload.path: s3a://my-bucket
+                s3connection:
+                  reference: spark-data-s3-connection
+                logFileDirectory:
+                  s3:
+                    prefix: eventlogs/
+                    bucket:
+                      reference: spark-history-s3-bucket
+                env:
+                  - name: TEST_SPARK_VAR_0
+                    value: ORIGINAL
+                  - name: TEST_SPARK_VAR_1
+                    value: DONOTREPLACE
+                job:
+                  envOverrides:
+                    TEST_SPARK_VAR_0: REPLACED
+                  configOverrides:
+                    security.properties:
+                      test.job.securityProperties: test
+                    spark-env.sh:
+                      TEST_JOB_SPARK-ENV-SH: TEST
+                  podOverrides:
+                    spec:
+                      serviceAccountName: override-sa
+                      containers:
+                        - name: spark-submit
+                          resources:
+                            requests:
+                              cpu: 500m
+                              memory: 512Mi
+                            limits:
+                              cpu: 1500m
+                              memory: 1024Mi
+                  jvmArgumentOverrides:
+                    add:
+                      - -Dhttps.proxyHost=proxy.my.corp
+                      - -Dhttps.proxyPort=8080
+                    remove:
+                      - -Dunwanted=true
+                    removeRegex:
+                      - -Dnoisy=.*
+                  config:
+                    retryOnFailureCount: 2
+                    requestedSecretLifetime: 7d
+                    affinity:
+                      nodeSelector:
+                        disktype: ssd
+                driver:
+                  envOverrides:
+                    TEST_SPARK_VAR_0: REPLACED
+                  configOverrides:
+                    security.properties:
+                      test.driver.securityProperties: test
+                    spark-env.sh:
+                      TEST_DRIVER_SPARK-ENV-SH: TEST
+                  podOverrides:
+                    spec:
+                      serviceAccountName: driver-sa
+                  jvmArgumentOverrides:
+                    add:
+                      - -XX:+UseG1GC
+                  config:
+                    resources:
+                      cpu:
+                        min: 300m
+                        max: 1200m
+                      memory:
+                        limit: 1024Mi
+                    requestedSecretLifetime: 7d
+                    affinity:
+                      nodeSelector:
+                        affinity-role: driver
+                executor:
+                  replicas: 1
+                  envOverrides:
+                    TEST_SPARK_VAR_0: REPLACED
+                  configOverrides:
+                    security.properties:
+                      test.executor.securityProperties: test
+                    spark-env.sh:
+                      TEST_EXECUTOR_SPARK-ENV-SH: TEST
+                  podOverrides:
+                    spec:
+                      serviceAccountName: executor-sa
+                  jvmArgumentOverrides:
+                    add:
+                      - -XX:+UseG1GC
+                  config:
+                    resources:
+                      cpu:
+                        min: 1250m
+                        max: 2000m
+                      memory:
+                        limit: 1024Mi
+                    requestedSecretLifetime: 7d
+                    affinity:
+                      nodeSelector:
+                        affinity-role: executor
+                deps:
+                  requirements:
+                    - tabulate==0.8.9
+                  packages:
+                    - org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.10.1
+                  repositories:
+                    - https://repo.maven.apache.org/maven2/
+                  excludePackages:
+                    - org.slf4j:slf4j-log4j12
+                volumes:
+                  - name: script
+                    configMap:
+                      name: write-to-iceberg
+                  - name: scratch
+                    emptyDir: {}
+              # Second entry covers the alternative enum variants:
+              # - logFileDirectory.customLogDirectory (vs s3)
+              # - s3connection.inline (vs reference)
+              - sparkImage:
+                  productVersion: 3.5.8
+                mode: cluster
+                mainApplicationFile: local:///stackable/app.jar
+                s3connection:
+                  inline:
+                    host: test-minio
+                    port: 9000
+                    accessStyle: Path
+                    credentials:
+                      secretClass: s3-credentials-class
+                logFileDirectory:
+                  customLogDirectory: /mnt/spark-events
+            "#})
+            .expect("Failed to parse SparkApplicationSpec YAML")
+        }
+    }
 }
diff --git a/rust/operator-binary/src/crd/template_spec.rs b/rust/operator-binary/src/crd/template_spec.rs
index 1a27ef58..6c50d2a8 100644
--- a/rust/operator-binary/src/crd/template_spec.rs
+++ b/rust/operator-binary/src/crd/template_spec.rs
@@ -353,6 +353,7 @@ async fn resolve(
 #[cfg(test)]
 mod tests {
     use indoc::indoc;
+    use stackable_operator::versioned::test_utils::RoundtripTestData;
 
     use super::*;
 
@@ -431,4 +432,15 @@ mod tests {
         ));
         assert!(options.template_names.is_empty());
     }
+
+    impl RoundtripTestData for v1alpha1::SparkApplicationTemplateSpec {
+        fn roundtrip_test_data() -> Vec<Self> {
+            // SparkApplicationTemplateSpec is just a wrapper around SparkApplicationSpec
+            let test_data = crate::crd::v1alpha1::SparkApplicationSpec::roundtrip_test_data();
+            test_data
+                .into_iter()
+                .map(|spec| v1alpha1::SparkApplicationTemplateSpec { spec })
+                .collect()
+        }
+    }
 }
diff --git a/rust/operator-binary/src/history/history_controller.rs b/rust/operator-binary/src/history/history_controller.rs
index a50d75b7..a5401fd8 100644
--- a/rust/operator-binary/src/history/history_controller.rs
+++ b/rust/operator-binary/src/history/history_controller.rs
@@ -57,10 +57,10 @@ use crate::{
     Ctx,
     crd::{
         constants::{
-            ACCESS_KEY_ID, HISTORY_APP_NAME, HISTORY_CONTROLLER_NAME, HISTORY_UI_PORT,
-            JVM_SECURITY_PROPERTIES_FILE, LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME,
-            MAX_SPARK_LOG_FILES_SIZE, METRICS_PORT, OPERATOR_NAME, SECRET_ACCESS_KEY,
-            SPARK_DEFAULTS_FILE_NAME, SPARK_ENV_SH_FILE_NAME, SPARK_IMAGE_BASE_NAME,
+            ACCESS_KEY_ID, CONTAINER_IMAGE_BASE_NAME, HISTORY_APP_NAME, HISTORY_CONTROLLER_NAME,
+            HISTORY_UI_PORT, JVM_SECURITY_PROPERTIES_FILE, LISTENER_VOLUME_DIR,
+            LISTENER_VOLUME_NAME, MAX_SPARK_LOG_FILES_SIZE, METRICS_PORT, OPERATOR_NAME,
+            SECRET_ACCESS_KEY, SPARK_DEFAULTS_FILE_NAME, SPARK_ENV_SH_FILE_NAME,
             STACKABLE_TRUST_STORE, VOLUME_MOUNT_NAME_CONFIG, VOLUME_MOUNT_NAME_LOG,
             VOLUME_MOUNT_NAME_LOG_CONFIG, VOLUME_MOUNT_PATH_CONFIG, VOLUME_MOUNT_PATH_LOG,
             VOLUME_MOUNT_PATH_LOG_CONFIG,
@@ -277,7 +277,11 @@ pub async fn reconcile(
     let resolved_product_image = shs
         .spec
         .image
-        .resolve(SPARK_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION)
+        .resolve(
+            CONTAINER_IMAGE_BASE_NAME,
+            &ctx.operator_environment.image_repository,
+            crate::built_info::PKG_VERSION,
+        )
         .context(ResolveProductImageSnafu)?;
     let log_dir = ResolvedLogDir::resolve(
         &shs.spec.log_file_directory,
diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs
index b24b8504..5f6024fb 100644
--- a/rust/operator-binary/src/main.rs
+++ b/rust/operator-binary/src/main.rs
@@ -11,7 +11,7 @@ use history::history_controller;
 use product_config::ProductConfigManager;
 use stackable_operator::{
     YamlSchema,
-    cli::{Command, RunArguments},
+    cli::{Command, OperatorEnvironmentOptions, RunArguments},
     eos::EndOfSupportChecker,
     k8s_openapi::api::{
         apps::v1::StatefulSet,
@@ -75,6 +75,7 @@ const PRODUCT_CONFIG_PATHS: [&str; 2] = [
 pub struct Ctx {
     pub client: stackable_operator::client::Client,
     pub product_config: ProductConfigManager,
+    pub operator_environment: OperatorEnvironmentOptions,
 }
 
 #[tokio::main]
@@ -144,7 +145,9 @@ async fn main() -> anyhow::Result<()> {
             let ctx = Ctx {
                 client: client.clone(),
                 product_config: product_config.load(&PRODUCT_CONFIG_PATHS)?,
+                operator_environment: operator_environment.clone(),
             };
+
             let spark_event_recorder = Arc::new(Recorder::new(
                 client.as_kube_client(),
                 Reporter {
@@ -232,6 +235,7 @@ async fn main() -> anyhow::Result<()> {
             let ctx = Ctx {
                 client: client.clone(),
                 product_config: product_config.load(&PRODUCT_CONFIG_PATHS)?,
+                operator_environment: operator_environment.clone(),
             };
             let history_event_recorder = Arc::new(Recorder::new(
                 client.as_kube_client(),
@@ -297,6 +301,7 @@ async fn main() -> anyhow::Result<()> {
             let ctx = Ctx {
                 client: client.clone(),
                 product_config: product_config.load(&PRODUCT_CONFIG_PATHS)?,
+                operator_environment,
             };
             let connect_event_recorder = Arc::new(Recorder::new(
                 client.as_kube_client(),
diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs
index 4769e22f..7bc059ea 100644
--- a/rust/operator-binary/src/spark_k8s_controller.rs
+++ b/rust/operator-binary/src/spark_k8s_controller.rs
@@ -306,7 +306,11 @@ pub async fn reconcile(
     let resolved_product_image = spark_application
         .spec
         .spark_image
-        .resolve(SPARK_IMAGE_BASE_NAME, crate::built_info::PKG_VERSION)
+        .resolve(
+            CONTAINER_IMAGE_BASE_NAME,
+            &ctx.operator_environment.image_repository,
+            crate::built_info::PKG_VERSION,
+        )
         .context(ResolveProductImageSnafu)?;
 
     let validated_product_config: ValidatedRoleConfigByPropertyKind = spark_application