diff --git a/tests/templates/kuttl/graceful-shutdown/00-patch-ns.yaml.j2 b/tests/templates/kuttl/graceful-shutdown/00-patch-ns.yaml.j2
new file mode 100644
index 00000000..67185acf
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/00-patch-ns.yaml.j2
@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}
diff --git a/tests/templates/kuttl/graceful-shutdown/01-assert.yaml.j2 b/tests/templates/kuttl/graceful-shutdown/01-assert.yaml.j2
new file mode 100644
index 00000000..50b1d4c3
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/01-assert.yaml.j2
@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}
diff --git a/tests/templates/kuttl/graceful-shutdown/01-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/graceful-shutdown/01-install-vector-aggregator-discovery-configmap.yaml.j2
new file mode 100644
index 00000000..2d6a0df5
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/01-install-vector-aggregator-discovery-configmap.yaml.j2
@@ -0,0 +1,9 @@
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+{% endif %}
diff --git a/tests/templates/kuttl/graceful-shutdown/10-assert.yaml b/tests/templates/kuttl/graceful-shutdown/10-assert.yaml
new file mode 100644
index 00000000..f30e7b3f
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/10-assert.yaml
@@ -0,0 +1,19 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 900
+commands:
+  # Both executors have to be registered and working: an executor that is still registering is
+  # not in the driver's list yet and would never be asked to shut down.
+  - script: |
+      set -eu
+
+      registered=0
+      for pod in $(kubectl -n "$NAMESPACE" get pods -o name \
+        -l spark-role=executor,app.kubernetes.io/instance=graceful-shutdown | cut -d/ -f2); do
+        log=$(kubectl -n "$NAMESPACE" logs --tail=-1 "$pod" -c spark)
+        echo "$log" | grep -q 'Successfully registered with driver'
+        echo "$log" | grep -q 'Running task'
+        registered=$(( registered + 1 ))
+      done
+      test "$registered" -eq 2
diff --git a/tests/templates/kuttl/graceful-shutdown/10-deploy-long-running-app.yaml.j2 b/tests/templates/kuttl/graceful-shutdown/10-deploy-long-running-app.yaml.j2
new file mode 100644
index 00000000..08f8152f
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/10-deploy-long-running-app.yaml.j2
@@ -0,0 +1,55 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: long-running-job
+data:
+  # Keeps the executors busy for longer than the test runs, so that they are stopped mid-task.
+  long-running-job.py: |
+    import time
+
+    from pyspark.sql import SparkSession
+
+    spark = SparkSession.builder.appName("long-running-job").getOrCreate()
+
+    def slow(i):
+        time.sleep(600)
+        return i
+
+    print(spark.sparkContext.parallelize(range(4), 4).map(slow).collect())
+---
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: graceful-shutdown
+spec:
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+  vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  sparkImage:
+{% if test_scenario['values']['spark'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['spark'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['spark'].split(',')[0] }}"
+{% else %}
+    productVersion: "{{ test_scenario['values']['spark'] }}"
+{% endif %}
+    pullPolicy: IfNotPresent
+  mode: cluster
+  mainApplicationFile: "local:///stackable/spark/jobs/long-running-job.py"
+  driver:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+      volumeMounts:
+        - name: script
+          mountPath: /stackable/spark/jobs
+  # Only the driver reads the script; the executors receive the closure over the wire.
+  executor:
+    replicas: 2
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+  volumes:
+    - name: script
+      configMap:
+        name: long-running-job
diff --git a/tests/templates/kuttl/graceful-shutdown/11-check-executor-shutdown.yaml b/tests/templates/kuttl/graceful-shutdown/11-check-executor-shutdown.yaml
new file mode 100644
index 00000000..a67f4682
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/11-check-executor-shutdown.yaml
@@ -0,0 +1,61 @@
+---
+# An executor whose Pod is deleted (node drain, eviction, downscale) has to be told to stop while
+# it is running a task. If SIGTERM is not forwarded to the JVM, the container ignores it and is
+# SIGKILLed when the grace period expires, taking the running task with it (docker-images#1564).
+#
+# TestStep and not TestAssert: kuttl retries asserts until they time out, which would turn every
+# failure into a generic timeout.
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - timeout: 600
+    script: |
+      set -eu
+
+      selector="spark-role=executor,app.kubernetes.io/instance=graceful-shutdown"
+      budget_seconds=20
+
+      victim=$(kubectl -n "$NAMESPACE" get pods -o name -l "$selector" | head -1 | cut -d/ -f2)
+      if [ -z "$victim" ]; then
+        echo "FAIL: no executor Pod found to stop"
+        exit 1
+      fi
+
+      echo "stopping executor $victim"
+
+      # The log cannot be read once the Pod object is gone, so stream it away first.
+      log=$(mktemp)
+      kubectl -n "$NAMESPACE" logs -f "$victim" -c spark >"$log" 2>&1 &
+      logs_pid=$!
+      sleep 2
+
+      if ! grep -q 'Running task' "$log"; then
+        echo "FAIL: could not stream the log of $victim, so its shutdown cannot be observed"
+        tail -5 "$log"
+        exit 1
+      fi
+
+      kubectl -n "$NAMESPACE" delete pod "$victim" --wait=false
+
+      start=$(date +%s)
+      stopped=no
+      while [ $(( $(date +%s) - start )) -lt "$budget_seconds" ]; do
+        if grep -q 'shutdown-hook' "$log"; then
+          stopped=yes
+          break
+        fi
+        sleep 1
+      done
+      elapsed=$(( $(date +%s) - start ))
+      kill "$logs_pid" 2>/dev/null || true
+
+      if [ "$stopped" != yes ]; then
+        echo "FAIL: ${elapsed}s after its Pod was deleted the executor had run no shutdown hook,"
+        echo "so SIGTERM did not reach the JVM and the kubelet will SIGKILL it at the end of the"
+        echo "grace period, killing the task it is running."
+        tail -5 "$log"
+        exit 1
+      fi
+
+      rm -f "$log"
+      echo "OK: the executor ran its shutdown hooks ${elapsed}s after its Pod was deleted"
diff --git a/tests/templates/kuttl/graceful-shutdown/12-assert.yaml b/tests/templates/kuttl/graceful-shutdown/12-assert.yaml
new file mode 100644
index 00000000..bf511321
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/12-assert.yaml
@@ -0,0 +1,19 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 600
+commands:
+  # Same check as in step 10: the replacement for the stopped executor has to be registered and
+  # working before the driver is stopped, otherwise it is not in the driver's list yet.
+  - script: |
+      set -eu
+
+      registered=0
+      for pod in $(kubectl -n "$NAMESPACE" get pods -o name \
+        -l spark-role=executor,app.kubernetes.io/instance=graceful-shutdown | cut -d/ -f2); do
+        log=$(kubectl -n "$NAMESPACE" logs --tail=-1 "$pod" -c spark)
+        echo "$log" | grep -q 'Successfully registered with driver'
+        echo "$log" | grep -q 'Running task'
+        registered=$(( registered + 1 ))
+      done
+      test "$registered" -eq 2
diff --git a/tests/templates/kuttl/graceful-shutdown/13-check-driver-shutdown-propagation.yaml b/tests/templates/kuttl/graceful-shutdown/13-check-driver-shutdown-propagation.yaml
new file mode 100644
index 00000000..683d94e5
--- /dev/null
+++ b/tests/templates/kuttl/graceful-shutdown/13-check-driver-shutdown-propagation.yaml
@@ -0,0 +1,77 @@
+---
+# Deleting the driver Pod has to let the driver JVM stop the SparkContext, which is what asks the
+# executors to shut down. Without SIGTERM reaching the driver it is SIGKILLed at the end of its
+# grace period, the executors are never asked, and they keep working until the driver's RPC
+# connection dies.
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - timeout: 600
+    script: |
+      set -eu
+
+      app=graceful-shutdown
+      # The stop command reaches registered executors about a second after the driver is signalled.
+      budget_seconds=20
+
+      capture=$(mktemp -d)
+      driver=$(kubectl -n "$NAMESPACE" get pods -o name \
+        -l "spark-role=driver,app.kubernetes.io/instance=$app" | head -1 | cut -d/ -f2)
+      executors=$(kubectl -n "$NAMESPACE" get pods -o name \
+        -l "spark-role=executor,app.kubernetes.io/instance=$app" | cut -d/ -f2)
+
+      # Without these the loop below has nothing to wait for and would report success while having
+      # checked nothing at all.
+      if [ -z "$driver" ]; then
+        echo "FAIL: no driver Pod found for $app"
+        exit 1
+      fi
+      if [ -z "$executors" ]; then
+        echo "FAIL: no executor Pods found for $app, so no shutdown can be propagated to any"
+        exit 1
+      fi
+
+      echo "stopping driver $driver, expecting a shutdown at: $(echo "$executors" | tr '\n' ' ')"
+
+      log_pids=""
+      for pod in $driver $executors; do
+        kubectl -n "$NAMESPACE" logs -f "$pod" -c spark >"$capture/$pod.log" 2>&1 &
+        log_pids="$log_pids $!"
+      done
+      sleep 2
+
+      kubectl -n "$NAMESPACE" delete pod "$driver" --wait=false
+
+      start=$(date +%s)
+      missing=$executors
+      while [ $(( $(date +%s) - start )) -lt "$budget_seconds" ]; do
+        missing=""
+        for pod in $executors; do
+          grep -q 'Driver commanded a shutdown' "$capture/$pod.log" || missing="$missing $pod"
+        done
+        if [ -z "$missing" ]; then
+          break
+        fi
+        sleep 1
+      done
+      elapsed=$(( $(date +%s) - start ))
+      sleep 1
+      kill $log_pids 2>/dev/null || true
+
+      if [ -n "$missing" ]; then
+        echo "FAIL: ${elapsed}s after the driver was stopped, these executors had not been asked"
+        echo "to shut down:$missing"
+        if grep -q 'Asking each executor to shut down' "$capture/$driver.log"; then
+          echo "The driver did ask, so the request was lost on the way to the executors:"
+          for pod in $missing; do
+            tail -3 "$capture/$pod.log"
+          done
+        else
+          echo "The driver ran no shutdown hook, so SIGTERM did not reach the driver JVM:"
+          tail -5 "$capture/$driver.log"
+        fi
+        exit 1
+      fi
+
+      rm -rf "$capture"
+      echo "OK: every executor was asked to shut down ${elapsed}s after the driver was stopped"
diff --git a/tests/templates/kuttl/signal-propagation/00-patch-ns.yaml.j2 b/tests/templates/kuttl/signal-propagation/00-patch-ns.yaml.j2
new file mode 100644
index 00000000..67185acf
--- /dev/null
+++ b/tests/templates/kuttl/signal-propagation/00-patch-ns.yaml.j2
@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}
diff --git a/tests/templates/kuttl/signal-propagation/01-assert.yaml.j2 b/tests/templates/kuttl/signal-propagation/01-assert.yaml.j2
new file mode 100644
index 00000000..aa13050d
--- /dev/null
+++ b/tests/templates/kuttl/signal-propagation/01-assert.yaml.j2
@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if test_scenario['values']['vector'] == 'true' %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}
diff --git a/tests/templates/kuttl/signal-propagation/01-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/signal-propagation/01-install-vector-aggregator-discovery-configmap.yaml.j2
new file mode 100644
index 00000000..ff3b36d0
--- /dev/null
+++ b/tests/templates/kuttl/signal-propagation/01-install-vector-aggregator-discovery-configmap.yaml.j2
@@ -0,0 +1,12 @@
+{% if test_scenario['values']['vector'] == 'true' %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  # This test only asserts that the Vector agent is shut down again, not that the log events
+  # arrive anywhere, so it does not need a reachable aggregator. If one is provided it is used,
+  # otherwise the agent runs and logs connection errors, which does not affect the assertions.
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') or 'localhost:6123' }}
+{% endif %}
diff --git a/tests/templates/kuttl/signal-propagation/10-deploy-failing-spark-app.yaml.j2 b/tests/templates/kuttl/signal-propagation/10-deploy-failing-spark-app.yaml.j2
new file mode 100644
index 00000000..01781890
--- /dev/null
+++ b/tests/templates/kuttl/signal-propagation/10-deploy-failing-spark-app.yaml.j2
@@ -0,0 +1,55 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: failing-job
+data:
+  # A job that starts up correctly, runs a real Spark action, and only then fails.
+  # This makes sure we are testing the exit code of a genuinely failing application
+  # and not a startup or submission error.
+  failing-job.py: |
+    from pyspark.sql import SparkSession
+
+    spark = SparkSession.builder.appName("failing-job").getOrCreate()
+
+    # forces an actual computation
+    print(f"count = {spark.range(1000).count()}")
+
+    raise RuntimeError("deliberate failure to test exit code propagation")
+---
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: failure-propagation
+spec:
+{% if test_scenario['values']['vector'] == 'true' %}
+  vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  sparkImage:
+{% if test_scenario['values']['spark'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['spark'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['spark'].split(',')[0] }}"
+{% else %}
+    productVersion: "{{ test_scenario['values']['spark'] }}"
+{% endif %}
+    pullPolicy: IfNotPresent
+  mode: cluster
+  mainApplicationFile: "local:///stackable/spark/jobs/failing-job.py"
+  driver:
+    config:
+      logging:
+        enableVectorAgent: {{ test_scenario['values']['vector'] }}
+      volumeMounts:
+        - name: script
+          mountPath: /stackable/spark/jobs
+  # The executors only run the tasks of the range/count job, they never read the script,
+  # so they do not need the volume mounted.
+  executor:
+    replicas: 1
+    config:
+      logging:
+        enableVectorAgent: {{ test_scenario['values']['vector'] }}
+  volumes:
+    - name: script
+      configMap:
+        name: failing-job
diff --git a/tests/templates/kuttl/signal-propagation/11-check-phase.yaml b/tests/templates/kuttl/signal-propagation/11-check-phase.yaml
new file mode 100644
index 00000000..3f8ce2f0
--- /dev/null
+++ b/tests/templates/kuttl/signal-propagation/11-check-phase.yaml
@@ -0,0 +1,74 @@
+---
+# This is a TestStep and not a TestAssert on purpose. kuttl retries an assert until it
+# times out, so a wrong-but-terminal phase could only ever be reported after the full
+# timeout. A TestStep command runs once, which lets us fail as soon as the application
+# reaches a terminal phase and say why.
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  # Deliberately above the script's own budget below, so that on a timeout the script
+  # reports the last phase it saw instead of kuttl killing it with a generic message.
+  - timeout: 960
+    script: |
+      set -eu
+
+      timeout_seconds=900
+      interval_seconds=5
+      # How long the application may still take to reach a terminal phase after the Spark
+      # container itself has terminated. The post hook sleeps 10 seconds before telling Vector
+      # to shut down, so this only has to be comfortably above that.
+      grace_seconds=120
+
+      driver_containers() {
+        kubectl -n "$NAMESPACE" get pods -l spark-role=driver \
+          -o jsonpath='{range .items[*].status.containerStatuses[*]}{.name}={.state}{"\n"}{end}' \
+          2>/dev/null || true
+      }
+
+      phase=""
+      elapsed=0
+      spark_done_at=-1
+      while [ "$elapsed" -lt "$timeout_seconds" ]; do
+        phase=$(kubectl -n "$NAMESPACE" get sparkapplication failure-propagation \
+          -o jsonpath='{.status.phase}' 2>/dev/null || true)
+
+        case "$phase" in
+          Failed)
+            echo "OK: the failing application was reported as Failed"
+            exit 0
+            ;;
+          Succeeded)
+            echo "FAIL: the application failed at runtime but was reported as Succeeded."
+            echo "The driver container exit code was masked between the Spark process and"
+            echo "the container entrypoint (see spark-k8s/stackable/run-spark.sh)."
+            exit 1
+            ;;
+        esac
+
+        # Once the Spark container has terminated, the Pod has to follow shortly after.
+        # If it does not, the entrypoint skipped its post hook after the Spark process
+        # exited, so a sidecar was never told to shut down and the Pod can never reach
+        # a terminal phase at all.
+        containers=$(driver_containers)
+        if echo "$containers" | grep -q '^spark=.*terminated'; then
+          if [ "$spark_done_at" -lt 0 ]; then
+            spark_done_at="$elapsed"
+          elif [ "$((elapsed - spark_done_at))" -ge "$grace_seconds" ]; then
+            echo "FAIL: the Spark container terminated ${grace_seconds}s ago, but the application"
+            echo "is still in phase '${phase}'. The entrypoint did not run its post hook after the"
+            echo "Spark process exited, so a sidecar was never told to shut down and the Pod cannot"
+            echo "terminate (see spark-k8s/stackable/run-spark.sh)."
+            echo "Driver container states:"
+            echo "$containers" | sed 's/^/  /'
+            exit 1
+          fi
+        fi
+
+        sleep "$interval_seconds"
+        elapsed=$((elapsed + interval_seconds))
+      done
+
+      echo "FAIL: no terminal phase after ${timeout_seconds}s (last seen: '${phase}')"
+      echo "Driver container states:"
+      driver_containers | sed 's/^/  /'
+      exit 1
diff --git a/tests/templates/kuttl/signal-propagation/12-assert.yaml b/tests/templates/kuttl/signal-propagation/12-assert.yaml
new file mode 100644
index 00000000..19880980
--- /dev/null
+++ b/tests/templates/kuttl/signal-propagation/12-assert.yaml
@@ -0,0 +1,16 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 300
+commands:
+  # Test that there is no spark driver pod left hanging. The Pod list is captured in its own step so
+  # that a failing kubectl cannot be mistaken for "no driver Pod present".
+  - script: |
+      set -eu
+
+      pods=$(kubectl -n "$NAMESPACE" get pods -o name)
+      if echo "$pods" | grep -qE 'driver$'; then
+        echo "FAIL: a driver Pod is still present after the application reached a terminal phase:"
+        echo "$pods" | grep -E 'driver$'
+        exit 1
+      fi
diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml
index 5d387b63..3b373261 100644
--- a/tests/test-definition.yaml
+++ b/tests/test-definition.yaml
@@ -11,6 +11,13 @@ dimensions:
       # Alternatively, if you want to use a custom image, append a comma and the full image name to the product version
       # as in the example below.
       # - 3.5.6,oci.stackable.tech/sandbox/spark-k8s:3.5.6-stackable0.0.0-dev
+  # Whether the Vector agent is enabled. With it enabled the operator sets a post hook on the
+  # product container, which the entrypoint has to run even when the application failed. If it
+  # does not, the agent is never shut down and the Pod cannot reach a terminal phase.
+  - name: vector
+    values:
+      - "false"
+      - "true"
   - name: spark-logging
     values:
       - 3.5.8
@@ -93,6 +100,15 @@ tests:
     dimensions:
       - spark
       - openshift
+  - name: signal-propagation
+    dimensions:
+      - spark
+      - vector
+      - openshift
+  - name: graceful-shutdown
+    dimensions:
+      - spark
+      - openshift
   - name: pyspark-ny-public-s3
     dimensions:
       - spark