From 4580ea26fe2285625cb1ad74a0db36c10f05f86c Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Mon, 13 Oct 2025 11:56:41 +0200
Subject: [PATCH 1/3] fix: remove usage of the pre-stop hook

---
 rust/operator-binary/src/config/command.rs    | 44 ++++++++++++++++++-
 .../src/resource/statefulset.rs               | 19 ++------
 2 files changed, 47 insertions(+), 16 deletions(-)

diff --git a/rust/operator-binary/src/config/command.rs b/rust/operator-binary/src/config/command.rs
index 8887c904..ce31bd5b 100644
--- a/rust/operator-binary/src/config/command.rs
+++ b/rust/operator-binary/src/config/command.rs
@@ -137,6 +137,48 @@ fn broker_start_command(
     }
 }
 
+// During a namespace or stacklet delete the Kafka controllers shut down too fast leaving the brokers
+// in a bad state.
+// Brokers try to connect to controllers before gracefully shutting down but by that time, all
+// controllers are already gone.
+// The broker pods are then kept alive until the value of `gracefulShutdownTimeout` is reached.
+// The environment variable `PRE_STOP_CONTROLLER_SLEEP_SECONDS` delays the termination of the
+// controller processes to give the brokers more time to offload data and shutdown gracefully.
+// Kubernetes has a built in `pre-stop` hook feature that is not yet generally available on all platforms
+// supported by the operator.
+const NOT_SO_COMMON_BASH_TRAP_FUNCTIONS: &str = r#"
+prepare_signal_handlers()
+{
+    unset term_child_pid
+    unset term_kill_needed
+    trap 'handle_term_signal' TERM
+}
+
+handle_term_signal()
+{
+    if [ "${term_child_pid}" ]; then
+        [ -z "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS"
+        kill -TERM "${term_child_pid}" 2>/dev/null
+    else
+        term_kill_needed="yes"
+    fi
+}
+
+wait_for_termination()
+{
+    set +e
+    term_child_pid=$1
+    if [[ -v term_kill_needed ]]; then
+        [ -z "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS"
+        kill -TERM "${term_child_pid}" 2>/dev/null
+    fi
+    wait ${term_child_pid} 2>/dev/null
+    trap - TERM
+    wait ${term_child_pid} 2>/dev/null
+    set -e
+}
+"#;
+
 pub fn controller_kafka_container_command(
     cluster_id: &str,
     controller_descriptors: Vec<KafkaPodDescriptor>,
@@ -152,7 +194,7 @@ pub fn controller_kafka_container_command(
     // - use config-utils for proper replacements?
     // - should we print the adapted properties file at startup?
     formatdoc! {"
-        {COMMON_BASH_TRAP_FUNCTIONS}
+        {NOT_SO_COMMON_BASH_TRAP_FUNCTIONS}
         {remove_vector_shutdown_file_command}
         prepare_signal_handlers
         containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
diff --git a/rust/operator-binary/src/resource/statefulset.rs b/rust/operator-binary/src/resource/statefulset.rs
index 18bcc04c..b5b758ec 100644
--- a/rust/operator-binary/src/resource/statefulset.rs
+++ b/rust/operator-binary/src/resource/statefulset.rs
@@ -23,8 +23,8 @@ use stackable_operator::{
             apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetUpdateStrategy},
             core::v1::{
                 ConfigMapKeySelector, ConfigMapVolumeSource, ContainerPort, EnvVar, EnvVarSource,
-                ExecAction, Lifecycle, LifecycleHandler, ObjectFieldSelector, PodSpec, Probe,
-                ServiceAccount, SleepAction, TCPSocketAction, Volume,
+                ExecAction, ObjectFieldSelector, PodSpec, Probe, ServiceAccount, TCPSocketAction,
+                Volume,
             },
         },
         apimachinery::pkg::{apis::meta::v1::LabelSelector, util::intstr::IntOrString},
@@ -658,6 +658,7 @@ pub fn build_controller_rolegroup_statefulset(
             kafka_security,
             &resolved_product_image.product_version,
         )])
+        .add_env_var("PRE_STOP_CONTROLLER_SLEEP_SECONDS", "5")
         .add_env_var(
             "EXTRA_ARGS",
             kafka_role
@@ -756,19 +757,7 @@ pub fn build_controller_rolegroup_statefulset(
         )
         .context(AddVolumesAndVolumeMountsSnafu)?;
 
-    // Currently, Controllers shutdown very fast, too fast in most times (flakyness) for the Brokers
-    // to off load properly. The Brokers then try to connect to any controllers until the
-    // `gracefulShutdownTimeout` is reached and the pod is finally killed.
-    // The `pre-stop` hook will delay the kill signal to the Controllers to provide the Brokers more
-    // time to offload data.
-    let mut kafka_container = cb_kafka.build();
-    kafka_container.lifecycle = Some(Lifecycle {
-        pre_stop: Some(LifecycleHandler {
-            sleep: Some(SleepAction { seconds: 10 }),
-            ..Default::default()
-        }),
-        ..Default::default()
-    });
+    let kafka_container = cb_kafka.build();
 
     pod_builder
         .metadata(metadata)

From 0db43a29aa90e50ad0c77dac3dc4f162a58961bb Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Mon, 13 Oct 2025 13:39:20 +0200
Subject: [PATCH 2/3] fix test switcharoo

---
 rust/operator-binary/src/config/command.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/rust/operator-binary/src/config/command.rs b/rust/operator-binary/src/config/command.rs
index ce31bd5b..a02474b5 100644
--- a/rust/operator-binary/src/config/command.rs
+++ b/rust/operator-binary/src/config/command.rs
@@ -157,7 +157,7 @@ prepare_signal_handlers()
 handle_term_signal()
 {
     if [ "${term_child_pid}" ]; then
-        [ -z "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS"
+        [ -n "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS"
         kill -TERM "${term_child_pid}" 2>/dev/null
     else
         term_kill_needed="yes"
@@ -169,7 +169,7 @@ wait_for_termination()
     set +e
     term_child_pid=$1
     if [[ -v term_kill_needed ]]; then
-        [ -z "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS"
+        [ -n "$PRE_STOP_CONTROLLER_SLEEP_SECONDS" ] && sleep "$PRE_STOP_CONTROLLER_SLEEP_SECONDS"
         kill -TERM "${term_child_pid}" 2>/dev/null
     fi
     wait ${term_child_pid} 2>/dev/null

From 0d8e71dadf23eac57a1c982e5d7c6b2ea366ccd8 Mon Sep 17 00:00:00 2001
From: Razvan-Daniel Mihai <84674+razvan@users.noreply.github.com>
Date: Mon, 13 Oct 2025 13:41:09 +0200
Subject: [PATCH 3/3] rename constant

---
 rust/operator-binary/src/config/command.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/rust/operator-binary/src/config/command.rs b/rust/operator-binary/src/config/command.rs
index a02474b5..3c25a8be 100644
--- a/rust/operator-binary/src/config/command.rs
+++ b/rust/operator-binary/src/config/command.rs
@@ -146,7 +146,7 @@ fn broker_start_command(
 // controller processes to give the brokers more time to offload data and shutdown gracefully.
 // Kubernetes has a built in `pre-stop` hook feature that is not yet generally available on all platforms
 // supported by the operator.
-const NOT_SO_COMMON_BASH_TRAP_FUNCTIONS: &str = r#"
+const BASH_TRAP_FUNCTIONS: &str = r#"
 prepare_signal_handlers()
 {
     unset term_child_pid
@@ -194,7 +194,7 @@ pub fn controller_kafka_container_command(
     // - use config-utils for proper replacements?
     // - should we print the adapted properties file at startup?
     formatdoc! {"
-        {NOT_SO_COMMON_BASH_TRAP_FUNCTIONS}
+        {BASH_TRAP_FUNCTIONS}
         {remove_vector_shutdown_file_command}
         prepare_signal_handlers
         containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &