feat: Support graceful shutdown

Author: sbernauer

CHANGELOG.md

@@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file.
 - Support PodDisruptionBudgets ([#625]).
 - Support new versions 2.8.2, 3.4.1, 3.5.1 ([#627]).
 - Document internal clusterId check ([#631]).
+- Support graceful shutdown ([#XXX]).
 
 ### Changed
 

Cargo.lock

@@ -14,14 +14,15 @@ repository = "https://github.com/stackabletech/kafka-operator"
 built = { version = "0.6", features = ["chrono", "git2"] }
 clap = "4.3"
 futures = { version = "0.3", features = ["compat"] }
+indoc = "2.0"
+product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.6.0" }
 rstest = "0.18"
 semver = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 serde_yaml = "0.9"
 snafu = "0.7"
 stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "0.56.0" }
-product-config = { git = "https://github.com/stackabletech/product-config.git", tag = "0.6.0" }
 strum = { version = "0.25", features = ["derive"] }
 tokio = { version = "1.29", features = ["full"] }
 tracing = "0.1"

Cargo.toml

@@ -507,6 +507,10 @@ spec:
  type: array
  type: object
  type: object
+ gracefulShutdownTimeout:
+ description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
+ nullable: true
+ type: string
  logging:
  default:
  enableVectorAgent: null
@@ -3991,6 +3995,10 @@ spec:
  type: array
  type: object
  type: object
+ gracefulShutdownTimeout:
+ description: Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
+ nullable: true
+ type: string
  logging:
  default:
  enableVectorAgent: null

deploy/helm/kafka-operator/crds/crds.yaml

@@ -9,9 +9,10 @@ repository.workspace = true
 publish = false
 
 [dependencies]
+indoc.workspace = true
 semver.workspace = true
-serde.workspace = true
 serde_json.workspace = true
+serde.workspace = true
 snafu.workspace = true
 stackable-operator.workspace = true
 strum.workspace = true

rust/crd/Cargo.toml

@@ -33,6 +33,7 @@ use stackable_operator::{
  role_utils::{GenericRoleConfig, Role, RoleGroup, RoleGroupRef},
  schemars::{self, JsonSchema},
  status::condition::{ClusterCondition, HasStatusCondition},
+ time::Duration,
 };
 use std::{collections::BTreeMap, str::FromStr};
 use strum::{Display, EnumIter, EnumString, IntoEnumIterator};
@@ -55,6 +56,9 @@ pub const STACKABLE_TMP_DIR: &str = "/stackable/tmp";
 pub const STACKABLE_DATA_DIR: &str = "/stackable/data";
 pub const STACKABLE_CONFIG_DIR: &str = "/stackable/config";
 pub const STACKABLE_LOG_CONFIG_DIR: &str = "/stackable/log_config";
+pub const STACKABLE_LOG_DIR: &str = "/stackable/log";
+
+const DEFAULT_BROKER_GRACEFUL_SHUTDOWN_TIMEOUT: Duration = Duration::from_minutes_unchecked(30);
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -382,10 +386,16 @@ pub enum Container {
 pub struct KafkaConfig {
  #[fragment_attrs(serde(default))]
  pub logging: Logging<Container>,
+
  #[fragment_attrs(serde(default))]
  pub resources: Resources<Storage, NoRuntimeLimits>,
+
  #[fragment_attrs(serde(default))]
  pub affinity: StackableAffinity,
+
+ /// Time period Pods have to gracefully shut down, e.g. `30m`, `1h` or `2d`. Consult the operator documentation for details.
+ #[fragment_attrs(serde(default))]
+ pub graceful_shutdown_timeout: Option<Duration>,
 }
 
 impl KafkaConfig {
@@ -410,6 +420,7 @@ impl KafkaConfig {
  },
  },
  affinity: get_affinity(cluster_name, role),
+ graceful_shutdown_timeout: Some(DEFAULT_BROKER_GRACEFUL_SHUTDOWN_TIMEOUT),
  }
  }
 }

rust/crd/src/lib.rs

@@ -4,22 +4,28 @@
 //! and helper functions
 //!
 //! This is required due to overlaps between TLS encryption and e.g. mTLS authentication or Kerberos
+use std::collections::BTreeMap;
 
-use crate::{
- authentication, authentication::ResolvedAuthenticationClasses, listener, tls, KafkaCluster,
- SERVER_PROPERTIES_FILE, STACKABLE_CONFIG_DIR, STACKABLE_TMP_DIR,
-};
-
-use crate::listener::KafkaListenerConfig;
+use indoc::formatdoc;
 use snafu::{ResultExt, Snafu};
 use stackable_operator::builder::SecretFormat;
+use stackable_operator::product_logging::framework::{
+ create_vector_shutdown_file_command, remove_vector_shutdown_file_command,
+};
 use stackable_operator::{
  builder::{ContainerBuilder, PodBuilder, SecretOperatorVolumeSourceBuilder, VolumeBuilder},
  client::Client,
  commons::authentication::{AuthenticationClass, AuthenticationClassProvider},
  k8s_openapi::api::core::v1::Volume,
+ utils::COMMON_BASH_TRAP_FUNCTIONS,
+};
+
+use crate::STACKABLE_LOG_DIR;
+use crate::{
+ authentication::{self, ResolvedAuthenticationClasses},
+ listener::{self, KafkaListenerConfig},
+ tls, KafkaCluster, SERVER_PROPERTIES_FILE, STACKABLE_CONFIG_DIR, STACKABLE_TMP_DIR,
 };
-use std::collections::BTreeMap;
 
 #[derive(Snafu, Debug)]
 pub enum Error {
@@ -225,23 +231,26 @@ impl KafkaTlsSecurity {
  kafka_listeners: &KafkaListenerConfig,
  opa_connect_string: Option<&str>,
  ) -> Vec<String> {
- vec![
- "bin/kafka-server-start.sh".to_string(),
- format!("{STACKABLE_CONFIG_DIR}/{SERVER_PROPERTIES_FILE}"),
- "--override \"zookeeper.connect=$ZOOKEEPER\"".to_string(),
- format!("--override \"listeners={}\"", kafka_listeners.listeners()),
- format!(
- "--override \"advertised.listeners={}\"",
- kafka_listeners.advertised_listeners()
- ),
- format!(
- "--override \"listener.security.protocol.map={}\"",
- kafka_listeners.listener_security_protocol_map()
- ),
- opa_connect_string.map_or("".to_string(), |opa| {
- format!("--override \"opa.authorizer.url={}\"", opa)
- }),
- ]
+ vec![formatdoc! {"
+ {COMMON_BASH_TRAP_FUNCTIONS}
+ {remove_vector_shutdown_file_command}
+ prepare_signal_handlers
+ bin/kafka-server-start.sh {STACKABLE_CONFIG_DIR}/{SERVER_PROPERTIES_FILE} --override \"zookeeper.connect=$ZOOKEEPER\" --override \"listeners={listeners}\" --override \"advertised.listeners={advertised_listeners}\" --override \"listener.security.protocol.map={listener_security_protocol_map}\"{opa_config} &
+ wait_for_termination $!
+ {create_vector_shutdown_file_command}
+ ",
+ remove_vector_shutdown_file_command =
+ remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
+ create_vector_shutdown_file_command =
+ create_vector_shutdown_file_command(STACKABLE_LOG_DIR),
+ listeners = kafka_listeners.listeners(),
+ advertised_listeners = kafka_listeners.advertised_listeners(),
+ listener_security_protocol_map = kafka_listeners.listener_security_protocol_map(),
+ opa_config = match opa_connect_string {
+ None => "".to_string(),
+ Some(opa_connect_string) => format!(" --override \"opa.authorizer.url={opa_connect_string}\""),
+ }
+ }]
  }
 
  /// Adds required volumes and volume mounts to the pod and container builders

rust/crd/src/security.rs

@@ -12,14 +12,14 @@ publish = false
 stackable-kafka-crd = { path = "../crd" }
 
 futures.workspace = true
-serde.workspace = true
+product-config.workspace = true
 serde_json.workspace = true
+serde.workspace = true
+snafu.workspace = true
 stackable-operator.workspace = true
-product-config.workspace = true
 strum.workspace = true
 tokio.workspace = true
 tracing.workspace = true
-snafu.workspace = true
 
 [dev-dependencies]
 serde_yaml.workspace = true