Re: [PR] test(gateways): add failover tests for gateways beside a 3-node cluster [iggy]

[krishvishal (via GitHub) via commits]

krishvishal commented on code in PR #4702:
URL: https://github.com/apache/iggy/pull/4702#discussion_r4441645541


##########
gateways/kafka/tests/cluster_failover_tests.rs:
##########
@@ -0,0 +1,1340 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Three gateways beside three Iggy nodes, under load from real Kafka 
clients, while gateways and
+//! nodes stop, crash and come back.
+//!
+//! The stack is real end to end. Three `iggy-server` processes form a 
cluster, the compiled
+//! `iggy-gateway-kafka` runs beside each node, and the clients run in Docker. 
An idempotent Java
+//! producer writes numbered records through the three gateways, and a Java 
consumer group of two
+//! members reads them. At the end, kcat reads back what each partition stores.
+//!
+//! A probe keeps a member of a second group alive with a heartbeat to every 
gateway, five times a
+//! second, and asks every gateway to describe the Java group. Both groups 
hash to one coordinator
+//! partition, so one gateway coordinates both.
+//!
+//! Each test checks the same things at the end. Every acknowledged record is 
stored once, at the
+//! offset that the producer got, and the group reads it once. The group does 
not rebalance, and no
+//! two gateways answer as coordinator of a group at once.
+//!
+//! Prerequisites are Docker, the two client images and an `iggy-server` built 
into the same target
+//! directory. Missing any skips, unless `KAFKA_E2E_REQUIRED=1` is set, as in
+//! `kafka_client_e2e_tests.rs`.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/deployment.rs"]
+mod deployment;
+#[path = "common/docker_clients.rs"]
+mod docker_clients;
+#[path = "common/gateways.rs"]
+mod gateways;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::fs;
+use std::future::Future;
+use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::{
+    DescribeGroupsRequest, DescribeGroupsResponse, FindCoordinatorRequest, 
FindCoordinatorResponse,
+    GroupId, HeartbeatRequest, HeartbeatResponse, MetadataRequest, 
MetadataResponse,
+};
+use kafka_protocol::protocol::{Decodable, StrBytes};
+use serde_json::Value;
+use serial_test::serial;
+use tempfile::TempDir;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, MissedTickBehavior};
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{COORDINATOR_LOG_TOPIC, cluster_id_stream};
+use iggy_gateway_kafka::coordinator::{PARTITIONS as COORDINATOR_PARTITIONS, 
partition_of_group};
+use iggy_gateway_kafka::protocol::api::{
+    API_KEY_DESCRIBE_GROUPS, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, 
API_KEY_METADATA,
+    ERROR_FENCED_INSTANCE_ID, ERROR_ILLEGAL_GENERATION, ERROR_NONE, 
ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_UNKNOWN_MEMBER_ID,
+};
+
+use deployment::{Deployment, KAFKA_STREAM, NODES, TAIL_LINES, tail};
+use docker_clients::{Container, missing_prerequisite, run_to_end, skip};
+
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+
+/// The topic of the load. Two partitions per node, so a stop of any node 
moves primaries.
+const TOPIC: &str = "failover";
+const PARTITIONS: u32 = 6;
+
+/// The Java consumer group, two members of `kafka-verifiable-consumer.sh`.
+const GROUP: &str = "failover-readers";
+const MEMBERS: usize = 2;
+
+/// Records per second that the producer writes, with keys that spread them 
over every partition.
+const RATE: u32 = 100;
+const KEYS: &str = "60";
+
+/// The idempotent Java producer. These are the defaults since Kafka 3.0, 
written out because
+/// `kafka-verifiable-producer.sh` sets `retries=0` before it reads this file. 
The delivery timeout
+/// outlasts every step, so no record expires in the producer.
+const PRODUCER_CONFIG: &str = "enable.idempotence=true\n\
+                               acks=all\n\
+                               retries=2147483647\n\
+                               max.in.flight.requests.per.connection=5\n\
+                               request.timeout.ms=10000\n\
+                               delivery.timeout.ms=300000\n";
+
+/// Acknowledgements, and reads of each member, that a step waits for, so the 
next step starts
+/// under load.
+const ACK_PROGRESS: usize = 300;
+const READ_PROGRESS: usize = 100;
+
+/// The Iggy session timeout by default: 1.2 x the 30 s `heartbeat.interval`.
+const SESSION_TIMEOUT: Duration = Duration::from_secs(36);
+
+/// Server settings for the shortest session timeout that a 1 s report 
interval allows. The server

Review Comment:
   Fixed. The comment now says that 8 s is the next whole second above the 
minimum of about 7.67 s
   that 1 s reports allow. The takeover comments now say that the metadata 
primary ends the session in
   its own pass once per interval, and the window comes from that.
   



##########
.config/nextest.toml:
##########
@@ -75,6 +75,23 @@ max-threads = 1
 filter = 'binary_id(iggy-gateway-kafka::kafka_client_e2e_tests)'
 test-group = "kafka_client_e2e"
 
+# Each test in these binaries runs a three-node cluster, so they run one at a 
time.
+[test-groups.kafka_cluster]
+max-threads = 1
+
+[[profile.default.overrides]]
+filter = 'binary_id(iggy-gateway-kafka::coordinator_cluster_tests)'
+test-group = "kafka_cluster"
+
+# The failover tests also run three gateway processes and Kafka clients in 
Docker for minutes, and
+# they time takeovers. Each gets the whole machine, so other tests do not 
distort that timing, and
+# up to ten minutes.
+[[profile.default.overrides]]
+filter = 'binary_id(iggy-gateway-kafka::cluster_failover_tests)'
+test-group = "kafka_cluster"
+threads-required = "num-cpus"
+slow-timeout = { period = "60s", terminate-after = 10 }

Review Comment:
   Fixed. `profile.ci` sets `retries = 0` for this binary. The 10 minute cap is 
gone: the slowest test
   took about 160 s here, inside the default 5 minute cap. Each test fails 
itself at 270 s, so it still
   prints its logs. Both overrides now filter with `and not 
test(/^iggy_server::/)`.
   



