diff --git a/CHANGELOG.md b/CHANGELOG.md
index 07f5ffe6..25da2e1c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,9 +8,11 @@ All notable changes to this project will be documented in this file.
 
 - Added new fields to govern image pull policy ([#75])
 - New `nodeSelector` fields for both the driver and the excutors ([#76])
+- Mirror driver pod status to the corresponding spark application ([#77])
 
 [#75]: https://github.com/stackabletech/spark-k8s-operator/pull/75
 [#76]: https://github.com/stackabletech/spark-k8s-operator/pull/76
+[#77]: https://github.com/stackabletech/spark-k8s-operator/pull/77
 
 ### Changed
 
diff --git a/Cargo.lock b/Cargo.lock
index b0371163..755ac244 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -467,12 +467,6 @@ dependencies = [
  "percent-encoding",
 ]
 
-[[package]]
-name = "futures"
-version = "0.1.31"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3a471a38ef8ed83cd6e40aa59c1ffe17db6855c18e3604d9c4ed8c08ebc28678"
-
 [[package]]
 name = "futures"
 version = "0.3.21"
@@ -550,7 +544,6 @@ version = "0.3.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d8b7abd5d659d9b90c8cba917f6ec750a74e2dc23902ef9cd4cc8c8b22e6036a"
 dependencies = [
- "futures 0.1.31",
  "futures-channel",
  "futures-core",
  "futures-io",
@@ -852,7 +845,7 @@ dependencies = [
  "chrono",
  "dirs-next",
  "either",
- "futures 0.3.21",
+ "futures",
  "http",
  "http-body",
  "hyper",
@@ -918,7 +911,7 @@ dependencies = [
  "ahash",
  "backoff",
  "derivative",
- "futures 0.3.21",
+ "futures",
  "json-patch",
  "k8s-openapi",
  "kube-client",
@@ -1680,7 +1673,7 @@ dependencies = [
  "const_format",
  "derivative",
  "either",
- "futures 0.3.21",
+ "futures",
  "json-patch",
  "k8s-openapi",
  "kube",
@@ -1734,9 +1727,8 @@ dependencies = [
  "anyhow",
  "built",
  "clap",
- "fnv",
- "futures 0.3.21",
- "lazy_static",
+ "futures",
+ "serde",
  "serde_yaml",
  "snafu",
  "stackable-operator",
@@ -1744,6 +1736,7 @@ dependencies = [
  "strum",
  "tokio",
  "tracing",
+ "tracing-futures",
 ]
 
 [[package]]
@@ -2068,6 +2061,18 @@ dependencies = [
  "valuable",
 ]
 
+[[package]]
+name = "tracing-futures"
+version = "0.2.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2"
+dependencies = [
+ "futures",
+ "futures-task",
+ "pin-project",
+ "tracing",
+]
+
 [[package]]
 name = "tracing-log"
 version = "0.1.3"
diff --git a/deploy/config-spec/properties.yaml b/deploy/config-spec/properties.yaml
index fca1a2fa..c8bdb8de 100644
--- a/deploy/config-spec/properties.yaml
+++ b/deploy/config-spec/properties.yaml
@@ -1,276 +1,6 @@
+---
+# Not used. Kept for compatibilty with Dockerfile.
 version: 0.1.0
 spec:
-  units:
-    - unit: &unitPassword
-        name: "password"
-        regex: "^[a-zA-Z]\\w{5,20}$"
-    - unit: &unitDirectory
-        name: "directory"
-        regex: "^(.*)/?([^/]+)$"
-        examples:
-          - "/tmp/xyz"
-    - unit: &unitMemory
-        name: "memory"
-        regex: "(^\\p{N}+)(?:\\s*)((?:b|k|m|g|t|p|kb|mb|gb|tb|pb)\\b$)"
-        examples:
-          - "1024b"
-          - "1024kb"
-          - "500m"
-          - "1g"
-
-properties:
-  - property: &sparkWorkerCores
-      propertyNames:
-        - name: "SPARK_WORKER_CORES"
-          kind:
-            type: "file"
-            file: "spark-env.sh"
-      datatype:
-         type: "integer"
-      roles:
-        - name: "slave"
-          required: false
-      asOfVersion: "0.6.2"
-      description: "Total number of cores to allow Spark jobs to use on the machine (default: all available cores)"
-
-  - property: &sparkWorkerMemory
-      propertyNames:
-        - name: "SPARK_WORKER_MEMORY"
-          kind:
-            type: "file"
-            file: "spark-env.sh"
-      datatype:
-        type: "string"
-        unit: *unitMemory
-      roles:
-        - name: "slave"
-          required: false
-      asOfVersion: "0.6.2"
-      description: "Total amount of memory to allow Spark jobs to use on the machine, e.g. 1000M, 2G (default: total memory minus 1 GB); note that each job's individual memory is configured using SPARK_MEM."
-
-  - property: &sparkDaemonMemory
-      propertyNames:
-        - name: "SPARK_DAEMON_MEMORY"
-          kind:
-            type: "file"
-            file: "spark-env.sh"
-      datatype:
-        type: "string"
-        unit: *unitMemory
-      defaultValues:
-        - fromVersion: "0.6.2"
-          to_version: "1.4.1"
-          value: "512m"
-        - fromVersion: "1.5.0"
-          value: "1g"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "0.6.2"
-      description: "Memory to allocate to the Spark master and worker daemons themselves (default: 512m/1g)"
-
-  - property: &sparkEventLogEnabled
-      propertyNames:
-        - name: "spark.eventLog.enabled"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "bool"
-      defaultValues:
-        - fromVersion: "1.0.0"
-          value: "false"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "true"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "1.0.0"
-      description: "Whether to log Spark events, useful for reconstructing the Web UI after the application has finished."
-
-  - property: &sparkEventLogDir
-      propertyNames:
-        - name: "spark.eventLog.dir"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitDirectory
-      defaultValues:
-        - fromVersion: "1.0.0"
-          value: "/tmp/spark-events"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "/stackable/log"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-      asOfVersion: "1.0.0"
-      expandsTo:
-        - property: *sparkEventLogEnabled
-          value: "true"
-      description: "Base directory in which Spark events are logged, if spark.eventLog.enabled is true. Within this base directory, Spark creates a sub-directory for each application, and logs the events specific to the application in this directory. Users may want to set this to a unified location like an HDFS directory so history files can be read by the history server."
-
-  - property: &sparkHistoryLogDirectory
-      propertyNames:
-        - name: "spark.history.fs.logDirectory"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitDirectory
-      defaultValues:
-        - fromVersion: "1.1.0"
-          value: "/tmp/spark-events"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "/stackable/log"
-      roles:
-        - name: "history-server"
-          required: true
-      expandsTo:
-        - property: *sparkEventLogEnabled
-          value: "true"
-      asOfVersion: "1.1.0"
-      description: "For the filesystem history provider, the URL to the directory containing application event logs to load. This can be a local file://path, an HDFS path hdfs://namenode/shared/spark-logs or that of an alternative filesystem supported by the Hadoop APIs."
-
-  - property: &sparkHistoryStorePath
-      propertyNames:
-        - name: "spark.history.store.path"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitDirectory
-      roles:
-        - name: "history-server"
-          required: false
-      asOfVersion: "2.3.0"
-      description: "Local directory where to cache application history data. If set, the history server will store application data on disk instead of keeping it in memory. The data written to disk will be re-used in the event of a history server restart."
-
-  - property: &sparkAuthenticate
-      propertyNames:
-        - name: "spark.authenticate"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "bool"
-      defaultValues:
-        - fromVersion: "1.0.0"
-          value: "false"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "true"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "1.0.0"
-      description: "Whether Spark authenticates its internal connections."
-
-  - property: &sparkAuthenticateSecret
-      propertyNames:
-        - name: "spark.authenticate.secret"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitPassword
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "secret"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "1.0.0"
-      expandsTo:
-        - property: *sparkAuthenticate
-          value: "true"
-      description: "The secret key used in the authentication. SPARK_AUTHENTICATE must be set to true."
-
-  - property: &sparkPortMaxRetries
-      propertyNames:
-        - name: "spark.port.maxRetries"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "integer"
-      defaultValues:
-        - fromVersion: "1.1.1"
-          value: "16"
-      recommendedValues:
-        - fromVersion: "1.1.1"
-          value: "0"
-      roles:
-        - name: "master"
-          required: true
-        - name: "slave"
-          required: true
-        - name: "history-server"
-          required: true
-      asOfVersion: "1.1.1"
-      description: "Maximum number of retries when binding to a port before giving up. When a port is given a specific value (non 0), each subsequent retry will increment the port used in the previous attempt by 1 before retrying. This essentially allows it to try a range of ports from the start port specified to port + maxRetries."
-
-  - property: &sparkNoDaemonize
-      propertyNames:
-        - name: "SPARK_NO_DAEMONIZE"
-          kind:
-            type: "env"
-      datatype:
-        type: "bool"
-      recommendedValues:
-        - fromVersion: "2.0.0"
-          value: "true"
-      roles:
-        - name: "master"
-          required: true
-        - name: "slave"
-          required: true
-        - name: "history-server"
-          required: true
-      asOfVersion: "2.0.0"
-      description: "Run spark processes in foreground if true. Useful for systemd (default: false)"
-
-  - property: &sparkConfDir
-      propertyNames:
-        - name: "SPARK_CONF_DIR"
-          kind:
-            type: "env"
-      datatype:
-        type: "string"
-      recommendedValues:
-        - fromVersion: "1.1.1"
-          value: "/stackable/config"
-      roles:
-        - name: "master"
-          required: true
-        - name: "slave"
-          required: true
-        - name: "history-server"
-          required: true
-      asOfVersion: "1.1.1"
-      description: "To specify a different configuration directory other than the default “SPARK_HOME/conf”, you can set SPARK_CONF_DIR. Spark will use the the configuration files (spark-defaults.conf, spark-env.sh, log4j.properties, etc) from this directory."
+  units: []
+properties: []
diff --git a/deploy/crd/sparkapplication.crd.yaml b/deploy/crd/sparkapplication.crd.yaml
index 6b7d1a1b..99b302ff 100644
--- a/deploy/crd/sparkapplication.crd.yaml
+++ b/deploy/crd/sparkapplication.crd.yaml
@@ -21,7 +21,6 @@ spec:
           description: "Auto-generated derived type for SparkApplicationSpec via `CustomResource`"
           properties:
             spec:
-              description: SparkApplicationStatus CommandStatus
               properties:
                 args:
                   items:
@@ -1403,16 +1402,10 @@ spec:
             status:
               nullable: true
               properties:
-                finishedAt:
-                  description: Time is a wrapper around time.Time which supports correct marshaling to YAML and JSON.  Wrappers are provided for many of the factory methods that the time package offers.
-                  format: date-time
-                  nullable: true
-                  type: string
-                startedAt:
-                  description: Time is a wrapper around time.Time which supports correct marshaling to YAML and JSON.  Wrappers are provided for many of the factory methods that the time package offers.
-                  format: date-time
-                  nullable: true
+                phase:
                   type: string
+              required:
+                - phase
               type: object
           required:
             - spec
diff --git a/deploy/helm/spark-k8s-operator/configs/properties.yaml b/deploy/helm/spark-k8s-operator/configs/properties.yaml
index fca1a2fa..c8bdb8de 100644
--- a/deploy/helm/spark-k8s-operator/configs/properties.yaml
+++ b/deploy/helm/spark-k8s-operator/configs/properties.yaml
@@ -1,276 +1,6 @@
+---
+# Not used. Kept for compatibilty with Dockerfile.
 version: 0.1.0
 spec:
-  units:
-    - unit: &unitPassword
-        name: "password"
-        regex: "^[a-zA-Z]\\w{5,20}$"
-    - unit: &unitDirectory
-        name: "directory"
-        regex: "^(.*)/?([^/]+)$"
-        examples:
-          - "/tmp/xyz"
-    - unit: &unitMemory
-        name: "memory"
-        regex: "(^\\p{N}+)(?:\\s*)((?:b|k|m|g|t|p|kb|mb|gb|tb|pb)\\b$)"
-        examples:
-          - "1024b"
-          - "1024kb"
-          - "500m"
-          - "1g"
-
-properties:
-  - property: &sparkWorkerCores
-      propertyNames:
-        - name: "SPARK_WORKER_CORES"
-          kind:
-            type: "file"
-            file: "spark-env.sh"
-      datatype:
-         type: "integer"
-      roles:
-        - name: "slave"
-          required: false
-      asOfVersion: "0.6.2"
-      description: "Total number of cores to allow Spark jobs to use on the machine (default: all available cores)"
-
-  - property: &sparkWorkerMemory
-      propertyNames:
-        - name: "SPARK_WORKER_MEMORY"
-          kind:
-            type: "file"
-            file: "spark-env.sh"
-      datatype:
-        type: "string"
-        unit: *unitMemory
-      roles:
-        - name: "slave"
-          required: false
-      asOfVersion: "0.6.2"
-      description: "Total amount of memory to allow Spark jobs to use on the machine, e.g. 1000M, 2G (default: total memory minus 1 GB); note that each job's individual memory is configured using SPARK_MEM."
-
-  - property: &sparkDaemonMemory
-      propertyNames:
-        - name: "SPARK_DAEMON_MEMORY"
-          kind:
-            type: "file"
-            file: "spark-env.sh"
-      datatype:
-        type: "string"
-        unit: *unitMemory
-      defaultValues:
-        - fromVersion: "0.6.2"
-          to_version: "1.4.1"
-          value: "512m"
-        - fromVersion: "1.5.0"
-          value: "1g"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "0.6.2"
-      description: "Memory to allocate to the Spark master and worker daemons themselves (default: 512m/1g)"
-
-  - property: &sparkEventLogEnabled
-      propertyNames:
-        - name: "spark.eventLog.enabled"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "bool"
-      defaultValues:
-        - fromVersion: "1.0.0"
-          value: "false"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "true"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "1.0.0"
-      description: "Whether to log Spark events, useful for reconstructing the Web UI after the application has finished."
-
-  - property: &sparkEventLogDir
-      propertyNames:
-        - name: "spark.eventLog.dir"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitDirectory
-      defaultValues:
-        - fromVersion: "1.0.0"
-          value: "/tmp/spark-events"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "/stackable/log"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-      asOfVersion: "1.0.0"
-      expandsTo:
-        - property: *sparkEventLogEnabled
-          value: "true"
-      description: "Base directory in which Spark events are logged, if spark.eventLog.enabled is true. Within this base directory, Spark creates a sub-directory for each application, and logs the events specific to the application in this directory. Users may want to set this to a unified location like an HDFS directory so history files can be read by the history server."
-
-  - property: &sparkHistoryLogDirectory
-      propertyNames:
-        - name: "spark.history.fs.logDirectory"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitDirectory
-      defaultValues:
-        - fromVersion: "1.1.0"
-          value: "/tmp/spark-events"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "/stackable/log"
-      roles:
-        - name: "history-server"
-          required: true
-      expandsTo:
-        - property: *sparkEventLogEnabled
-          value: "true"
-      asOfVersion: "1.1.0"
-      description: "For the filesystem history provider, the URL to the directory containing application event logs to load. This can be a local file://path, an HDFS path hdfs://namenode/shared/spark-logs or that of an alternative filesystem supported by the Hadoop APIs."
-
-  - property: &sparkHistoryStorePath
-      propertyNames:
-        - name: "spark.history.store.path"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitDirectory
-      roles:
-        - name: "history-server"
-          required: false
-      asOfVersion: "2.3.0"
-      description: "Local directory where to cache application history data. If set, the history server will store application data on disk instead of keeping it in memory. The data written to disk will be re-used in the event of a history server restart."
-
-  - property: &sparkAuthenticate
-      propertyNames:
-        - name: "spark.authenticate"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "bool"
-      defaultValues:
-        - fromVersion: "1.0.0"
-          value: "false"
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "true"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "1.0.0"
-      description: "Whether Spark authenticates its internal connections."
-
-  - property: &sparkAuthenticateSecret
-      propertyNames:
-        - name: "spark.authenticate.secret"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "string"
-        unit: *unitPassword
-      recommendedValues:
-        - fromVersion: "1.0.0"
-          value: "secret"
-      roles:
-        - name: "master"
-          required: false
-        - name: "slave"
-          required: false
-        - name: "history-server"
-          required: false
-      asOfVersion: "1.0.0"
-      expandsTo:
-        - property: *sparkAuthenticate
-          value: "true"
-      description: "The secret key used in the authentication. SPARK_AUTHENTICATE must be set to true."
-
-  - property: &sparkPortMaxRetries
-      propertyNames:
-        - name: "spark.port.maxRetries"
-          kind:
-            type: "file"
-            file: "spark-defaults.conf"
-      datatype:
-        type: "integer"
-      defaultValues:
-        - fromVersion: "1.1.1"
-          value: "16"
-      recommendedValues:
-        - fromVersion: "1.1.1"
-          value: "0"
-      roles:
-        - name: "master"
-          required: true
-        - name: "slave"
-          required: true
-        - name: "history-server"
-          required: true
-      asOfVersion: "1.1.1"
-      description: "Maximum number of retries when binding to a port before giving up. When a port is given a specific value (non 0), each subsequent retry will increment the port used in the previous attempt by 1 before retrying. This essentially allows it to try a range of ports from the start port specified to port + maxRetries."
-
-  - property: &sparkNoDaemonize
-      propertyNames:
-        - name: "SPARK_NO_DAEMONIZE"
-          kind:
-            type: "env"
-      datatype:
-        type: "bool"
-      recommendedValues:
-        - fromVersion: "2.0.0"
-          value: "true"
-      roles:
-        - name: "master"
-          required: true
-        - name: "slave"
-          required: true
-        - name: "history-server"
-          required: true
-      asOfVersion: "2.0.0"
-      description: "Run spark processes in foreground if true. Useful for systemd (default: false)"
-
-  - property: &sparkConfDir
-      propertyNames:
-        - name: "SPARK_CONF_DIR"
-          kind:
-            type: "env"
-      datatype:
-        type: "string"
-      recommendedValues:
-        - fromVersion: "1.1.1"
-          value: "/stackable/config"
-      roles:
-        - name: "master"
-          required: true
-        - name: "slave"
-          required: true
-        - name: "history-server"
-          required: true
-      asOfVersion: "1.1.1"
-      description: "To specify a different configuration directory other than the default “SPARK_HOME/conf”, you can set SPARK_CONF_DIR. Spark will use the the configuration files (spark-defaults.conf, spark-env.sh, log4j.properties, etc) from this directory."
+  units: []
+properties: []
diff --git a/deploy/helm/spark-k8s-operator/crds/crds.yaml b/deploy/helm/spark-k8s-operator/crds/crds.yaml
index 26cc7dbf..0c31374d 100644
--- a/deploy/helm/spark-k8s-operator/crds/crds.yaml
+++ b/deploy/helm/spark-k8s-operator/crds/crds.yaml
@@ -23,7 +23,6 @@ spec:
           description: "Auto-generated derived type for SparkApplicationSpec via `CustomResource`"
           properties:
             spec:
-              description: SparkApplicationStatus CommandStatus
               properties:
                 args:
                   items:
@@ -1405,16 +1404,10 @@ spec:
             status:
               nullable: true
               properties:
-                finishedAt:
-                  description: Time is a wrapper around time.Time which supports correct marshaling to YAML and JSON.  Wrappers are provided for many of the factory methods that the time package offers.
-                  format: date-time
-                  nullable: true
-                  type: string
-                startedAt:
-                  description: Time is a wrapper around time.Time which supports correct marshaling to YAML and JSON.  Wrappers are provided for many of the factory methods that the time package offers.
-                  format: date-time
-                  nullable: true
+                phase:
                   type: string
+              required:
+                - phase
               type: object
           required:
             - spec
diff --git a/deploy/manifests/configmap.yaml b/deploy/manifests/configmap.yaml
index bd86b1c9..4f43785f 100644
--- a/deploy/manifests/configmap.yaml
+++ b/deploy/manifests/configmap.yaml
@@ -2,282 +2,12 @@
 apiVersion: v1
 data:
   properties.yaml: |
+    ---
+    # Not used. Kept for compatibilty with Dockerfile.
     version: 0.1.0
     spec:
-      units:
-        - unit: &unitPassword
-            name: "password"
-            regex: "^[a-zA-Z]\\w{5,20}$"
-        - unit: &unitDirectory
-            name: "directory"
-            regex: "^(.*)/?([^/]+)$"
-            examples:
-              - "/tmp/xyz"
-        - unit: &unitMemory
-            name: "memory"
-            regex: "(^\\p{N}+)(?:\\s*)((?:b|k|m|g|t|p|kb|mb|gb|tb|pb)\\b$)"
-            examples:
-              - "1024b"
-              - "1024kb"
-              - "500m"
-              - "1g"
-
-    properties:
-      - property: &sparkWorkerCores
-          propertyNames:
-            - name: "SPARK_WORKER_CORES"
-              kind:
-                type: "file"
-                file: "spark-env.sh"
-          datatype:
-             type: "integer"
-          roles:
-            - name: "slave"
-              required: false
-          asOfVersion: "0.6.2"
-          description: "Total number of cores to allow Spark jobs to use on the machine (default: all available cores)"
-
-      - property: &sparkWorkerMemory
-          propertyNames:
-            - name: "SPARK_WORKER_MEMORY"
-              kind:
-                type: "file"
-                file: "spark-env.sh"
-          datatype:
-            type: "string"
-            unit: *unitMemory
-          roles:
-            - name: "slave"
-              required: false
-          asOfVersion: "0.6.2"
-          description: "Total amount of memory to allow Spark jobs to use on the machine, e.g. 1000M, 2G (default: total memory minus 1 GB); note that each job's individual memory is configured using SPARK_MEM."
-
-      - property: &sparkDaemonMemory
-          propertyNames:
-            - name: "SPARK_DAEMON_MEMORY"
-              kind:
-                type: "file"
-                file: "spark-env.sh"
-          datatype:
-            type: "string"
-            unit: *unitMemory
-          defaultValues:
-            - fromVersion: "0.6.2"
-              to_version: "1.4.1"
-              value: "512m"
-            - fromVersion: "1.5.0"
-              value: "1g"
-          roles:
-            - name: "master"
-              required: false
-            - name: "slave"
-              required: false
-            - name: "history-server"
-              required: false
-          asOfVersion: "0.6.2"
-          description: "Memory to allocate to the Spark master and worker daemons themselves (default: 512m/1g)"
-
-      - property: &sparkEventLogEnabled
-          propertyNames:
-            - name: "spark.eventLog.enabled"
-              kind:
-                type: "file"
-                file: "spark-defaults.conf"
-          datatype:
-            type: "bool"
-          defaultValues:
-            - fromVersion: "1.0.0"
-              value: "false"
-          recommendedValues:
-            - fromVersion: "1.0.0"
-              value: "true"
-          roles:
-            - name: "master"
-              required: false
-            - name: "slave"
-              required: false
-            - name: "history-server"
-              required: false
-          asOfVersion: "1.0.0"
-          description: "Whether to log Spark events, useful for reconstructing the Web UI after the application has finished."
-
-      - property: &sparkEventLogDir
-          propertyNames:
-            - name: "spark.eventLog.dir"
-              kind:
-                type: "file"
-                file: "spark-defaults.conf"
-          datatype:
-            type: "string"
-            unit: *unitDirectory
-          defaultValues:
-            - fromVersion: "1.0.0"
-              value: "/tmp/spark-events"
-          recommendedValues:
-            - fromVersion: "1.0.0"
-              value: "/stackable/log"
-          roles:
-            - name: "master"
-              required: false
-            - name: "slave"
-              required: false
-          asOfVersion: "1.0.0"
-          expandsTo:
-            - property: *sparkEventLogEnabled
-              value: "true"
-          description: "Base directory in which Spark events are logged, if spark.eventLog.enabled is true. Within this base directory, Spark creates a sub-directory for each application, and logs the events specific to the application in this directory. Users may want to set this to a unified location like an HDFS directory so history files can be read by the history server."
-
-      - property: &sparkHistoryLogDirectory
-          propertyNames:
-            - name: "spark.history.fs.logDirectory"
-              kind:
-                type: "file"
-                file: "spark-defaults.conf"
-          datatype:
-            type: "string"
-            unit: *unitDirectory
-          defaultValues:
-            - fromVersion: "1.1.0"
-              value: "/tmp/spark-events"
-          recommendedValues:
-            - fromVersion: "1.0.0"
-              value: "/stackable/log"
-          roles:
-            - name: "history-server"
-              required: true
-          expandsTo:
-            - property: *sparkEventLogEnabled
-              value: "true"
-          asOfVersion: "1.1.0"
-          description: "For the filesystem history provider, the URL to the directory containing application event logs to load. This can be a local file://path, an HDFS path hdfs://namenode/shared/spark-logs or that of an alternative filesystem supported by the Hadoop APIs."
-
-      - property: &sparkHistoryStorePath
-          propertyNames:
-            - name: "spark.history.store.path"
-              kind:
-                type: "file"
-                file: "spark-defaults.conf"
-          datatype:
-            type: "string"
-            unit: *unitDirectory
-          roles:
-            - name: "history-server"
-              required: false
-          asOfVersion: "2.3.0"
-          description: "Local directory where to cache application history data. If set, the history server will store application data on disk instead of keeping it in memory. The data written to disk will be re-used in the event of a history server restart."
-
-      - property: &sparkAuthenticate
-          propertyNames:
-            - name: "spark.authenticate"
-              kind:
-                type: "file"
-                file: "spark-defaults.conf"
-          datatype:
-            type: "bool"
-          defaultValues:
-            - fromVersion: "1.0.0"
-              value: "false"
-          recommendedValues:
-            - fromVersion: "1.0.0"
-              value: "true"
-          roles:
-            - name: "master"
-              required: false
-            - name: "slave"
-              required: false
-            - name: "history-server"
-              required: false
-          asOfVersion: "1.0.0"
-          description: "Whether Spark authenticates its internal connections."
-
-      - property: &sparkAuthenticateSecret
-          propertyNames:
-            - name: "spark.authenticate.secret"
-              kind:
-                type: "file"
-                file: "spark-defaults.conf"
-          datatype:
-            type: "string"
-            unit: *unitPassword
-          recommendedValues:
-            - fromVersion: "1.0.0"
-              value: "secret"
-          roles:
-            - name: "master"
-              required: false
-            - name: "slave"
-              required: false
-            - name: "history-server"
-              required: false
-          asOfVersion: "1.0.0"
-          expandsTo:
-            - property: *sparkAuthenticate
-              value: "true"
-          description: "The secret key used in the authentication. SPARK_AUTHENTICATE must be set to true."
-
-      - property: &sparkPortMaxRetries
-          propertyNames:
-            - name: "spark.port.maxRetries"
-              kind:
-                type: "file"
-                file: "spark-defaults.conf"
-          datatype:
-            type: "integer"
-          defaultValues:
-            - fromVersion: "1.1.1"
-              value: "16"
-          recommendedValues:
-            - fromVersion: "1.1.1"
-              value: "0"
-          roles:
-            - name: "master"
-              required: true
-            - name: "slave"
-              required: true
-            - name: "history-server"
-              required: true
-          asOfVersion: "1.1.1"
-          description: "Maximum number of retries when binding to a port before giving up. When a port is given a specific value (non 0), each subsequent retry will increment the port used in the previous attempt by 1 before retrying. This essentially allows it to try a range of ports from the start port specified to port + maxRetries."
-
-      - property: &sparkNoDaemonize
-          propertyNames:
-            - name: "SPARK_NO_DAEMONIZE"
-              kind:
-                type: "env"
-          datatype:
-            type: "bool"
-          recommendedValues:
-            - fromVersion: "2.0.0"
-              value: "true"
-          roles:
-            - name: "master"
-              required: true
-            - name: "slave"
-              required: true
-            - name: "history-server"
-              required: true
-          asOfVersion: "2.0.0"
-          description: "Run spark processes in foreground if true. Useful for systemd (default: false)"
-
-      - property: &sparkConfDir
-          propertyNames:
-            - name: "SPARK_CONF_DIR"
-              kind:
-                type: "env"
-          datatype:
-            type: "string"
-          recommendedValues:
-            - fromVersion: "1.1.1"
-              value: "/stackable/config"
-          roles:
-            - name: "master"
-              required: true
-            - name: "slave"
-              required: true
-            - name: "history-server"
-              required: true
-          asOfVersion: "1.1.1"
-          description: "To specify a different configuration directory other than the default “SPARK_HOME/conf”, you can set SPARK_CONF_DIR. Spark will use the the configuration files (spark-defaults.conf, spark-env.sh, log4j.properties, etc) from this directory."
+      units: []
+    properties: []
 kind: ConfigMap
 metadata:
   name: spark-k8s-operator-configmap
diff --git a/deploy/manifests/crds.yaml b/deploy/manifests/crds.yaml
index 20d5d011..0540e287 100644
--- a/deploy/manifests/crds.yaml
+++ b/deploy/manifests/crds.yaml
@@ -24,7 +24,6 @@ spec:
           description: "Auto-generated derived type for SparkApplicationSpec via `CustomResource`"
           properties:
             spec:
-              description: SparkApplicationStatus CommandStatus
               properties:
                 args:
                   items:
@@ -1406,16 +1405,10 @@ spec:
             status:
               nullable: true
               properties:
-                finishedAt:
-                  description: Time is a wrapper around time.Time which supports correct marshaling to YAML and JSON.  Wrappers are provided for many of the factory methods that the time package offers.
-                  format: date-time
-                  nullable: true
-                  type: string
-                startedAt:
-                  description: Time is a wrapper around time.Time which supports correct marshaling to YAML and JSON.  Wrappers are provided for many of the factory methods that the time package offers.
-                  format: date-time
-                  nullable: true
+                phase:
                   type: string
+              required:
+                - phase
               type: object
           required:
             - spec
diff --git a/docs/modules/ROOT/pages/commandline_args.adoc b/docs/modules/ROOT/pages/commandline_args.adoc
index c3628eae..a790f307 100644
--- a/docs/modules/ROOT/pages/commandline_args.adoc
+++ b/docs/modules/ROOT/pages/commandline_args.adoc
@@ -1,17 +1,4 @@
 
-=== product-config
-
-*Default value*: `/etc/stackable/spark-k8s-operator/config-spec/properties.yaml`
-
-*Required*: false
-
-*Multiple values:* false
-
-[source]
-----
-stackable-spark-k8s-operator run --product-config /foo/bar/properties.yaml
-----
-
 === watch-namespace
 
 *Default value*: All namespaces
diff --git a/docs/modules/ROOT/pages/env_var_args.adoc b/docs/modules/ROOT/pages/env_var_args.adoc
index 6e33ddff..d283a70a 100644
--- a/docs/modules/ROOT/pages/env_var_args.adoc
+++ b/docs/modules/ROOT/pages/env_var_args.adoc
@@ -1,30 +1,4 @@
 
-=== PRODUCT_CONFIG
-
-*Default value*: `/etc/stackable/spark-k8s-operator/config-spec/properties.yaml`
-
-*Required*: false
-
-*Multiple values:* false
-
-[source]
-----
-export PRODUCT_CONFIG=/foo/bar/properties.yaml
-stackable-spark-k8s-operator run
-----
-
-or via docker:
-
-----
-docker run \
-    --name spark-k8s-operator \
-    --network host \
-    --env KUBECONFIG=/home/stackable/.kube/config \
-    --env PRODUCT_CONFIG=/my/product/config.yaml \
-    --mount type=bind,source="$HOME/.kube/config",target="/home/stackable/.kube/config" \
-    docker.stackable.tech/stackable/spark-k8s-operator:latest
-----
-
 === WATCH_NAMESPACE
 
 *Default value*: All namespaces
@@ -53,4 +27,3 @@ docker run \
 --mount type=bind,source="$HOME/.kube/config",target="/home/stackable/.kube/config" \
 docker.stackable.tech/stackable/spark-k8s-operator:latest
 ----
-
diff --git a/docs/modules/ROOT/pages/installation.adoc b/docs/modules/ROOT/pages/installation.adoc
index 2d68c955..d2bfb8f1 100644
--- a/docs/modules/ROOT/pages/installation.adoc
+++ b/docs/modules/ROOT/pages/installation.adoc
@@ -27,19 +27,19 @@ service. You are now ready to deploy Apache Spark in Kubernetes.
 
 == Building the operator from source
 
-This operator is written in Rust and is developed against a recent stable Rust release.
+To run it from your local machine - usually for development purposes - you need to install the required manifest files.
 
 [source,bash]
 ----
-cargo run -- crd | kubectl apply -f -
-cargo run -- run
+make renenerate-charts
+kubectl create -f deploy/manifests
 ----
 
-To run it from your local machine - usually for development purposes - you need to create a `ClusterRoleBinding` :
+Then, start the operator:
 
 [source,bash]
 ----
-kubectl create clusterrolebinding spark-role --clusterrole=spark-driver-edit-role  --serviceaccount=default:default
+cargo run -- run
 ----
 
 == Additional/Optional components
@@ -48,4 +48,4 @@ The above describes the installation of the operator alone and is sufficient for
 
 == Examples
 
-The examples provided with the operator code show different ways of combining these elements.
\ No newline at end of file
+The examples provided with the operator code show different ways of combining these elements.
diff --git a/docs/modules/ROOT/pages/rbac.adoc b/docs/modules/ROOT/pages/rbac.adoc
index 83d890d2..7abb691a 100644
--- a/docs/modules/ROOT/pages/rbac.adoc
+++ b/docs/modules/ROOT/pages/rbac.adoc
@@ -6,30 +6,6 @@ The https://spark.apache.org/docs/latest/running-on-kubernetes.html#rbac[Spark-K
 
 However, to add security, each `spark-submit` job launched by the spark-k8s operator will be assigned its own service account.
 
-When the spark-k8s operator is launched via helm, like this:
+When the spark-k8s operator is installed via helm, a cluster role named `spark-driver-edit-role` is created with pre-defined permissions.
 
-[source,bash]
-----
-helm install spark-k8s-operator stackable/spark-k8s-operator
-----
-
-the operator will take care of these steps automatically:
-
-- a cluster role will be created with pre-defined permissions
-- for each `SparkApplication` job, a service account will be created
-- for each `SparkApplication` job, a role binding will connect the cluster role with the service account
-
-If the operator is started outside the cluster, like this:
-
-[source,bash]
-----
-cargo run -- crd | kubectl apply -f -
-cargo run -- run
-----
-
-then the cluster-role has to be created assigned to the service account manually e.g.
-
-[source,bash]
-----
-kubectl create clusterrolebinding spark-role --clusterrole=spark-driver-edit-role  --serviceaccount=default:default
-----
\ No newline at end of file
+When a new Spark application is submitted, the operator creates a new service account with the same name as the application and binds this account to the cluster role `spark-driver-edit-role` created by helm.
diff --git a/docs/modules/ROOT/pages/usage.adoc b/docs/modules/ROOT/pages/usage.adoc
index 5822c6f6..569d9f5b 100644
--- a/docs/modules/ROOT/pages/usage.adoc
+++ b/docs/modules/ROOT/pages/usage.adoc
@@ -43,7 +43,7 @@ The following examples have the following `spec` fields in common:
 
 Job-specific settings are annotated below.
 
-=== Pyspark: externally located artifact and dataset 
+=== Pyspark: externally located artifact and dataset
 
 [source,yaml]
 ----
@@ -72,7 +72,7 @@ include::example$example-sparkapp-image.yaml[]
 <6> the name of the volume mount backed by a `PersistentVolumeClaim` that must be pre-existing
 <7> the path on the volume mount: this is referenced in the `sparkConf` section where the extra class path is defined for the driver and executors
 
-=== JVM (Scala): externally located artifact and dataset 
+=== JVM (Scala): externally located artifact and dataset
 
 [source,yaml]
 ----
@@ -283,4 +283,3 @@ Below are listed the CRD fields that can be defined by the user:
 |`spec.executor.nodeSelector`
 |A dictionary of labels to use for node selection when scheduling the executors N.B. this assumes there are no implicit node dependencies (e.g. `PVC`, `VolumeMount`) defined elsewhere.
 |===
-
diff --git a/rust/crd/src/lib.rs b/rust/crd/src/lib.rs
index 39d8c1ad..fbe7bf18 100644
--- a/rust/crd/src/lib.rs
+++ b/rust/crd/src/lib.rs
@@ -14,7 +14,6 @@ use std::collections::{BTreeMap, HashMap};
 
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, Snafu};
-use stackable_operator::k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
 use stackable_operator::kube::ResourceExt;
 use stackable_operator::labels;
 use stackable_operator::{
@@ -41,14 +40,20 @@ pub enum Error {
     #[snafu(display("application has no Spark image"))]
     NoSparkImage,
 }
-/// SparkApplicationStatus CommandStatus
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct SparkApplicationStatus {
+    pub phase: String,
+}
+
 #[derive(Clone, CustomResource, Debug, Default, Deserialize, JsonSchema, PartialEq, Serialize)]
 #[kube(
     group = "spark.stackable.tech",
     version = "v1alpha1",
     kind = "SparkApplication",
     shortname = "sc",
-    status = "CommandStatus",
+    status = "SparkApplicationStatus",
     namespaced,
     crates(
         kube_core = "stackable_operator::kube::core",
@@ -404,14 +409,6 @@ impl SparkApplication {
     }
 }
 
-#[derive(Clone, Default, Debug, Deserialize, JsonSchema, PartialEq, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct SparkApplicationStatus {
-    /// An opaque value that changes every time a discovery detail does
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub discovery_hash: Option<String>,
-}
-
 #[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct CommonConfig {
@@ -479,15 +476,6 @@ impl ExecutorConfig {
     }
 }
 
-#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize, JsonSchema)]
-#[serde(rename_all = "camelCase")]
-pub struct CommandStatus {
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub started_at: Option<Time>,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub finished_at: Option<Time>,
-}
-
 #[cfg(test)]
 mod tests {
     use crate::ImagePullPolicy;
diff --git a/rust/operator-binary/Cargo.toml b/rust/operator-binary/Cargo.toml
index 5805db38..03deb94f 100644
--- a/rust/operator-binary/Cargo.toml
+++ b/rust/operator-binary/Cargo.toml
@@ -11,15 +11,15 @@ version = "0.2.0-nightly"
 stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag="0.19.0" }
 stackable-spark-k8s-crd = { path = "../crd" }
 anyhow = "1.0"
-clap = "3.1"
-fnv = "1.0"
-futures = { version = "0.3", features = ["compat"] }
-lazy_static = "1.4"
+clap = { version = "3.1", features = ["derive"] }
+futures = "0.3"
+serde = "1.0"
 serde_yaml = "0.8"
 snafu = "0.7"
 strum = { version = "0.24.0", features = ["derive"] }
 tokio = { version = "1.18", features = ["macros", "rt-multi-thread"] }
 tracing = "0.1"
+tracing-futures = { version = "0.2", features = ["futures-03"] }
 
 [build-dependencies]
 built = { version =  "0.5", features = ["chrono", "git2"] }
diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs
index 1e2b4645..80bbd789 100644
--- a/rust/operator-binary/src/main.rs
+++ b/rust/operator-binary/src/main.rs
@@ -1,14 +1,18 @@
+mod pod_driver_controller;
 mod spark_k8s_controller;
 
 use clap::Parser;
-use futures::stream::StreamExt;
+use futures::StreamExt;
 use stackable_operator::cli::{Command, ProductOperatorRun};
 use stackable_operator::k8s_openapi::api::core::v1::ConfigMap;
+use stackable_operator::k8s_openapi::api::core::v1::Pod;
 use stackable_operator::kube::api::ListParams;
 use stackable_operator::kube::runtime::controller::{Context, Controller};
 use stackable_operator::kube::CustomResourceExt;
 use stackable_operator::logging::controller::report_controller_reconciled;
 use stackable_spark_k8s_crd::SparkApplication;
+use tracing::info_span;
+use tracing_futures::Instrument;
 
 mod built_info {
     include!(concat!(env!("OUT_DIR"), "/built.rs"));
@@ -27,7 +31,7 @@ async fn main() -> anyhow::Result<()> {
     match opts.cmd {
         Command::Crd => println!("{}", serde_yaml::to_string(&SparkApplication::crd())?,),
         Command::Run(ProductOperatorRun {
-            product_config,
+            product_config: _,
             watch_namespace,
             tracing_target,
         }) => {
@@ -45,16 +49,11 @@ async fn main() -> anyhow::Result<()> {
                 built_info::RUSTC_VERSION,
             );
 
-            let product_config = product_config.load(&[
-                "deploy/config-spec/properties.yaml",
-                "/etc/stackable/spark-k8s-operator/config-spec/properties.yaml",
-            ])?;
-
             let client =
                 stackable_operator::client::create_client(Some("spark.stackable.tech".to_string()))
                     .await?;
 
-            Controller::new(
+            let app_controller = Controller::new(
                 watch_namespace.get_api::<SparkApplication>(&client),
                 ListParams::default(),
             )
@@ -68,7 +67,6 @@ async fn main() -> anyhow::Result<()> {
                 spark_k8s_controller::error_policy,
                 Context::new(spark_k8s_controller::Ctx {
                     client: client.clone(),
-                    product_config,
                 }),
             )
             .map(|res| {
@@ -78,10 +76,38 @@ async fn main() -> anyhow::Result<()> {
                     &res,
                 )
             })
-            .collect::<()>()
-            .await;
+            .instrument(info_span!("app_controller"));
+
+            let pod_driver_controller = Controller::new(
+                watch_namespace.get_api::<Pod>(&client),
+                ListParams::default()
+                    .labels("app.kubernetes.io/managed-by=spark-k8s-operator,spark-role=driver"),
+            )
+            .owns(
+                watch_namespace.get_api::<Pod>(&client),
+                ListParams::default(),
+            )
+            .shutdown_on_signal()
+            .run(
+                pod_driver_controller::reconcile,
+                pod_driver_controller::error_policy,
+                Context::new(pod_driver_controller::Ctx {
+                    client: client.clone(),
+                }),
+            )
+            .map(|res| {
+                report_controller_reconciled(
+                    &client,
+                    "pod-driver.sparkapplications.stackable.tech",
+                    &res,
+                )
+            })
+            .instrument(info_span!("pod_driver_controller"));
+
+            futures::stream::select(app_controller, pod_driver_controller)
+                .collect::<()>()
+                .await;
         }
     }
-
     Ok(())
 }
diff --git a/rust/operator-binary/src/pod_driver_controller.rs b/rust/operator-binary/src/pod_driver_controller.rs
new file mode 100644
index 00000000..1910188b
--- /dev/null
+++ b/rust/operator-binary/src/pod_driver_controller.rs
@@ -0,0 +1,100 @@
+use stackable_operator::{
+    k8s_openapi::api::core::v1::Pod,
+    kube::runtime::controller::{Action, Context},
+};
+use stackable_spark_k8s_crd::{SparkApplication, SparkApplicationStatus};
+use std::sync::Arc;
+use std::time::Duration;
+
+use snafu::{OptionExt, ResultExt, Snafu};
+use stackable_operator::logging::controller::ReconcilerError;
+use strum::{EnumDiscriminants, IntoStaticStr};
+
+const LABEL_NAME_INSTANCE: &str = "app.kubernetes.io/instance";
+
+pub struct Ctx {
+    pub client: stackable_operator::client::Client,
+}
+
+#[derive(Snafu, Debug, EnumDiscriminants)]
+#[strum_discriminants(derive(IntoStaticStr))]
+#[allow(clippy::enum_variant_names)]
+pub enum Error {
+    #[snafu(display("Label [{LABEL_NAME_INSTANCE}] not found for pod name [{pod_name}]"))]
+    LabelInstaceNotFound { pod_name: String },
+    #[snafu(display("Failed to update status for application [{name}]"))]
+    ApplySparkApplicationStatus {
+        source: stackable_operator::error::Error,
+        name: String,
+    },
+    #[snafu(display("Pod name not found"))]
+    PodNameNotFound,
+    #[snafu(display("Status phase not found for pod [{pod_name}]"))]
+    PodStatusPhaseNotFound { pod_name: String },
+    #[snafu(display("Spark application [{name}] not found"))]
+    SparkApplicationNotFound {
+        source: stackable_operator::error::Error,
+        name: String,
+    },
+}
+
+type Result<T, E = Error> = std::result::Result<T, E>;
+
+impl ReconcilerError for Error {
+    fn category(&self) -> &'static str {
+        ErrorDiscriminants::from(self).into()
+    }
+}
+/// Updates the status of the SparkApplication that started the pod.
+pub async fn reconcile(pod: Arc<Pod>, ctx: Context<Ctx>) -> Result<Action> {
+    tracing::info!("Starting reconcile driver pod");
+
+    let pod_name = pod.metadata.name.as_ref().context(PodNameNotFoundSnafu)?;
+    let app_name = pod
+        .metadata
+        .labels
+        .as_ref()
+        .and_then(|l| l.get(&String::from(LABEL_NAME_INSTANCE)))
+        .context(LabelInstaceNotFoundSnafu {
+            pod_name: pod_name.clone(),
+        })?;
+    let phase = pod.status.as_ref().and_then(|s| s.phase.as_ref()).context(
+        PodStatusPhaseNotFoundSnafu {
+            pod_name: pod_name.clone(),
+        },
+    )?;
+
+    let app = ctx
+        .get_ref()
+        .client
+        .get::<SparkApplication>(
+            app_name.as_ref(),
+            pod.metadata.namespace.as_ref().map(|ns| ns.as_ref()),
+        )
+        .await
+        .context(SparkApplicationNotFoundSnafu {
+            name: app_name.clone(),
+        })?;
+
+    tracing::info!("Update spark application [{app_name}] status to [{phase}]");
+
+    ctx.get_ref()
+        .client
+        .apply_patch_status(
+            "pod-driver.sparkapplications.stackable.tech",
+            &app,
+            &SparkApplicationStatus {
+                phase: phase.clone(),
+            },
+        )
+        .await
+        .with_context(|_| ApplySparkApplicationStatusSnafu {
+            name: app_name.clone(),
+        })?;
+
+    Ok(Action::await_change())
+}
+
+pub fn error_policy(_error: &Error, _ctx: Context<Ctx>) -> Action {
+    Action::requeue(Duration::from_secs(5))
+}
diff --git a/rust/operator-binary/src/spark_k8s_controller.rs b/rust/operator-binary/src/spark_k8s_controller.rs
index 27c4d7f9..09e8058e 100644
--- a/rust/operator-binary/src/spark_k8s_controller.rs
+++ b/rust/operator-binary/src/spark_k8s_controller.rs
@@ -8,10 +8,8 @@ use stackable_operator::k8s_openapi::api::core::v1::{
 };
 use stackable_operator::k8s_openapi::api::rbac::v1::{ClusterRole, RoleBinding, RoleRef, Subject};
 use stackable_operator::k8s_openapi::Resource;
-use stackable_operator::kube::api::ObjectMeta;
 use stackable_operator::kube::runtime::controller::{Action, Context};
 use stackable_operator::logging::controller::ReconcilerError;
-use stackable_operator::product_config::ProductConfigManager;
 use stackable_spark_k8s_crd::constants::*;
 use stackable_spark_k8s_crd::SparkApplication;
 use std::collections::BTreeMap;
@@ -23,7 +21,6 @@ const SPARK_CLUSTER_ROLE: &str = "spark-driver-edit-role";
 
 pub struct Ctx {
     pub client: stackable_operator::client::Client,
-    pub product_config: ProductConfigManager,
 }
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
@@ -218,7 +215,10 @@ fn pod_template(
         pod_spec.node_selector = node_selector;
     }
     Ok(Pod {
-        metadata: ObjectMeta::default(),
+        metadata: ObjectMetaBuilder::new()
+            .name(container_name)
+            .with_labels(spark_application.recommended_labels())
+            .build(),
         spec: Some(pod_spec),
         ..Pod::default()
     })
diff --git a/tests/README.md b/tests/README.md
index 0ef47ef1..a3ef496e 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -1,3 +1,5 @@
 # spark-k8s-operator-integration-tests
 
-TODO
+This is a list of integration tests:
+
+- `spark-examples` : Runs the simplest form of Spark applications, without any external dependencies. The artifact (jar files) is bundled with the Spark image it's self.
diff --git a/tests/templates/kuttl/node-selector/10-assert.yaml b/tests/templates/kuttl/node-selector/10-assert.yaml
index e9be32f8..8ed6eb40 100644
--- a/tests/templates/kuttl/node-selector/10-assert.yaml
+++ b/tests/templates/kuttl/node-selector/10-assert.yaml
@@ -6,9 +6,9 @@ metadata:
 timeout: 300
 ---
 # The Job starting the whole process
-apiVersion: batch/v1
-kind: Job
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
 metadata:
   name: node-selector-test
 status:
-  succeeded: 1
+  phase: Succeeded
diff --git a/tests/templates/kuttl/spark-examples/10-assert.yaml b/tests/templates/kuttl/spark-examples/10-assert.yaml
new file mode 100644
index 00000000..e8c13bc2
--- /dev/null
+++ b/tests/templates/kuttl/spark-examples/10-assert.yaml
@@ -0,0 +1,14 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+metadata:
+  name: spark-examples
+timeout: 900
+---
+# The Job starting the whole process
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: spark-examples
+status:
+  phase: Succeeded
diff --git a/tests/templates/kuttl/spark-examples/10-deploy-spark-app.yaml.j2 b/tests/templates/kuttl/spark-examples/10-deploy-spark-app.yaml.j2
new file mode 100644
index 00000000..d1e24741
--- /dev/null
+++ b/tests/templates/kuttl/spark-examples/10-deploy-spark-app.yaml.j2
@@ -0,0 +1,20 @@
+---
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: spark-examples
+spec:
+  version: "1.0"
+  sparkImage: docker.stackable.tech/stackable/spark-k8s:{{ test_scenario['values']['spark'] }}-hadoop{{ test_scenario['values']['hadoop'][:-2] }}-stackable{{ test_scenario['values']['stackable'] }}
+  sparkImagePullPolicy: IfNotPresent
+  mode: cluster
+  mainClass: org.apache.spark.examples.SparkALS
+  mainApplicationFile: local:///stackable/spark/examples/jars/spark-examples_2.12-{{ test_scenario['values']['spark'] }}.jar
+  driver:
+    cores: 1
+    coreLimit: "1200m"
+    memory: "512m"
+  executor:
+    cores: 1
+    instances: 1
+    memory: "512m"
diff --git a/tests/templates/kuttl/spark-ny-public-s3/10-assert.yaml b/tests/templates/kuttl/spark-ny-public-s3/10-assert.yaml
index ac886d20..76c719f3 100644
--- a/tests/templates/kuttl/spark-ny-public-s3/10-assert.yaml
+++ b/tests/templates/kuttl/spark-ny-public-s3/10-assert.yaml
@@ -6,9 +6,9 @@ metadata:
 timeout: 900
 ---
 # The Job starting the whole process
-apiVersion: batch/v1
-kind: Job
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
 metadata:
   name: spark-ny-cm
 status:
-  succeeded: 1
+  phase: Succeeded
diff --git a/tests/templates/kuttl/spark-pi-private-s3/10-assert.yaml b/tests/templates/kuttl/spark-pi-private-s3/10-assert.yaml
index e391554d..fda7e6cc 100644
--- a/tests/templates/kuttl/spark-pi-private-s3/10-assert.yaml
+++ b/tests/templates/kuttl/spark-pi-private-s3/10-assert.yaml
@@ -6,9 +6,9 @@ metadata:
 timeout: 900
 ---
 # The Job starting the whole process
-apiVersion: batch/v1
-kind: Job
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
 metadata:
   name: spark-pi-private-s3
 status:
-  succeeded: 1
+  phase: Succeeded
diff --git a/tests/templates/kuttl/spark-pi-public-s3/10-assert.yaml b/tests/templates/kuttl/spark-pi-public-s3/10-assert.yaml
index 5d2e1de9..5c3c55ad 100644
--- a/tests/templates/kuttl/spark-pi-public-s3/10-assert.yaml
+++ b/tests/templates/kuttl/spark-pi-public-s3/10-assert.yaml
@@ -6,9 +6,9 @@ metadata:
 timeout: 900
 ---
 # The Job starting the whole process
-apiVersion: batch/v1
-kind: Job
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
 metadata:
   name: spark-pi-public-s3
 status:
-  succeeded: 1
+  phase: Succeeded
diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml
index d1f68a64..e4350a0e 100644
--- a/tests/test-definition.yaml
+++ b/tests/test-definition.yaml
@@ -30,3 +30,8 @@ tests:
       - spark
       - hadoop
       - stackable
+  - name: spark-examples
+    dimensions:
+      - spark
+      - hadoop
+      - stackable