##########
gateways/kafka/tests/common/deployment.rs:
##########
@@ -0,0 +1,417 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Three Iggy nodes and a gateway process beside each, deployed as the 
gateway documentation
+//! describes it. Gateway `n` dials node `n - 1`, and every gateway lists all 
of them in
+//! `IGGY_KAFKA_BROKERS`. The gateways run the compiled binary, so a stop is a 
SIGTERM and a crash
+//! a SIGKILL, as an orchestrator sends them.
+//!
+//! Callers must also declare the `codec`, `gateways`, `iggy_server`, `tcp` 
and `wire` modules at
+//! their crate root, as for `gateways.rs`.
+#![allow(dead_code)]
+
+use std::fs::{self, OpenOptions};
+use std::net::SocketAddr;
+use std::path::{Path, PathBuf};
+use std::process::{Child, Command};
+use std::time::Duration;
+
+use iggy::prelude::{
+    AutoLogin, Client, ClusterClient, ClusterNodeRole, Credentials, 
IggyClient, IggyClientBuilder,
+    UserClient,
+};
+use iggy_binary_protocol::codes::GET_PARTITION_ROUTE_CODE;
+use iggy_binary_protocol::requests::messages::GetPartitionRouteRequest;
+use iggy_binary_protocol::responses::messages::PartitionRouteResponse;
+use iggy_binary_protocol::{WireDecode, WireEncode, WireIdentifier};
+use secrecy::SecretString;
+use tokio::time::Instant;
+
+use super::gateways;
+use super::iggy_server::{PortGuard, TestCluster, graceful_kill};
+
+/// Iggy nodes, each with a gateway beside it.
+pub const NODES: usize = 3;
+
+/// The stream of the Kafka topics under the default topic mapping.
+pub const KAFKA_STREAM: &str = "kafka";
+
+/// How long a gateway may take to listen. Its bridge waits up to 60 s for the 
cluster's metadata
+/// primary (`IGGY_STARTUP_LIMIT` in `main.rs`).
+const START_LIMIT: Duration = Duration::from_secs(70);
+
+/// How long a gateway may take to exit after SIGTERM: the hand-over of its 
groups, the drain of
+/// its connections and the sign-out of its Iggy clients. The default drain 
alone may take 25 s.
+const EXIT_LIMIT: Duration = Duration::from_secs(40);
+
+/// How long one question to the cluster may take.
+const QUERY_LIMIT: Duration = Duration::from_secs(10);
+
+/// The log lines of each gateway that a failure report shows.
+pub const TAIL_LINES: usize = 40;
+
+/// The nodes and their gateways, until the test drops it.
+pub struct Deployment {
+    pub cluster: TestCluster,
+    gateways: Vec<GatewayProcess>,
+    log_dir: PathBuf,
+}
+
+impl Deployment {
+    /// Starts the nodes with `settings`, as `IGGY_*` variables, creates each 
`(topic,
+    /// partitions)` of `topics`, then starts a gateway beside each node. The 
node data goes to
+    /// `data/` in `work_dir`, and every log to `logs/`.
+    pub async fn start(work_dir: &Path, settings: &[(&str, &str)], topics: 
&[(&str, u32)]) -> Self {
+        let log_dir = work_dir.join("logs");
+        fs::create_dir_all(&log_dir).expect("create the log dir");
+        let cluster =
+            TestCluster::spawn_with(&work_dir.join("data"), NODES, settings, 
Some(&log_dir)).await;
+        let bridge = gateways::connect(cluster.bridge_config(0)).await;
+        for &(topic, partitions) in topics {
+            bridge
+                .create_kafka_topic(topic, partitions)
+                .await
+                .expect("create a topic");
+        }
+        bridge.close().await.expect("close the bridge");
+
+        let ports: Vec<PortGuard> = (0..NODES).map(|_| 
PortGuard::acquire()).collect();
+        let brokers = (1_u16..)
+            .zip(&ports)
+            .map(|(node_id, port)| format!("{node_id}@127.0.0.1:{}", 
port.port))
+            .collect::<Vec<_>>()
+            .join(",");
+        let mut gateways: Vec<GatewayProcess> = (1_u16..)
+            .zip(ports)
+            .enumerate()
+            .map(|(node, (node_id, port))| GatewayProcess {
+                node_id,
+                addr: SocketAddr::from(([127, 0, 0, 1], port.port)),
+                iggy_addr: cluster.address(node),
+                brokers: brokers.clone(),
+                log: log_dir.join(format!("gateway-{node_id}.log")),
+                child: None,
+                _port: port,
+            })
+            .collect();
+        for gateway in &mut gateways {
+            gateway.spawn();
+        }
+        for gateway in &mut gateways {
+            gateway.wait_listening().await;
+        }
+        Self {
+            cluster,
+            gateways,
+            log_dir,
+        }
+    }
+
+    pub fn log_dir(&self) -> &Path {
+        &self.log_dir
+    }
+
+    /// The node ids of the gateways, in order.
+    pub fn gateway_ids(&self) -> Vec<u16> {
+        self.gateways
+            .iter()
+            .map(|gateway| gateway.node_id)
+            .collect()
+    }
+
+    /// The address of every gateway, stopped ones included, in node id order.
+    pub fn addrs(&self) -> Vec<SocketAddr> {
+        self.gateways.iter().map(|gateway| gateway.addr).collect()
+    }
+
+    /// The address of each running gateway.
+    pub fn running_addrs(&self) -> Vec<SocketAddr> {
+        self.gateways
+            .iter()
+            .filter(|gateway| gateway.child.is_some())
+            .map(|gateway| gateway.addr)
+            .collect()
+    }
+
+    /// `bootstrap.servers` for a client: every gateway, as the deployment 
guide says.
+    pub fn bootstrap(&self) -> String {
+        self.addrs()
+            .iter()
+            .map(ToString::to_string)
+            .collect::<Vec<_>>()
+            .join(",")
+    }
+
+    /// The node of gateway `node_id`: gateway `n` runs beside node `n - 1`.
+    pub fn node_of(node_id: u16) -> usize {
+        usize::from(node_id) - 1
+    }
+
+    /// Stops gateway `node_id` with SIGTERM and waits until it exits. Returns 
how long the exit
+    /// took. Fails the test when the gateway is still running after 
[`EXIT_LIMIT`].
+    pub async fn terminate_gateway(&mut self, node_id: u16) -> Duration {
+        let gateway = self.gateway_mut(node_id);
+        gateway.signal(libc::SIGTERM);
+        gateway.wait_exit().await
+    }
+
+    /// Stops gateway `node_id` and its node with SIGTERM at once, as when 
their machine shuts
+    /// down, and waits until both exit.
+    pub async fn terminate_gateway_and_node(&mut self, node_id: u16) {
+        self.gateway_mut(node_id).signal(libc::SIGTERM);
+        self.cluster.stop(Self::node_of(node_id));
+        self.gateway_mut(node_id).wait_exit().await;
+    }
+
+    /// Kills gateway `node_id` with SIGKILL, as a crash would. Its 
connections close, but Iggy
+    /// keeps its sessions until their timeout.
+    pub fn kill_gateway(&mut self, node_id: u16) {
+        let gateway = self.gateway_mut(node_id);
+        let mut child = gateway.child.take().expect("a running gateway");
+        child.kill().expect("SIGKILL the gateway");
+        child.wait().expect("reap the gateway");
+    }
+
+    /// Starts gateway `node_id` again on its port, and waits until it listens.
+    pub async fn start_gateway(&mut self, node_id: u16) {
+        let gateway = self.gateway_mut(node_id);
+        gateway.spawn();
+        gateway.wait_listening().await;
+    }
+
+    /// The replica id of the metadata primary, as a running node reports it, 
or `None` when no
+    /// node answers.
+    pub async fn metadata_primary(&self) -> Option<usize> {
+        let client = self.client().await?;
+        let metadata = tokio::time::timeout(QUERY_LIMIT, 
client.get_cluster_metadata()).await;
+        sign_out(&client).await;
+        let leader = metadata
+            .ok()?
+            .ok()?
+            .nodes
+            .into_iter()
+            .find(|node| node.role == ClusterNodeRole::Leader)?;
+        Some(usize::from(leader.replica_id))
+    }
+
+    /// The replica id of the primary of each of `partition_ids` of `topic` in 
`stream`, as a
+    /// running node reports them, or `None` when no node answers.
+    pub async fn primaries(
+        &self,
+        stream: &str,
+        topic: &str,
+        partition_ids: impl IntoIterator<Item = u32>,
+    ) -> Option<Vec<usize>> {
+        let client = self.client().await?;
+        let routes = async {
+            let mut primaries = Vec::new();
+            for partition_id in partition_ids {
+                let request = GetPartitionRouteRequest {
+                    stream_id: WireIdentifier::named(stream).ok()?,
+                    topic_id: WireIdentifier::named(topic).ok()?,
+                    partition_id,
+                }
+                .to_bytes();
+                let reply = tokio::time::timeout(
+                    QUERY_LIMIT,
+                    client.send_binary_request(GET_PARTITION_ROUTE_CODE, 
request),
+                )
+                .await
+                .ok()?
+                .ok()?;
+                let route = PartitionRouteResponse::decode_from(&reply).ok()?;
+                primaries.push(usize::from(route.primary_replica));
+            }
+            Some(primaries)
+        }
+        .await;
+        sign_out(&client).await;
+        routes
+    }
+
+    /// The last [`TAIL_LINES`] lines of the log of each gateway.
+    pub fn gateway_log_tails(&self) -> String {
+        self.gateways
+            .iter()
+            .map(|gateway| {
+                format!(
+                    "--- gateway {} ---\n{}",
+                    gateway.node_id,
+                    tail(&gateway.log, TAIL_LINES)
+                )
+            })
+            .collect::<Vec<_>>()
+            .join("\n")
+    }
+
+    fn gateway_mut(&mut self, node_id: u16) -> &mut GatewayProcess {
+        self.gateways
+            .iter_mut()
+            .find(|gateway| gateway.node_id == node_id)
+            .expect("a listed gateway")
+    }
+
+    /// A client signed in to the first running node that lets it in.
+    async fn client(&self) -> Option<IggyClient> {
+        let running = (0..self.cluster.node_count()).filter(|node| 
self.cluster.is_running(*node));
+        for node in running {
+            let client = IggyClientBuilder::new()
+                .with_tcp()
+                .with_server_address(self.cluster.address(node))
+                
.with_auto_sign_in(AutoLogin::Enabled(Credentials::UsernamePassword(
+                    "iggy".to_string(),
+                    SecretString::from("iggy"),
+                )))
+                .build()
+                .ok()?;
+            if matches!(
+                tokio::time::timeout(QUERY_LIMIT, client.connect()).await,
+                Ok(Ok(()))
+            ) {
+                return Some(client);
+            }
+            let _ = tokio::time::timeout(QUERY_LIMIT, client.shutdown()).await;
+        }
+        None
+    }
+}
+
+/// The last `lines` lines of the file at `path`.
+pub fn tail(path: &Path, lines: usize) -> String {
+    let text = fs::read(path).map_or_else(
+        |error| format!("cannot read {}: {error}", path.display()),
+        |bytes| String::from_utf8_lossy(&bytes).into_owned(),
+    );
+    let all: Vec<&str> = text.lines().collect();
+    all[all.len().saturating_sub(lines)..].join("\n")
+}
+
+/// Ends the session of `client` in Iggy, so the clients table keeps no slot 
for it.
+async fn sign_out(client: &IggyClient) {
+    let _ = tokio::time::timeout(QUERY_LIMIT, client.logout_user()).await;
+    let _ = tokio::time::timeout(QUERY_LIMIT, client.shutdown()).await;
+}
+
+/// One `iggy-gateway-kafka` process, started again on the same port after a 
stop.
+struct GatewayProcess {
+    node_id: u16,
+    addr: SocketAddr,
+    iggy_addr: String,
+    brokers: String,
+    log: PathBuf,
+    child: Option<Child>,
+    _port: PortGuard,
+}
+
+impl GatewayProcess {
+    fn spawn(&mut self) {
+        assert!(self.child.is_none(), "gateway {} runs", self.node_id);
+        let log = OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(&self.log)
+            .expect("open the gateway log");
+        let mut command = 
Command::new(env!("CARGO_BIN_EXE_iggy-gateway-kafka"));
+        // The gateway refuses to start on an unknown inherited IGGY_KAFKA_* 
variable, and a known
+        // one would change what the test runs.
+        command.env_clear();
+        if let Some(llvm_profile_file) = std::env::var_os("LLVM_PROFILE_FILE") 
{
+            command.env("LLVM_PROFILE_FILE", llvm_profile_file);
+        }
+        let child = command
+            .env("IGGY_KAFKA_BIND_ADDR", self.addr.to_string())
+            .env("IGGY_KAFKA_BRIDGE_ENABLED", "true")
+            .env("IGGY_KAFKA_IGGY_ADDR", &self.iggy_addr)
+            .env("IGGY_KAFKA_IGGY_USERNAME", "iggy")
+            .env("IGGY_KAFKA_IGGY_PASSWORD", "iggy")
+            .env("IGGY_KAFKA_NODE_ID", self.node_id.to_string())
+            .env("IGGY_KAFKA_BROKERS", &self.brokers)
+            .env("IGGY_KAFKA_COORDINATOR_LOG", "true")

Review Comment:
   Fixed, as in the reply to spetz on this line. The gateways run with
   `IGGY_KAFKA_IDEMPOTENT_PRODUCERS=true`. Each crash now also checks that Iggy 
answered a resend of
   a written batch with its first offset.
   



##########
gateways/kafka/tests/cluster_failover_tests.rs:
##########
@@ -0,0 +1,1340 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Three gateways beside three Iggy nodes, under load from real Kafka 
clients, while gateways and
+//! nodes stop, crash and come back.
+//!
+//! The stack is real end to end. Three `iggy-server` processes form a 
cluster, the compiled
+//! `iggy-gateway-kafka` runs beside each node, and the clients run in Docker. 
An idempotent Java
+//! producer writes numbered records through the three gateways, and a Java 
consumer group of two
+//! members reads them. At the end, kcat reads back what each partition stores.
+//!
+//! A probe keeps a member of a second group alive with a heartbeat to every 
gateway, five times a
+//! second, and asks every gateway to describe the Java group. Both groups 
hash to one coordinator
+//! partition, so one gateway coordinates both.
+//!
+//! Each test checks the same things at the end. Every acknowledged record is 
stored once, at the
+//! offset that the producer got, and the group reads it once. The group does 
not rebalance, and no
+//! two gateways answer as coordinator of a group at once.
+//!
+//! Prerequisites are Docker, the two client images and an `iggy-server` built 
into the same target
+//! directory. Missing any skips, unless `KAFKA_E2E_REQUIRED=1` is set, as in
+//! `kafka_client_e2e_tests.rs`.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/deployment.rs"]
+mod deployment;
+#[path = "common/docker_clients.rs"]
+mod docker_clients;
+#[path = "common/gateways.rs"]
+mod gateways;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::fs;
+use std::future::Future;
+use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::{
+    DescribeGroupsRequest, DescribeGroupsResponse, FindCoordinatorRequest, 
FindCoordinatorResponse,
+    GroupId, HeartbeatRequest, HeartbeatResponse, MetadataRequest, 
MetadataResponse,
+};
+use kafka_protocol::protocol::{Decodable, StrBytes};
+use serde_json::Value;
+use serial_test::serial;
+use tempfile::TempDir;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, MissedTickBehavior};
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{COORDINATOR_LOG_TOPIC, cluster_id_stream};
+use iggy_gateway_kafka::coordinator::{PARTITIONS as COORDINATOR_PARTITIONS, 
partition_of_group};
+use iggy_gateway_kafka::protocol::api::{
+    API_KEY_DESCRIBE_GROUPS, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, 
API_KEY_METADATA,
+    ERROR_FENCED_INSTANCE_ID, ERROR_ILLEGAL_GENERATION, ERROR_NONE, 
ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_UNKNOWN_MEMBER_ID,
+};
+
+use deployment::{Deployment, KAFKA_STREAM, NODES, TAIL_LINES, tail};
+use docker_clients::{Container, missing_prerequisite, run_to_end, skip};
+
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+
+/// The topic of the load. Two partitions per node, so a stop of any node 
moves primaries.
+const TOPIC: &str = "failover";
+const PARTITIONS: u32 = 6;
+
+/// The Java consumer group, two members of `kafka-verifiable-consumer.sh`.
+const GROUP: &str = "failover-readers";
+const MEMBERS: usize = 2;
+
+/// Records per second that the producer writes, with keys that spread them 
over every partition.
+const RATE: u32 = 100;
+const KEYS: &str = "60";
+
+/// The idempotent Java producer. These are the defaults since Kafka 3.0, 
written out because
+/// `kafka-verifiable-producer.sh` sets `retries=0` before it reads this file. 
The delivery timeout
+/// outlasts every step, so no record expires in the producer.
+const PRODUCER_CONFIG: &str = "enable.idempotence=true\n\
+                               acks=all\n\
+                               retries=2147483647\n\
+                               max.in.flight.requests.per.connection=5\n\
+                               request.timeout.ms=10000\n\
+                               delivery.timeout.ms=300000\n";
+
+/// Acknowledgements, and reads of each member, that a step waits for, so the 
next step starts
+/// under load.
+const ACK_PROGRESS: usize = 300;
+const READ_PROGRESS: usize = 100;
+
+/// The Iggy session timeout by default: 1.2 x the 30 s `heartbeat.interval`.
+const SESSION_TIMEOUT: Duration = Duration::from_secs(36);
+
+/// Server settings for the shortest session timeout that a 1 s report 
interval allows. The server
+/// needs 1.2 x `heartbeat.interval` above three report intervals plus 6.2 s, 
so 8 s is the next
+/// whole second, and the timeout is 9.6 s.
+const SHORT_SESSION: [(&str, &str); 2] = [
+    ("IGGY_HEARTBEAT_INTERVAL", "8s"),
+    ("IGGY_CONSUMER_GROUP_HEARTBEAT_INTERVAL", "1s"),
+];
+const SHORT_SESSION_TIMEOUT: Duration = Duration::from_millis(9_600);
+
+/// What a takeover after a crash may take beyond the session timeout: the 
report that ends the
+/// session, the lease read of the successor once a second, and its load of 
the partition.
+const TAKEOVER_SLACK: Duration = Duration::from_secs(15);
+
+/// How long the cluster takes to elect a new primary for a lost one: its 
`cluster.heartbeat_timeout`
+/// of 5 s, with an election.
+const VIEW_CHANGE: Duration = Duration::from_secs(10);
+
+/// How long the groups of a gateway may take to move after its SIGTERM. A 
successor reads its lease
+/// once a second, then loads the partition. A move that waits for the session 
timeout fails this.
+const HANDOVER_LIMIT: Duration = Duration::from_secs(15);
+
+/// How long the producer may take to write again through another gateway 
after one crashed. Below
+/// the session timeout, so the data path does not wait for it.
+const DATA_PATH_LIMIT: Duration = Duration::from_secs(20);
+
+/// How long a step may take to recover: view changes, reconnects and group 
moves.
+const RECOVERY: Duration = Duration::from_secs(90);
+
+/// How long a restarted node may take to catch up and get a primary back from 
the balancer of
+/// #4619. It moves one primary at a time, after the cluster is stable for 5 s.
+const PRIMARY_MOVE_LIMIT: Duration = Duration::from_secs(120);
+
+/// How often the probe asks every gateway, and how long one answer may take.
+const PROBE_INTERVAL: Duration = Duration::from_millis(200);
+const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// Probe rounds that a test must see, so a probe that died early cannot pass 
the checks.
+const MIN_ROUNDS: usize = 50;
+
+/// The versions of the requests the probe sends.
+const HEARTBEAT_VERSION: i16 = 2;
+const DESCRIBE_GROUPS_VERSION: i16 = 5;
+const FIND_COORDINATOR_VERSION: i16 = 4;
+const METADATA_VERSION: i16 = 9;
+
+/// The answers that make a member join again, so their arrival is a rebalance.
+const REJOIN: [i16; 4] = [
+    ERROR_ILLEGAL_GENERATION,
+    ERROR_UNKNOWN_MEMBER_ID,
+    ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_FENCED_INSTANCE_ID,
+];
+
+/// The cap of each client container, which the `timeout` inside it enforces.
+const CLIENT_LIFETIME: Duration = Duration::from_mins(10);
+
+/// How long a client may take to exit after SIGTERM, and kcat to read the 
topic.
+const CLIENT_EXIT_LIMIT: Duration = Duration::from_mins(1);
+
+/// The events of the clients that the tests count.
+const ACKED: &str = r#""name":"producer_send_success""#;
+const READ: &str = r#""name":"record_data""#;
+
+/// A rolling restart. Each gateway gets SIGTERM, as an orchestrator sends it, 
and comes back
+/// before the next one stops. Its groups move to the other gateways at once, 
with their members,
+/// so the Java group reads on without a rebalance.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_restarts_in_turn_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    for node_id in run.rotation().await {
+        run.note(&format!("SIGTERM to gateway {node_id}"));
+        let stopped_at = Instant::now();
+        run.deployment.terminate_gateway(node_id).await;
+        let moved = run.wait_served(stopped_at, node_id).await;
+        run.note(&format!("its groups moved {moved:?} after the SIGTERM"));
+        assert!(
+            moved <= HANDOVER_LIMIT,
+            "the groups of gateway {node_id} moved {moved:?} after its 
SIGTERM, above \
+             {HANDOVER_LIMIT:?}"
+        );
+        run.deployment.start_gateway(node_id).await;
+        run.wait_listed(node_id).await;
+        run.wait_for_progress(&format!("the load after gateway {node_id} came 
back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// Each Iggy node stops with SIGTERM and comes back: first the metadata 
primary, then the node
+/// that holds the primary of the group's coordinator partition, then each 
node not stopped yet.
+/// Once a node is back, the balancer of #4619 moves partition primaries to 
it, under load.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_node_stops_in_turn_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    let mut stopped = Vec::with_capacity(NODES + 1);
+    while (0..NODES).any(|node| !stopped.contains(&node)) {
+        let (node, role) = run.next_node_to_stop(&stopped).await;
+        run.note(&format!("SIGTERM to node {node}, {role}"));
+        run.deployment.cluster.stop(node);
+        stopped.push(node);
+        run.wait_no_primary_on(node).await;
+        run.wait_for_progress(&format!("the load while node {node} is down"))
+            .await;
+        run.deployment.cluster.start(node).await;
+        run.wait_primary_on(node).await;
+        run.wait_for_progress(&format!("the load after node {node} came back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// Each gateway stops together with its node, with SIGTERM to both, as when 
their machine shuts
+/// down. Both come back before the next pair stops.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_gateway_stops_with_its_node_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    for node_id in run.rotation().await {
+        let node = Deployment::node_of(node_id);
+        run.note(&format!("SIGTERM to gateway {node_id} and node {node}"));
+        let stopped_at = Instant::now();
+        run.deployment.terminate_gateway_and_node(node_id).await;
+        let moved = run.wait_served(stopped_at, node_id).await;
+        run.note(&format!("its groups moved {moved:?} after the SIGTERM"));
+        // A coordinator partition with its primary on the stopped node waits 
for a view change.
+        assert!(
+            moved <= HANDOVER_LIMIT,
+            "the groups of gateway {node_id} moved {moved:?} after its 
SIGTERM, above \
+             {HANDOVER_LIMIT:?}"
+        );
+        run.wait_no_primary_on(node).await;
+        run.wait_for_progress(&format!("the load while gateway {node_id} is 
down"))
+            .await;
+        run.deployment.cluster.start(node).await;
+        run.deployment.start_gateway(node_id).await;
+        run.wait_listed(node_id).await;
+        run.wait_primary_on(node).await;
+        run.wait_for_progress(&format!("the load after gateway {node_id} came 
back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// The gateway that coordinates the group crashes, with SIGKILL. The producer 
writes on through
+/// the other gateways at once. The group moves only when Iggy ends the 
sessions of the dead
+/// gateway, after the default session timeout of 36 s, and keeps its members 
and generation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_the_coordinator_crashes_should_move_its_groups_after_the_session_timeout()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::Gateway).await;
+    assert!(
+        takeover <= SESSION_TIMEOUT + TAKEOVER_SLACK,
+        "the takeover took {takeover:?}, above the session timeout 
{SESSION_TIMEOUT:?} and \
+         {TAKEOVER_SLACK:?}"
+    );
+    run.finish().await.assert_all();
+}
+
+/// As the crash above, on nodes with the shortest session timeout that a 1 s
+/// `consumer_group.heartbeat_interval` allows: 9.6 s, from an 8 s 
`heartbeat.interval`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_a_short_session_timeout_when_the_coordinator_crashes_should_move_its_groups_within_it()
+ {
+    let Some(mut run) = Run::start(&SHORT_SESSION).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::Gateway).await;
+    assert!(
+        takeover <= SHORT_SESSION_TIMEOUT + TAKEOVER_SLACK,
+        "the takeover took {takeover:?}, above the session timeout 
{SHORT_SESSION_TIMEOUT:?} \
+         and {TAKEOVER_SLACK:?}"
+    );
+    run.finish().await.assert_all();
+}
+
+/// The gateway that coordinates the group crashes together with its node, as 
when their machine
+/// loses power. When that node held the metadata primary, a view change comes 
first, and the new
+/// primary starts the session timeout again. The node recovers from its log 
when it comes back.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_a_short_session_timeout_when_the_coordinator_crashes_with_its_node_should_move_its_groups()
+ {
+    let Some(mut run) = Run::start(&SHORT_SESSION).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::GatewayAndNode).await;
+    assert!(
+        takeover <= SHORT_SESSION_TIMEOUT + VIEW_CHANGE + TAKEOVER_SLACK,
+        "the takeover took {takeover:?}, above the session timeout 
{SHORT_SESSION_TIMEOUT:?}, \
+         a view change {VIEW_CHANGE:?} and {TAKEOVER_SLACK:?}"
+    );
+    run.finish().await.assert_all();
+}
+
+/// What crashes with SIGKILL in a crash test.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Crash {
+    Gateway,
+    /// The gateway and the node beside it.
+    GatewayAndNode,
+}
+
+/// One test's deployment, under load from the clients, watched by the probe.
+struct Run {
+    deployment: Deployment,
+    producer: Container,
+    members: Vec<Container>,
+    probe: Probe,
+    /// The lines each member printed when the group settled. An assignment 
after that is a
+    /// rebalance.
+    settled: Vec<usize>,
+    started: Instant,
+    /// Last, so the processes stop before a failed test reports its logs.
+    work_dir: WorkDir,
+}
+
+impl Run {
+    /// Deploys the nodes with `settings` and their gateways, settles the 
group, then starts the
+    /// probe and the producer. `None` when the host cannot run the clients.
+    async fn start(settings: &[(&str, &str)]) -> Option<Self> {
+        if let Some(reason) =
+            missing_prerequisite(&[KAFKA_IMAGE, 
KCAT_IMAGE]).or_else(missing_server)
+        {
+            skip(&reason);
+            return None;
+        }
+        let started = Instant::now();
+        let work_dir = WorkDir::new();
+        let deployment = Deployment::start(work_dir.path(), settings, 
&[(TOPIC, PARTITIONS)]).await;
+        note(started, "deployed three nodes and three gateways");
+        let bootstrap = deployment.bootstrap();
+        let logs = deployment.log_dir().to_path_buf();
+        let members: Vec<Container> = (1..=MEMBERS)
+            .map(|member| start_member(&bootstrap, 
&logs.join(format!("member-{member}"))))
+            .collect();
+        let settled = settle(&members).await;
+        note(started, "the group settled");
+        let probe = Probe::start(deployment.addrs()).await;
+        let producer = start_producer(&bootstrap, &logs);
+        let run = Self {
+            deployment,
+            producer,
+            members,
+            probe,
+            settled,
+            started,
+            work_dir,
+        };
+        run.wait_for_progress("the first records").await;
+        run.note("the load runs");
+        Some(run)
+    }
+
+    fn note(&self, what: &str) {
+        note(self.started, what);
+    }
+
+    /// The gateways in the order that a rolling restart takes them: the 
coordinator of the group
+    /// first, so that the group moves at the first step.
+    async fn rotation(&self) -> Vec<u16> {
+        let first = self.coordinator().await;
+        let mut ids = self.deployment.gateway_ids();
+        let at = ids
+            .iter()
+            .position(|node_id| *node_id == first)
+            .expect("the coordinator is a gateway");
+        ids.rotate_left(at);
+        ids
+    }
+
+    /// The gateway that coordinates the group, once every running gateway 
names the same one.
+    async fn coordinator(&self) -> u16 {
+        let addrs = self.deployment.running_addrs();
+        wait_until(
+            RECOVERY,
+            PROBE_INTERVAL,
+            "agreement on the coordinator",
+            || async {
+                let mut named = BTreeSet::new();
+                for addr in &addrs {
+                    named.insert(named_coordinator(*addr, GROUP).await?);
+                }
+                let node_id = *named.first()?;
+                (named.len() == 1).then_some(u16::try_from(node_id).ok()?)
+            },
+        )
+        .await
+    }
+
+    /// The node to stop next, with its role: first the metadata primary, then 
the node that holds
+    /// the primary of the group's coordinator partition, then each node not 
stopped yet.
+    async fn next_node_to_stop(&self, stopped: &[usize]) -> (usize, &'static 
str) {
+        if stopped.is_empty() {
+            let primary = wait_until(RECOVERY, PROBE_INTERVAL, "a metadata 
primary", || {
+                self.deployment.metadata_primary()
+            })
+            .await;
+            return (primary, "the metadata primary");
+        }
+        if stopped.len() == 1 {
+            let stream = cluster_id_stream(KAFKA_STREAM);
+            let partition = partition_of_group(GROUP);
+            let primary = wait_until(
+                RECOVERY,
+                PROBE_INTERVAL,
+                "the primary of the coordinator partition",
+                || async {
+                    let primaries = self
+                        .deployment
+                        .primaries(&stream, COORDINATOR_LOG_TOPIC, [partition])
+                        .await?;
+                    primaries.first().copied()
+                },
+            )
+            .await;
+            return (primary, "the primary of the coordinator partition");
+        }
+        let node = (0..NODES)
+            .find(|node| !stopped.contains(node))
+            .expect("a node not stopped yet");
+        (node, "a node not stopped yet")
+    }
+
+    /// Kills the gateway that coordinates the group, with its node for 
[`Crash::GatewayAndNode`],
+    /// and restarts what it killed once another gateway took the group over. 
Returns how long
+    /// after the kill the takeover came.
+    async fn crash_coordinator(&mut self, crash: Crash) -> Duration {
+        let node_id = self.coordinator().await;
+        let node = Deployment::node_of(node_id);
+        self.note(&format!(
+            "SIGKILL to gateway {node_id}, the coordinator ({crash:?})"
+        ));
+        let crashed_at = Instant::now();
+        self.deployment.kill_gateway(node_id);
+        // A lost metadata primary stalls every partition until the next one 
is elected, since a
+        // node refuses partition requests on metadata evidence older than 8 s.
+        let data_path_limit = match crash {
+            Crash::Gateway => DATA_PATH_LIMIT,
+            Crash::GatewayAndNode => {
+                self.deployment.cluster.crash(node);
+                RECOVERY
+            }
+        };
+        self.wait_for_acks("writes through the other gateways", 
data_path_limit)
+            .await;
+        let takeover = self.wait_served(crashed_at, node_id).await;
+        self.note(&format!("its groups moved {takeover:?} after the crash"));
+        if crash == Crash::GatewayAndNode {
+            self.deployment.cluster.start(node).await;
+        }
+        self.deployment.start_gateway(node_id).await;
+        self.wait_listed(node_id).await;
+        if crash == Crash::GatewayAndNode {
+            self.wait_primary_on(node).await;
+        }
+        self.wait_for_progress("the load after the crash").await;
+        takeover
+    }
+
+    /// Waits until the producer gets more acknowledgements and every member 
reads more records.
+    async fn wait_for_progress(&self, what: &str) {
+        self.wait_for_acks(what, RECOVERY).await;
+        let before: Vec<usize> = self
+            .members
+            .iter()
+            .map(|member| member.count(READ))
+            .collect();
+        wait_until(RECOVERY, PROBE_INTERVAL, what, || async {
+            self.members
+                .iter()
+                .zip(&before)
+                .all(|(member, before)| member.count(READ) >= before + 
READ_PROGRESS)
+                .then_some(())
+        })
+        .await;
+    }
+
+    /// Waits until the producer gets [`ACK_PROGRESS`] more acknowledgements, 
at most `limit`.
+    async fn wait_for_acks(&self, what: &str, limit: Duration) {
+        let before = self.producer.count(ACKED);
+        wait_until(limit, PROBE_INTERVAL, what, || async {
+            (self.producer.count(ACKED) >= before + ACK_PROGRESS).then_some(())
+        })
+        .await;
+    }
+
+    /// Waits until a gateway other than `gone` answers the probe member as 
its coordinator, in a
+    /// round that started at `since` or later. Returns how long after `since` 
that round started.
+    async fn wait_served(&self, since: Instant, gone: u16) -> Duration {
+        let what = format!("a coordinator other than gateway {gone}");
+        wait_until(RECOVERY, PROBE_INTERVAL, &what, || async {
+            self.probe
+                .first_served(since, usize::from(gone) - 1)
+                .map(|at| at.saturating_duration_since(since))
+        })
+        .await
+    }
+
+    /// Waits until every running gateway lists gateway `node_id` in 
`Metadata`. A gateway lists
+    /// another once that one owns coordinator partitions, so a restarted 
gateway is back then.
+    async fn wait_listed(&self, node_id: u16) {
+        let addrs = self.deployment.running_addrs();
+        let what = format!("every gateway listing gateway {node_id}");
+        wait_until(RECOVERY, PROBE_INTERVAL, &what, || async {
+            for addr in &addrs {
+                if !listed_brokers(*addr).await?.contains(&i32::from(node_id)) 
{
+                    return None;
+                }
+            }
+            Some(())
+        })
+        .await;
+    }
+
+    /// Waits until node `node`, which stopped, leads no partition under load: 
its partitions
+    /// changed view and have primaries elsewhere.
+    async fn wait_no_primary_on(&self, node: usize) {
+        let what = format!("new primaries for the partitions of node {node}");
+        wait_until(RECOVERY, Duration::from_secs(1), &what, || async {
+            let primaries = self.loaded_primaries().await?;
+            (!primaries.contains(&node)).then_some(())
+        })
+        .await;
+    }
+
+    /// Waits until the balancer of #4619 moves a primary of a partition under 
load to node
+    /// `node`, which came back with none. The balancer evens out the count of 
primaries over all
+    /// partitions, so the move can be of the topic or of the coordinator log.
+    async fn wait_primary_on(&self, node: usize) {
+        let what = format!("a primary move to node {node}");
+        let primaries = wait_until(
+            PRIMARY_MOVE_LIMIT,
+            Duration::from_secs(1),
+            &what,
+            || async {
+                let primaries = self.loaded_primaries().await?;
+                primaries.contains(&node).then_some(primaries)
+            },
+        )
+        .await;
+        let held: Vec<usize> = (0..NODES)
+            .map(|replica| {
+                primaries
+                    .iter()
+                    .filter(|primary| **primary == replica)
+                    .count()
+            })
+            .collect();
+        self.note(&format!(
+            "primaries of the topic and the coordinator log per node: {held:?}"
+        ));
+    }
+
+    /// The primary of each partition under load: those of the topic, then 
those of the
+    /// coordinator log.
+    async fn loaded_primaries(&self) -> Option<Vec<usize>> {
+        let mut primaries = self
+            .deployment
+            .primaries(KAFKA_STREAM, TOPIC, 0..PARTITIONS)
+            .await?;
+        let coordinator_log = self
+            .deployment
+            .primaries(
+                &cluster_id_stream(KAFKA_STREAM),
+                COORDINATOR_LOG_TOPIC,
+                0..COORDINATOR_PARTITIONS,
+            )
+            .await?;
+        primaries.extend(coordinator_log);
+        Some(primaries)
+    }
+
+    /// Stops the producer, waits until the group read everything it 
acknowledged, stops the
+    /// probe and the members, and reads back what the partitions store.
+    async fn finish(self) -> Outcome {
+        let Self {
+            deployment,
+            mut producer,
+            mut members,
+            probe,
+            settled,
+            started,
+            work_dir,
+        } = self;
+        note(started, "stopping the producer");
+        producer.signal("TERM");
+        assert!(
+            producer.wait_exit(CLIENT_EXIT_LIMIT).await,
+            "the producer exits after SIGTERM"
+        );
+        let produced = producer.with_stdout(Produced::parse);
+        wait_until(
+            RECOVERY,
+            Duration::from_secs(1),
+            "the group reading every acknowledged record",
+            || async {
+                let read: BTreeSet<u64> = members
+                    .iter()
+                    .flat_map(|member| member.with_stdout(read_values))
+                    .collect();
+                produced
+                    .acked
+                    .keys()
+                    .all(|value| read.contains(value))
+                    .then_some(())
+            },
+        )
+        .await;
+        let rounds = probe.stop().await;
+        let printed: Vec<Vec<String>> = members
+            .iter()
+            .map(|member| member.with_stdout(<[String]>::to_vec))
+            .collect();
+        for member in &members {
+            member.signal("TERM");
+        }
+        for member in &mut members {
+            assert!(
+                member.wait_exit(CLIENT_EXIT_LIMIT).await,
+                "a member exits after SIGTERM"
+            );
+        }
+        note(started, "reading the partitions back with kcat");
+        let stored = stored_records(&deployment.bootstrap()).await;
+        let listings = broker_listings(&deployment).await;
+        note(started, "done");
+        Outcome {
+            produced,
+            stored,
+            read: printed
+                .iter()
+                .flat_map(|lines| read_values(lines))
+                .collect(),
+            rebalances: rebalances(&printed, &settled),
+            rounds,
+            listings,
+            _work_dir: work_dir,
+        }
+    }
+}
+
+/// What a test saw, checked once every client stopped.
+struct Outcome {
+    produced: Produced,
+    /// Every stored record, as `(partition, offset, value)`.
+    stored: Vec<(i64, i64, u64)>,
+    /// The value of each record the group read, once per read.
+    read: Vec<u64>,
+    /// The assignment events that came after the group settled.
+    rebalances: Vec<String>,
+    rounds: Vec<Round>,
+    /// What `kcat -L` printed through each gateway, with the line for each 
gateway it must list.
+    listings: Vec<(String, Vec<String>)>,
+    _work_dir: WorkDir,
+}
+
+impl Outcome {
+    fn assert_all(&self) {
+        self.assert_each_record_once();
+        self.assert_no_rebalance();
+        self.assert_one_coordinator_at_a_time();
+        self.assert_every_gateway_listed();
+    }
+
+    /// Every acknowledged record is stored once, at the partition and offset 
that the producer
+    /// got, and the group reads it once.
+    fn assert_each_record_once(&self) {
+        let errors = &self.produced.errors;
+        assert!(
+            errors.is_empty(),
+            "{} sends failed: {errors:#?}",
+            errors.len()
+        );
+        let acked = &self.produced.acked;
+        assert!(!acked.is_empty(), "the producer must write records");
+        let mut stored: BTreeMap<u64, Vec<(i64, i64)>> = BTreeMap::new();
+        for &(partition, offset, value) in &self.stored {
+            stored.entry(value).or_default().push((partition, offset));
+        }
+        let twice: Vec<_> = stored
+            .iter()
+            .filter(|(_, at)| at.len() > 1)
+            .take(20)
+            .collect();
+        assert!(
+            twice.is_empty(),
+            "records stored more than once, as value and (partition, offset): 
{twice:?}"
+        );
+        let misplaced: Vec<_> = acked
+            .iter()
+            .filter(|(value, at)| stored.get(*value).is_none_or(|stored| 
stored != &[**at]))
+            .map(|(value, at)| (value, at, stored.get(value)))
+            .take(20)
+            .collect();
+        assert!(
+            misplaced.is_empty(),
+            "acknowledged records not stored where the producer was told, as 
value, acknowledged \
+             (partition, offset) and stored ones: {misplaced:?}"
+        );
+        let mut reads: BTreeMap<u64, usize> = BTreeMap::new();
+        for value in &self.read {
+            *reads.entry(*value).or_default() += 1;
+        }
+        let read_twice: Vec<_> = reads
+            .iter()
+            .filter(|(_, count)| **count > 1)
+            .take(20)
+            .collect();
+        assert!(
+            read_twice.is_empty(),
+            "records the group read more than once, as value and reads: 
{read_twice:?}"
+        );
+        let unread: Vec<&u64> = acked
+            .keys()
+            .filter(|value| !reads.contains_key(value))
+            .take(20)
+            .collect();
+        assert!(
+            unread.is_empty(),
+            "acknowledged records the group never read: {unread:?}"
+        );
+        eprintln!("{} records acknowledged, stored and read once", 
acked.len());
+    }
+
+    /// The group keeps its members and generation: no member gets a new 
assignment, and the
+    /// coordinator describes the group as stable every time.
+    fn assert_no_rebalance(&self) {
+        assert!(
+            self.rebalances.is_empty(),
+            "the group rebalanced after it settled: {:#?}",
+            self.rebalances
+        );
+        let unstable: Vec<&(i16, String)> = self
+            .rounds
+            .iter()
+            .flat_map(|round| round.descriptions.iter().flatten())
+            .filter(|(code, state)| *code == ERROR_NONE && state != "Stable")
+            .collect();
+        assert!(
+            unstable.is_empty(),
+            "the coordinator described the group as not stable: {unstable:?}"
+        );
+    }
+
+    /// No two gateways answer as coordinator of a group in one probe round, 
and no answer makes
+    /// the probe member join again.
+    fn assert_one_coordinator_at_a_time(&self) {
+        assert!(
+            self.rounds.len() >= MIN_ROUNDS,
+            "the probe ran only {} rounds",
+            self.rounds.len()
+        );
+        let coordinating = |answers: &[Option<i16>]| {
+            answers
+                .iter()
+                .filter(|code| **code == Some(ERROR_NONE))
+                .count()
+        };
+        for round in &self.rounds {
+            let described: Vec<Option<i16>> = round
+                .descriptions
+                .iter()
+                .map(|answer| answer.as_ref().map(|(code, _)| *code))
+                .collect();
+            assert!(
+                coordinating(&round.heartbeats) <= 1 && 
coordinating(&described) <= 1,
+                "two gateways answered as coordinator at once: {round:?}"
+            );
+            assert!(
+                !round
+                    .heartbeats
+                    .iter()
+                    .flatten()
+                    .any(|code| REJOIN.contains(code)),
+                "a gateway told the probe member to join again: {round:?}"
+            );
+        }
+    }
+
+    /// Each gateway lists all three at the end, as kcat sees it.
+    fn assert_every_gateway_listed(&self) {
+        for (listing, expected) in &self.listings {
+            for line in expected {
+                assert!(
+                    listing.contains(line.as_str()),
+                    "`kcat -L` must show `{line}`:\n{listing}"
+                );
+            }
+        }
+    }
+}
+
+/// What the producer printed.
+#[derive(Default)]
+struct Produced {
+    /// Where each acknowledged record went, by value, as `(partition, 
offset)`.
+    acked: BTreeMap<u64, (i64, i64)>,
+    errors: Vec<String>,
+}
+
+impl Produced {
+    fn parse(lines: &[String]) -> Self {
+        let mut produced = Self::default();
+        for line in lines {
+            let Some(event) = parse_event(line) else {
+                continue;
+            };
+            match event["name"].as_str() {
+                Some("producer_send_success") => {
+                    if let (Some(value), Some(partition), Some(offset)) = (
+                        event["value"].as_str().and_then(|value| 
value.parse().ok()),
+                        event["partition"].as_i64(),
+                        event["offset"].as_i64(),
+                    ) {
+                        produced.acked.insert(value, (partition, offset));
+                    }
+                }
+                Some("producer_send_error") => 
produced.errors.push(line.clone()),
+                _ => {}
+            }
+        }
+        produced
+    }
+}
+
+/// The answers of one probe round, one per gateway in node id order. `None` 
where a gateway did
+/// not answer.
+#[derive(Debug, Clone)]
+struct Round {
+    at: Instant,
+    /// The error code of the probe member's heartbeat.
+    heartbeats: Vec<Option<i16>>,
+    /// The error code and state of the Java group in `DescribeGroups`.
+    descriptions: Vec<Option<(i16, String)>>,
+}
+
+/// A member of a second group that heartbeats at every gateway, and asks 
every gateway to describe
+/// the Java group, until it stops.
+struct Probe {
+    rounds: Arc<Mutex<Vec<Round>>>,
+    stop: CancellationToken,
+    task: JoinHandle<()>,
+}
+
+impl Probe {
+    /// Joins a member to a group in the coordinator partition of the Java 
group, then starts the
+    /// rounds.
+    async fn start(addrs: Vec<SocketAddr>) -> Self {
+        let group = probe_group();
+        let coordinator = wait_until(
+            RECOVERY,
+            PROBE_INTERVAL,
+            "the coordinator of the probe group",
+            || coordinator_addr(&addrs, &group),
+        )
+        .await;
+        let (member_id, generation) = gateways::join_and_sync(coordinator, 
&group).await;
+        let rounds = Arc::new(Mutex::new(Vec::new()));
+        let stop = CancellationToken::new();
+        let member = ProbeMember {
+            group,
+            member_id,
+            generation,
+        };
+        let task = tokio::spawn(probe_rounds(
+            addrs,
+            member,
+            Arc::clone(&rounds),
+            stop.clone(),
+        ));
+        Self { rounds, stop, task }
+    }
+
+    /// When the first round at `since` or later started in which a gateway 
other than the one at
+    /// index `gone` answered the probe member as its coordinator.
+    fn first_served(&self, since: Instant, gone: usize) -> Option<Instant> {
+        self.rounds
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .iter()
+            .filter(|round| round.at >= since)
+            .find(|round| {
+                round
+                    .heartbeats
+                    .iter()
+                    .enumerate()
+                    .any(|(index, code)| index != gone && *code == 
Some(ERROR_NONE))
+            })
+            .map(|round| round.at)
+    }
+
+    async fn stop(self) -> Vec<Round> {
+        self.stop.cancel();
+        self.task.await.expect("the probe does not panic");
+        std::mem::take(&mut 
*self.rounds.lock().unwrap_or_else(PoisonError::into_inner))
+    }
+}
+
+struct ProbeMember {
+    group: String,
+    member_id: StrBytes,
+    generation: i32,
+}
+
+async fn probe_rounds(
+    addrs: Vec<SocketAddr>,
+    member: ProbeMember,
+    rounds: Arc<Mutex<Vec<Round>>>,
+    stop: CancellationToken,
+) {
+    let mut ticks = tokio::time::interval(PROBE_INTERVAL);
+    ticks.set_missed_tick_behavior(MissedTickBehavior::Delay);
+    loop {
+        tokio::select! {
+            () = stop.cancelled() => return,
+            _ = ticks.tick() => {}
+        }
+        let at = Instant::now();
+        let heartbeats = join_all(addrs.iter().map(|addr| heartbeat(*addr, 
&member))).await;
+        let descriptions = join_all(addrs.iter().map(|addr| describe(*addr, 
GROUP))).await;
+        rounds
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .push(Round {
+                at,
+                heartbeats,
+                descriptions,
+            });
+    }
+}
+
+/// A group in the coordinator partition of the Java group, so one gateway 
coordinates both.
+fn probe_group() -> String {
+    let partition = partition_of_group(GROUP);
+    // A thousand names cover the 16 partitions many times over.
+    (0..1_000)
+        .map(|n| format!("failover-probe-{n}"))
+        .find(|group| partition_of_group(group) == partition)
+        .expect("a group in every partition")
+}
+
+/// The test's directory. A failed test keeps the logs there, drops the node 
data, and prints where
+/// they are with the end of each gateway log.
+struct WorkDir(Option<TempDir>);
+
+impl WorkDir {
+    fn new() -> Self {
+        let dir = tempfile::Builder::new()
+            .prefix("iggy-kafka-failover-")
+            .tempdir()
+            .expect("create the work dir");
+        Self(Some(dir))
+    }
+
+    fn path(&self) -> &Path {
+        self.0.as_ref().expect("a work dir").path()
+    }
+}
+
+impl Drop for WorkDir {
+    fn drop(&mut self) {
+        if !std::thread::panicking() {
+            return;
+        }
+        let Some(dir) = self.0.take() else {
+            return;
+        };
+        let path = dir.keep();
+        let _ = fs::remove_dir_all(path.join("data"));
+        let logs = path.join("logs");
+        let mut report = vec![format!(
+            "the failed test kept its logs in {}",
+            logs.display()
+        )];
+        for node_id in 1..=NODES {
+            let log = logs.join(format!("gateway-{node_id}.log"));
+            report.push(format!(
+                "--- gateway {node_id} ---\n{}",
+                tail(&log, TAIL_LINES)
+            ));
+        }
+        for client in ["producer", "member-1", "member-2"] {
+            let log = logs.join(format!("{client}.err"));
+            report.push(format!("--- {client} stderr ---\n{}", tail(&log, 
10)));
+        }
+        eprintln!("{}", report.join("\n"));
+    }
+}
+
+fn note(started: Instant, what: &str) {
+    eprintln!("[{:>6.1} s] {what}", started.elapsed().as_secs_f64());
+}
+
+/// Why the cluster cannot start, or `None` when it can: `iggy-server` must be 
built into the
+/// target directory of this test binary.
+fn missing_server() -> Option<String> {
+    let binary = format!("iggy-server{}", std::env::consts::EXE_SUFFIX);
+    let built = std::env::current_exe()
+        .ok()?
+        .ancestors()
+        .any(|dir| dir.join(&binary).is_file());
+    (!built).then(|| "iggy-server is not built; run `cargo build --bin 
iggy-server` first".into())
+}
+
+/// Polls `check` every `interval` until it returns a value, and fails the 
test after `limit`.
+async fn wait_until<T, F: Future<Output = Option<T>>>(
+    limit: Duration,
+    interval: Duration,
+    what: &str,
+    mut check: impl FnMut() -> F,
+) -> T {
+    let deadline = Instant::now() + limit;
+    loop {
+        if let Some(value) = check().await {
+            return value;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "{what} did not happen within {limit:?}"
+        );
+        tokio::time::sleep(interval).await;
+    }
+}
+
+/// A member of the group, which commits each batch after it reads it, prints 
a JSON event for each
+/// step and one for each record.
+fn start_member(bootstrap: &str, log: &Path) -> Container {

Review Comment:
   Fixed. The suite fails on any `offsets_committed` event with `success: 
false`. After each SIGTERM,
   node stop and crash takeover, it compares OffsetFetch from the current 
coordinator with the
   acknowledged commits. At the end the coordinator must return the last commit 
of each partition,
   and that must be the last read plus one.
   



##########
gateways/kafka/tests/cluster_failover_tests.rs:
##########
@@ -0,0 +1,1340 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Three gateways beside three Iggy nodes, under load from real Kafka 
clients, while gateways and
+//! nodes stop, crash and come back.
+//!
+//! The stack is real end to end. Three `iggy-server` processes form a 
cluster, the compiled
+//! `iggy-gateway-kafka` runs beside each node, and the clients run in Docker. 
An idempotent Java
+//! producer writes numbered records through the three gateways, and a Java 
consumer group of two
+//! members reads them. At the end, kcat reads back what each partition stores.
+//!
+//! A probe keeps a member of a second group alive with a heartbeat to every 
gateway, five times a
+//! second, and asks every gateway to describe the Java group. Both groups 
hash to one coordinator
+//! partition, so one gateway coordinates both.
+//!
+//! Each test checks the same things at the end. Every acknowledged record is 
stored once, at the
+//! offset that the producer got, and the group reads it once. The group does 
not rebalance, and no
+//! two gateways answer as coordinator of a group at once.
+//!
+//! Prerequisites are Docker, the two client images and an `iggy-server` built 
into the same target
+//! directory. Missing any skips, unless `KAFKA_E2E_REQUIRED=1` is set, as in
+//! `kafka_client_e2e_tests.rs`.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/deployment.rs"]
+mod deployment;
+#[path = "common/docker_clients.rs"]
+mod docker_clients;
+#[path = "common/gateways.rs"]
+mod gateways;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::fs;
+use std::future::Future;
+use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::{
+    DescribeGroupsRequest, DescribeGroupsResponse, FindCoordinatorRequest, 
FindCoordinatorResponse,
+    GroupId, HeartbeatRequest, HeartbeatResponse, MetadataRequest, 
MetadataResponse,
+};
+use kafka_protocol::protocol::{Decodable, StrBytes};
+use serde_json::Value;
+use serial_test::serial;
+use tempfile::TempDir;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, MissedTickBehavior};
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{COORDINATOR_LOG_TOPIC, cluster_id_stream};
+use iggy_gateway_kafka::coordinator::{PARTITIONS as COORDINATOR_PARTITIONS, 
partition_of_group};
+use iggy_gateway_kafka::protocol::api::{
+    API_KEY_DESCRIBE_GROUPS, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, 
API_KEY_METADATA,
+    ERROR_FENCED_INSTANCE_ID, ERROR_ILLEGAL_GENERATION, ERROR_NONE, 
ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_UNKNOWN_MEMBER_ID,
+};
+
+use deployment::{Deployment, KAFKA_STREAM, NODES, TAIL_LINES, tail};
+use docker_clients::{Container, missing_prerequisite, run_to_end, skip};
+
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+
+/// The topic of the load. Two partitions per node, so a stop of any node 
moves primaries.
+const TOPIC: &str = "failover";
+const PARTITIONS: u32 = 6;
+
+/// The Java consumer group, two members of `kafka-verifiable-consumer.sh`.
+const GROUP: &str = "failover-readers";
+const MEMBERS: usize = 2;
+
+/// Records per second that the producer writes, with keys that spread them 
over every partition.
+const RATE: u32 = 100;
+const KEYS: &str = "60";
+
+/// The idempotent Java producer. These are the defaults since Kafka 3.0, 
written out because
+/// `kafka-verifiable-producer.sh` sets `retries=0` before it reads this file. 
The delivery timeout
+/// outlasts every step, so no record expires in the producer.
+const PRODUCER_CONFIG: &str = "enable.idempotence=true\n\
+                               acks=all\n\
+                               retries=2147483647\n\
+                               max.in.flight.requests.per.connection=5\n\
+                               request.timeout.ms=10000\n\
+                               delivery.timeout.ms=300000\n";
+
+/// Acknowledgements, and reads of each member, that a step waits for, so the 
next step starts
+/// under load.
+const ACK_PROGRESS: usize = 300;
+const READ_PROGRESS: usize = 100;
+
+/// The Iggy session timeout by default: 1.2 x the 30 s `heartbeat.interval`.
+const SESSION_TIMEOUT: Duration = Duration::from_secs(36);
+
+/// Server settings for the shortest session timeout that a 1 s report 
interval allows. The server
+/// needs 1.2 x `heartbeat.interval` above three report intervals plus 6.2 s, 
so 8 s is the next
+/// whole second, and the timeout is 9.6 s.
+const SHORT_SESSION: [(&str, &str); 2] = [
+    ("IGGY_HEARTBEAT_INTERVAL", "8s"),
+    ("IGGY_CONSUMER_GROUP_HEARTBEAT_INTERVAL", "1s"),
+];
+const SHORT_SESSION_TIMEOUT: Duration = Duration::from_millis(9_600);
+
+/// What a takeover after a crash may take beyond the session timeout: the 
report that ends the
+/// session, the lease read of the successor once a second, and its load of 
the partition.
+const TAKEOVER_SLACK: Duration = Duration::from_secs(15);
+
+/// How long the cluster takes to elect a new primary for a lost one: its 
`cluster.heartbeat_timeout`
+/// of 5 s, with an election.
+const VIEW_CHANGE: Duration = Duration::from_secs(10);
+
+/// How long the groups of a gateway may take to move after its SIGTERM. A 
successor reads its lease
+/// once a second, then loads the partition. A move that waits for the session 
timeout fails this.
+const HANDOVER_LIMIT: Duration = Duration::from_secs(15);
+
+/// How long the producer may take to write again through another gateway 
after one crashed. Below
+/// the session timeout, so the data path does not wait for it.
+const DATA_PATH_LIMIT: Duration = Duration::from_secs(20);
+
+/// How long a step may take to recover: view changes, reconnects and group 
moves.
+const RECOVERY: Duration = Duration::from_secs(90);
+
+/// How long a restarted node may take to catch up and get a primary back from 
the balancer of
+/// #4619. It moves one primary at a time, after the cluster is stable for 5 s.
+const PRIMARY_MOVE_LIMIT: Duration = Duration::from_secs(120);
+
+/// How often the probe asks every gateway, and how long one answer may take.
+const PROBE_INTERVAL: Duration = Duration::from_millis(200);
+const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// Probe rounds that a test must see, so a probe that died early cannot pass 
the checks.
+const MIN_ROUNDS: usize = 50;
+
+/// The versions of the requests the probe sends.
+const HEARTBEAT_VERSION: i16 = 2;
+const DESCRIBE_GROUPS_VERSION: i16 = 5;
+const FIND_COORDINATOR_VERSION: i16 = 4;
+const METADATA_VERSION: i16 = 9;
+
+/// The answers that make a member join again, so their arrival is a rebalance.
+const REJOIN: [i16; 4] = [
+    ERROR_ILLEGAL_GENERATION,
+    ERROR_UNKNOWN_MEMBER_ID,
+    ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_FENCED_INSTANCE_ID,
+];
+
+/// The cap of each client container, which the `timeout` inside it enforces.
+const CLIENT_LIFETIME: Duration = Duration::from_mins(10);
+
+/// How long a client may take to exit after SIGTERM, and kcat to read the 
topic.
+const CLIENT_EXIT_LIMIT: Duration = Duration::from_mins(1);
+
+/// The events of the clients that the tests count.
+const ACKED: &str = r#""name":"producer_send_success""#;
+const READ: &str = r#""name":"record_data""#;
+
+/// A rolling restart. Each gateway gets SIGTERM, as an orchestrator sends it, 
and comes back
+/// before the next one stops. Its groups move to the other gateways at once, 
with their members,
+/// so the Java group reads on without a rebalance.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_restarts_in_turn_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    for node_id in run.rotation().await {
+        run.note(&format!("SIGTERM to gateway {node_id}"));
+        let stopped_at = Instant::now();
+        run.deployment.terminate_gateway(node_id).await;
+        let moved = run.wait_served(stopped_at, node_id).await;

Review Comment:
   Fixed. The probe holds a member in a group of each coordinator partition. A 
step waits until
   another gateway answers for every partition that the stopped gateway 
answered for. Each step also
   stops the current coordinator of the Java group first, among the gateways 
that did not stop yet.
   Measured moves are now 0.55 to 1.25 s after a SIGTERM, and 5.9 to 7.0 s with 
the node stopped too.
   



##########
gateways/kafka/tests/cluster_failover_tests.rs:
##########
@@ -0,0 +1,1340 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Three gateways beside three Iggy nodes, under load from real Kafka 
clients, while gateways and
+//! nodes stop, crash and come back.
+//!
+//! The stack is real end to end. Three `iggy-server` processes form a 
cluster, the compiled
+//! `iggy-gateway-kafka` runs beside each node, and the clients run in Docker. 
An idempotent Java
+//! producer writes numbered records through the three gateways, and a Java 
consumer group of two
+//! members reads them. At the end, kcat reads back what each partition stores.
+//!
+//! A probe keeps a member of a second group alive with a heartbeat to every 
gateway, five times a
+//! second, and asks every gateway to describe the Java group. Both groups 
hash to one coordinator
+//! partition, so one gateway coordinates both.
+//!
+//! Each test checks the same things at the end. Every acknowledged record is 
stored once, at the
+//! offset that the producer got, and the group reads it once. The group does 
not rebalance, and no
+//! two gateways answer as coordinator of a group at once.
+//!
+//! Prerequisites are Docker, the two client images and an `iggy-server` built 
into the same target
+//! directory. Missing any skips, unless `KAFKA_E2E_REQUIRED=1` is set, as in
+//! `kafka_client_e2e_tests.rs`.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/deployment.rs"]
+mod deployment;
+#[path = "common/docker_clients.rs"]
+mod docker_clients;
+#[path = "common/gateways.rs"]
+mod gateways;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::fs;
+use std::future::Future;
+use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::{
+    DescribeGroupsRequest, DescribeGroupsResponse, FindCoordinatorRequest, 
FindCoordinatorResponse,
+    GroupId, HeartbeatRequest, HeartbeatResponse, MetadataRequest, 
MetadataResponse,
+};
+use kafka_protocol::protocol::{Decodable, StrBytes};
+use serde_json::Value;
+use serial_test::serial;
+use tempfile::TempDir;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, MissedTickBehavior};
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{COORDINATOR_LOG_TOPIC, cluster_id_stream};
+use iggy_gateway_kafka::coordinator::{PARTITIONS as COORDINATOR_PARTITIONS, 
partition_of_group};
+use iggy_gateway_kafka::protocol::api::{
+    API_KEY_DESCRIBE_GROUPS, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, 
API_KEY_METADATA,
+    ERROR_FENCED_INSTANCE_ID, ERROR_ILLEGAL_GENERATION, ERROR_NONE, 
ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_UNKNOWN_MEMBER_ID,
+};
+
+use deployment::{Deployment, KAFKA_STREAM, NODES, TAIL_LINES, tail};
+use docker_clients::{Container, missing_prerequisite, run_to_end, skip};
+
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+
+/// The topic of the load. Two partitions per node, so a stop of any node 
moves primaries.
+const TOPIC: &str = "failover";
+const PARTITIONS: u32 = 6;
+
+/// The Java consumer group, two members of `kafka-verifiable-consumer.sh`.
+const GROUP: &str = "failover-readers";
+const MEMBERS: usize = 2;
+
+/// Records per second that the producer writes, with keys that spread them 
over every partition.
+const RATE: u32 = 100;
+const KEYS: &str = "60";
+
+/// The idempotent Java producer. These are the defaults since Kafka 3.0, 
written out because
+/// `kafka-verifiable-producer.sh` sets `retries=0` before it reads this file. 
The delivery timeout
+/// outlasts every step, so no record expires in the producer.
+const PRODUCER_CONFIG: &str = "enable.idempotence=true\n\
+                               acks=all\n\
+                               retries=2147483647\n\
+                               max.in.flight.requests.per.connection=5\n\
+                               request.timeout.ms=10000\n\
+                               delivery.timeout.ms=300000\n";
+
+/// Acknowledgements, and reads of each member, that a step waits for, so the 
next step starts
+/// under load.
+const ACK_PROGRESS: usize = 300;
+const READ_PROGRESS: usize = 100;
+
+/// The Iggy session timeout by default: 1.2 x the 30 s `heartbeat.interval`.
+const SESSION_TIMEOUT: Duration = Duration::from_secs(36);
+
+/// Server settings for the shortest session timeout that a 1 s report 
interval allows. The server
+/// needs 1.2 x `heartbeat.interval` above three report intervals plus 6.2 s, 
so 8 s is the next
+/// whole second, and the timeout is 9.6 s.
+const SHORT_SESSION: [(&str, &str); 2] = [
+    ("IGGY_HEARTBEAT_INTERVAL", "8s"),
+    ("IGGY_CONSUMER_GROUP_HEARTBEAT_INTERVAL", "1s"),
+];
+const SHORT_SESSION_TIMEOUT: Duration = Duration::from_millis(9_600);
+
+/// What a takeover after a crash may take beyond the session timeout: the 
report that ends the
+/// session, the lease read of the successor once a second, and its load of 
the partition.
+const TAKEOVER_SLACK: Duration = Duration::from_secs(15);
+
+/// How long the cluster takes to elect a new primary for a lost one: its 
`cluster.heartbeat_timeout`
+/// of 5 s, with an election.
+const VIEW_CHANGE: Duration = Duration::from_secs(10);
+
+/// How long the groups of a gateway may take to move after its SIGTERM. A 
successor reads its lease
+/// once a second, then loads the partition. A move that waits for the session 
timeout fails this.
+const HANDOVER_LIMIT: Duration = Duration::from_secs(15);
+
+/// How long the producer may take to write again through another gateway 
after one crashed. Below
+/// the session timeout, so the data path does not wait for it.
+const DATA_PATH_LIMIT: Duration = Duration::from_secs(20);
+
+/// How long a step may take to recover: view changes, reconnects and group 
moves.
+const RECOVERY: Duration = Duration::from_secs(90);
+
+/// How long a restarted node may take to catch up and get a primary back from 
the balancer of
+/// #4619. It moves one primary at a time, after the cluster is stable for 5 s.
+const PRIMARY_MOVE_LIMIT: Duration = Duration::from_secs(120);
+
+/// How often the probe asks every gateway, and how long one answer may take.
+const PROBE_INTERVAL: Duration = Duration::from_millis(200);
+const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// Probe rounds that a test must see, so a probe that died early cannot pass 
the checks.
+const MIN_ROUNDS: usize = 50;
+
+/// The versions of the requests the probe sends.
+const HEARTBEAT_VERSION: i16 = 2;
+const DESCRIBE_GROUPS_VERSION: i16 = 5;
+const FIND_COORDINATOR_VERSION: i16 = 4;
+const METADATA_VERSION: i16 = 9;
+
+/// The answers that make a member join again, so their arrival is a rebalance.
+const REJOIN: [i16; 4] = [
+    ERROR_ILLEGAL_GENERATION,
+    ERROR_UNKNOWN_MEMBER_ID,
+    ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_FENCED_INSTANCE_ID,
+];
+
+/// The cap of each client container, which the `timeout` inside it enforces.
+const CLIENT_LIFETIME: Duration = Duration::from_mins(10);
+
+/// How long a client may take to exit after SIGTERM, and kcat to read the 
topic.
+const CLIENT_EXIT_LIMIT: Duration = Duration::from_mins(1);
+
+/// The events of the clients that the tests count.
+const ACKED: &str = r#""name":"producer_send_success""#;
+const READ: &str = r#""name":"record_data""#;
+
+/// A rolling restart. Each gateway gets SIGTERM, as an orchestrator sends it, 
and comes back
+/// before the next one stops. Its groups move to the other gateways at once, 
with their members,
+/// so the Java group reads on without a rebalance.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_restarts_in_turn_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    for node_id in run.rotation().await {
+        run.note(&format!("SIGTERM to gateway {node_id}"));
+        let stopped_at = Instant::now();
+        run.deployment.terminate_gateway(node_id).await;
+        let moved = run.wait_served(stopped_at, node_id).await;
+        run.note(&format!("its groups moved {moved:?} after the SIGTERM"));
+        assert!(
+            moved <= HANDOVER_LIMIT,
+            "the groups of gateway {node_id} moved {moved:?} after its 
SIGTERM, above \
+             {HANDOVER_LIMIT:?}"
+        );
+        run.deployment.start_gateway(node_id).await;
+        run.wait_listed(node_id).await;
+        run.wait_for_progress(&format!("the load after gateway {node_id} came 
back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// Each Iggy node stops with SIGTERM and comes back: first the metadata 
primary, then the node
+/// that holds the primary of the group's coordinator partition, then each 
node not stopped yet.
+/// Once a node is back, the balancer of #4619 moves partition primaries to 
it, under load.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_node_stops_in_turn_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    let mut stopped = Vec::with_capacity(NODES + 1);
+    while (0..NODES).any(|node| !stopped.contains(&node)) {
+        let (node, role) = run.next_node_to_stop(&stopped).await;
+        run.note(&format!("SIGTERM to node {node}, {role}"));
+        run.deployment.cluster.stop(node);
+        stopped.push(node);
+        run.wait_no_primary_on(node).await;
+        run.wait_for_progress(&format!("the load while node {node} is down"))
+            .await;
+        run.deployment.cluster.start(node).await;
+        run.wait_primary_on(node).await;
+        run.wait_for_progress(&format!("the load after node {node} came back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// Each gateway stops together with its node, with SIGTERM to both, as when 
their machine shuts
+/// down. Both come back before the next pair stops.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_gateway_stops_with_its_node_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    for node_id in run.rotation().await {
+        let node = Deployment::node_of(node_id);
+        run.note(&format!("SIGTERM to gateway {node_id} and node {node}"));
+        let stopped_at = Instant::now();
+        run.deployment.terminate_gateway_and_node(node_id).await;
+        let moved = run.wait_served(stopped_at, node_id).await;
+        run.note(&format!("its groups moved {moved:?} after the SIGTERM"));
+        // A coordinator partition with its primary on the stopped node waits 
for a view change.
+        assert!(
+            moved <= HANDOVER_LIMIT,
+            "the groups of gateway {node_id} moved {moved:?} after its 
SIGTERM, above \
+             {HANDOVER_LIMIT:?}"
+        );
+        run.wait_no_primary_on(node).await;
+        run.wait_for_progress(&format!("the load while gateway {node_id} is 
down"))
+            .await;
+        run.deployment.cluster.start(node).await;
+        run.deployment.start_gateway(node_id).await;
+        run.wait_listed(node_id).await;
+        run.wait_primary_on(node).await;
+        run.wait_for_progress(&format!("the load after gateway {node_id} came 
back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// The gateway that coordinates the group crashes, with SIGKILL. The producer 
writes on through
+/// the other gateways at once. The group moves only when Iggy ends the 
sessions of the dead
+/// gateway, after the default session timeout of 36 s, and keeps its members 
and generation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_the_coordinator_crashes_should_move_its_groups_after_the_session_timeout()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::Gateway).await;
+    assert!(
+        takeover <= SESSION_TIMEOUT + TAKEOVER_SLACK,
+        "the takeover took {takeover:?}, above the session timeout 
{SESSION_TIMEOUT:?} and \
+         {TAKEOVER_SLACK:?}"
+    );
+    run.finish().await.assert_all();
+}
+
+/// As the crash above, on nodes with the shortest session timeout that a 1 s
+/// `consumer_group.heartbeat_interval` allows: 9.6 s, from an 8 s 
`heartbeat.interval`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_a_short_session_timeout_when_the_coordinator_crashes_should_move_its_groups_within_it()
+ {
+    let Some(mut run) = Run::start(&SHORT_SESSION).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::Gateway).await;
+    assert!(
+        takeover <= SHORT_SESSION_TIMEOUT + TAKEOVER_SLACK,

Review Comment:
   Fixed. Each crash checks a window: from the timeout minus one report 
interval, to the timeout plus
   one interval and 4 s. That is 8.6 to 14.6 s and 31 to 45 s, and 10 s more 
for a crash of the
   metadata primary. The short-session tests are renamed `..._after_it`.
   



##########
gateways/kafka/tests/cluster_failover_tests.rs:
##########
@@ -0,0 +1,1340 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Three gateways beside three Iggy nodes, under load from real Kafka 
clients, while gateways and
+//! nodes stop, crash and come back.
+//!
+//! The stack is real end to end. Three `iggy-server` processes form a 
cluster, the compiled
+//! `iggy-gateway-kafka` runs beside each node, and the clients run in Docker. 
An idempotent Java
+//! producer writes numbered records through the three gateways, and a Java 
consumer group of two
+//! members reads them. At the end, kcat reads back what each partition stores.
+//!
+//! A probe keeps a member of a second group alive with a heartbeat to every 
gateway, five times a
+//! second, and asks every gateway to describe the Java group. Both groups 
hash to one coordinator
+//! partition, so one gateway coordinates both.
+//!
+//! Each test checks the same things at the end. Every acknowledged record is 
stored once, at the
+//! offset that the producer got, and the group reads it once. The group does 
not rebalance, and no
+//! two gateways answer as coordinator of a group at once.
+//!
+//! Prerequisites are Docker, the two client images and an `iggy-server` built 
into the same target
+//! directory. Missing any skips, unless `KAFKA_E2E_REQUIRED=1` is set, as in
+//! `kafka_client_e2e_tests.rs`.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/deployment.rs"]
+mod deployment;
+#[path = "common/docker_clients.rs"]
+mod docker_clients;
+#[path = "common/gateways.rs"]
+mod gateways;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/tcp.rs"]
+mod tcp;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::fs;
+use std::future::Future;
+use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::{Arc, Mutex, PoisonError};
+use std::time::Duration;
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::{
+    DescribeGroupsRequest, DescribeGroupsResponse, FindCoordinatorRequest, 
FindCoordinatorResponse,
+    GroupId, HeartbeatRequest, HeartbeatResponse, MetadataRequest, 
MetadataResponse,
+};
+use kafka_protocol::protocol::{Decodable, StrBytes};
+use serde_json::Value;
+use serial_test::serial;
+use tempfile::TempDir;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, MissedTickBehavior};
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{COORDINATOR_LOG_TOPIC, cluster_id_stream};
+use iggy_gateway_kafka::coordinator::{PARTITIONS as COORDINATOR_PARTITIONS, 
partition_of_group};
+use iggy_gateway_kafka::protocol::api::{
+    API_KEY_DESCRIBE_GROUPS, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, 
API_KEY_METADATA,
+    ERROR_FENCED_INSTANCE_ID, ERROR_ILLEGAL_GENERATION, ERROR_NONE, 
ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_UNKNOWN_MEMBER_ID,
+};
+
+use deployment::{Deployment, KAFKA_STREAM, NODES, TAIL_LINES, tail};
+use docker_clients::{Container, missing_prerequisite, run_to_end, skip};
+
+const KAFKA_IMAGE: &str = "apache/kafka:3.9.0";
+const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1";
+
+/// The topic of the load. Two partitions per node, so a stop of any node 
moves primaries.
+const TOPIC: &str = "failover";
+const PARTITIONS: u32 = 6;
+
+/// The Java consumer group, two members of `kafka-verifiable-consumer.sh`.
+const GROUP: &str = "failover-readers";
+const MEMBERS: usize = 2;
+
+/// Records per second that the producer writes, with keys that spread them 
over every partition.
+const RATE: u32 = 100;
+const KEYS: &str = "60";
+
+/// The idempotent Java producer. These are the defaults since Kafka 3.0, 
written out because
+/// `kafka-verifiable-producer.sh` sets `retries=0` before it reads this file. 
The delivery timeout
+/// outlasts every step, so no record expires in the producer.
+const PRODUCER_CONFIG: &str = "enable.idempotence=true\n\
+                               acks=all\n\
+                               retries=2147483647\n\
+                               max.in.flight.requests.per.connection=5\n\
+                               request.timeout.ms=10000\n\
+                               delivery.timeout.ms=300000\n";
+
+/// Acknowledgements, and reads of each member, that a step waits for, so the 
next step starts
+/// under load.
+const ACK_PROGRESS: usize = 300;
+const READ_PROGRESS: usize = 100;
+
+/// The Iggy session timeout by default: 1.2 x the 30 s `heartbeat.interval`.
+const SESSION_TIMEOUT: Duration = Duration::from_secs(36);
+
+/// Server settings for the shortest session timeout that a 1 s report 
interval allows. The server
+/// needs 1.2 x `heartbeat.interval` above three report intervals plus 6.2 s, 
so 8 s is the next
+/// whole second, and the timeout is 9.6 s.
+const SHORT_SESSION: [(&str, &str); 2] = [
+    ("IGGY_HEARTBEAT_INTERVAL", "8s"),
+    ("IGGY_CONSUMER_GROUP_HEARTBEAT_INTERVAL", "1s"),
+];
+const SHORT_SESSION_TIMEOUT: Duration = Duration::from_millis(9_600);
+
+/// What a takeover after a crash may take beyond the session timeout: the 
report that ends the
+/// session, the lease read of the successor once a second, and its load of 
the partition.
+const TAKEOVER_SLACK: Duration = Duration::from_secs(15);
+
+/// How long the cluster takes to elect a new primary for a lost one: its 
`cluster.heartbeat_timeout`
+/// of 5 s, with an election.
+const VIEW_CHANGE: Duration = Duration::from_secs(10);
+
+/// How long the groups of a gateway may take to move after its SIGTERM. A 
successor reads its lease
+/// once a second, then loads the partition. A move that waits for the session 
timeout fails this.
+const HANDOVER_LIMIT: Duration = Duration::from_secs(15);
+
+/// How long the producer may take to write again through another gateway 
after one crashed. Below
+/// the session timeout, so the data path does not wait for it.
+const DATA_PATH_LIMIT: Duration = Duration::from_secs(20);
+
+/// How long a step may take to recover: view changes, reconnects and group 
moves.
+const RECOVERY: Duration = Duration::from_secs(90);
+
+/// How long a restarted node may take to catch up and get a primary back from 
the balancer of
+/// #4619. It moves one primary at a time, after the cluster is stable for 5 s.
+const PRIMARY_MOVE_LIMIT: Duration = Duration::from_secs(120);
+
+/// How often the probe asks every gateway, and how long one answer may take.
+const PROBE_INTERVAL: Duration = Duration::from_millis(200);
+const PROBE_TIMEOUT: Duration = Duration::from_secs(2);
+
+/// Probe rounds that a test must see, so a probe that died early cannot pass 
the checks.
+const MIN_ROUNDS: usize = 50;
+
+/// The versions of the requests the probe sends.
+const HEARTBEAT_VERSION: i16 = 2;
+const DESCRIBE_GROUPS_VERSION: i16 = 5;
+const FIND_COORDINATOR_VERSION: i16 = 4;
+const METADATA_VERSION: i16 = 9;
+
+/// The answers that make a member join again, so their arrival is a rebalance.
+const REJOIN: [i16; 4] = [
+    ERROR_ILLEGAL_GENERATION,
+    ERROR_UNKNOWN_MEMBER_ID,
+    ERROR_REBALANCE_IN_PROGRESS,
+    ERROR_FENCED_INSTANCE_ID,
+];
+
+/// The cap of each client container, which the `timeout` inside it enforces.
+const CLIENT_LIFETIME: Duration = Duration::from_mins(10);
+
+/// How long a client may take to exit after SIGTERM, and kcat to read the 
topic.
+const CLIENT_EXIT_LIMIT: Duration = Duration::from_mins(1);
+
+/// The events of the clients that the tests count.
+const ACKED: &str = r#""name":"producer_send_success""#;
+const READ: &str = r#""name":"record_data""#;
+
+/// A rolling restart. Each gateway gets SIGTERM, as an orchestrator sends it, 
and comes back
+/// before the next one stops. Its groups move to the other gateways at once, 
with their members,
+/// so the Java group reads on without a rebalance.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_restarts_in_turn_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    for node_id in run.rotation().await {
+        run.note(&format!("SIGTERM to gateway {node_id}"));
+        let stopped_at = Instant::now();
+        run.deployment.terminate_gateway(node_id).await;
+        let moved = run.wait_served(stopped_at, node_id).await;
+        run.note(&format!("its groups moved {moved:?} after the SIGTERM"));
+        assert!(
+            moved <= HANDOVER_LIMIT,
+            "the groups of gateway {node_id} moved {moved:?} after its 
SIGTERM, above \
+             {HANDOVER_LIMIT:?}"
+        );
+        run.deployment.start_gateway(node_id).await;
+        run.wait_listed(node_id).await;
+        run.wait_for_progress(&format!("the load after gateway {node_id} came 
back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// Each Iggy node stops with SIGTERM and comes back: first the metadata 
primary, then the node
+/// that holds the primary of the group's coordinator partition, then each 
node not stopped yet.
+/// Once a node is back, the balancer of #4619 moves partition primaries to 
it, under load.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_node_stops_in_turn_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    let mut stopped = Vec::with_capacity(NODES + 1);
+    while (0..NODES).any(|node| !stopped.contains(&node)) {
+        let (node, role) = run.next_node_to_stop(&stopped).await;
+        run.note(&format!("SIGTERM to node {node}, {role}"));
+        run.deployment.cluster.stop(node);
+        stopped.push(node);
+        run.wait_no_primary_on(node).await;
+        run.wait_for_progress(&format!("the load while node {node} is down"))
+            .await;
+        run.deployment.cluster.start(node).await;
+        run.wait_primary_on(node).await;
+        run.wait_for_progress(&format!("the load after node {node} came back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// Each gateway stops together with its node, with SIGTERM to both, as when 
their machine shuts
+/// down. Both come back before the next pair stops.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_each_gateway_stops_with_its_node_should_write_and_read_each_record_once()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    for node_id in run.rotation().await {
+        let node = Deployment::node_of(node_id);
+        run.note(&format!("SIGTERM to gateway {node_id} and node {node}"));
+        let stopped_at = Instant::now();
+        run.deployment.terminate_gateway_and_node(node_id).await;
+        let moved = run.wait_served(stopped_at, node_id).await;
+        run.note(&format!("its groups moved {moved:?} after the SIGTERM"));
+        // A coordinator partition with its primary on the stopped node waits 
for a view change.
+        assert!(
+            moved <= HANDOVER_LIMIT,
+            "the groups of gateway {node_id} moved {moved:?} after its 
SIGTERM, above \
+             {HANDOVER_LIMIT:?}"
+        );
+        run.wait_no_primary_on(node).await;
+        run.wait_for_progress(&format!("the load while gateway {node_id} is 
down"))
+            .await;
+        run.deployment.cluster.start(node).await;
+        run.deployment.start_gateway(node_id).await;
+        run.wait_listed(node_id).await;
+        run.wait_primary_on(node).await;
+        run.wait_for_progress(&format!("the load after gateway {node_id} came 
back"))
+            .await;
+    }
+    run.finish().await.assert_all();
+}
+
+/// The gateway that coordinates the group crashes, with SIGKILL. The producer 
writes on through
+/// the other gateways at once. The group moves only when Iggy ends the 
sessions of the dead
+/// gateway, after the default session timeout of 36 s, and keeps its members 
and generation.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_load_through_three_gateways_when_the_coordinator_crashes_should_move_its_groups_after_the_session_timeout()
+ {
+    let Some(mut run) = Run::start(&[]).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::Gateway).await;
+    assert!(
+        takeover <= SESSION_TIMEOUT + TAKEOVER_SLACK,
+        "the takeover took {takeover:?}, above the session timeout 
{SESSION_TIMEOUT:?} and \
+         {TAKEOVER_SLACK:?}"
+    );
+    run.finish().await.assert_all();
+}
+
+/// As the crash above, on nodes with the shortest session timeout that a 1 s
+/// `consumer_group.heartbeat_interval` allows: 9.6 s, from an 8 s 
`heartbeat.interval`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_a_short_session_timeout_when_the_coordinator_crashes_should_move_its_groups_within_it()
+ {
+    let Some(mut run) = Run::start(&SHORT_SESSION).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::Gateway).await;
+    assert!(
+        takeover <= SHORT_SESSION_TIMEOUT + TAKEOVER_SLACK,
+        "the takeover took {takeover:?}, above the session timeout 
{SHORT_SESSION_TIMEOUT:?} \
+         and {TAKEOVER_SLACK:?}"
+    );
+    run.finish().await.assert_all();
+}
+
+/// The gateway that coordinates the group crashes together with its node, as 
when their machine
+/// loses power. When that node held the metadata primary, a view change comes 
first, and the new
+/// primary starts the session timeout again. The node recovers from its log 
when it comes back.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+#[serial]
+async fn 
given_a_short_session_timeout_when_the_coordinator_crashes_with_its_node_should_move_its_groups()
+ {
+    let Some(mut run) = Run::start(&SHORT_SESSION).await else {
+        return;
+    };
+    let takeover = run.crash_coordinator(Crash::GatewayAndNode).await;
+    assert!(
+        takeover <= SHORT_SESSION_TIMEOUT + VIEW_CHANGE + TAKEOVER_SLACK,

Review Comment:
   Fixed. That test now picks its group in a coordinator partition of the 
gateway beside the metadata
   primary. So the crash takes the metadata primary, and the test asserts that 
it did. It is renamed
   `..._when_the_coordinator_crashes_with_the_metadata_primary_...`. Measured: 
16.2 and 17.1 s.
   



##########
gateways/kafka/README.md:
##########
@@ -386,8 +386,9 @@ cargo run -p iggy-gateway-kafka
   gateway log names it.
 - To lower the takeover time after a crash, lower Iggy's `heartbeat.interval`. 
It must stay above
   (3 times `consumer_group.heartbeat_interval` + 6.2 s) / 1.2. With
-  `consumer_group.heartbeat_interval = "1s"`, that is about 7.7 s, so a 
takeover after about 9 s
-  (not tested). See `core/server/config.toml`.
+  `consumer_group.heartbeat_interval = "1s"`, that is about 7.7 s. An 8 s 
`heartbeat.interval`
+  gives a session timeout of 9.6 s, and `tests/cluster_failover_tests.rs` 
measures a takeover
+  after 10 to 11 s. See `core/server/config.toml`.

Review Comment:
   Fixed. README.md, at both places, and CONSUMER_GROUPS.md now give the 
asserted windows and the
   mechanism.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@iggy.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org