spetz commented on code in PR #4702: URL: https://github.com/apache/iggy/pull/4702#discussion_r4440232521
########## 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: The gateways start with an empty environment and no `IGGY_KAFKA_IDEMPOTENT_PRODUCERS`, so every one of them runs with it off (`gateways/kafka/src/bridge/config.rs:283`), and the README says that a resend of an idempotent producer is then not checked and can write twice (`gateways/kafka/README.md:592`). The suite pairs an idempotent Java producer with an assertion that each record is stored once, but nothing on the gateway side makes that hold across a failover. When a SIGKILL lands on a Produce that Iggy already wrote, the producer resends the batch through another gateway and the record is stored twice, so a pass of the crash tests says only that no resend crossed the kill. The property the README does promise, that an idempotent producer moving to another gateway writes each batch once with the setting on (`gateways/kafka/README.md:400`), never runs. ########## 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 Review Comment: No check reads a committed offset. The members never restart and the group never rebalances, so they fetch from positions held in memory, and a successor that replayed the group without its offsets after a SIGKILL or a node loss passes every assertion in this file. `gateways/kafka/tests/coordinator_real_bridge_tests.rs:98` and `gateways/kafka/tests/coordinator_cluster_tests.rs:126` read offsets back only after a graceful in-process stop, so no suite covers offsets after a crash takeover, while #4515 lists "each group keeps its committed offsets" among its acceptance criteria. `kafka-verifiable-consumer.sh` prints an `offsets_committed` event with the offsets and a `success` flag for each commit, and the suite parses only `record_data` and the assignment events, so commits that fail during a move go unnoticed as well. ########## 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: The server takes millisecond intervals, so 9.6 s is the shortest timeout from a whole-second `heartbeat.interval`, not the shortest that a 1 s report interval allows: `7667ms` passes the rule at `core/configs/src/server_config/validators.rs:379` with a 9.2 s timeout. The claim repeats at `gateways/kafka/tests/cluster_failover_tests.rs:294`. At `gateways/kafka/tests/cluster_failover_tests.rs:130` no report ends a session, the metadata primary expires it in its own pass every `consumer_group.heartbeat_interval` (`core/server/src/consumer_group/liveness.rs:1362`). ########## 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) Review Comment: Metadata names the answering gateway as leader of every partition (`gateways/kafka/src/protocol/handlers/metadata.rs:546`), so the producer sends every Produce to the one gateway whose Metadata it holds, chosen from the bootstrap list, and nothing here records or picks which one. `crash_coordinator` kills only the coordinator, so when the producer sits on another gateway this wait passes with no failover on the data path, and the claim at `gateways/kafka/tests/cluster_failover_tests.rs:275` that the producer writes on through the other gateways goes untested in that run. The members' Fetch path has the same gap. In the short-session tests the 20 s `DATA_PATH_LIMIT` is also above the 9.6 s session timeout, so a data path that waited for the takeover would pass there even with the producer on the coordinator. ########## 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: `rotation()` reads the coordinator once, before the loop, and after the first SIGTERM the group sits on whichever gateway took it over. `wait_served` returns at the first probe round in which a gateway other than the stopped one answers, so a step that stops a gateway that does not coordinate the group returns after one round and the `HANDOVER_LIMIT` assert checks nothing. Locally step 2 of this test reported "its groups moved 62.894031ms", and step 3 of the gateway-with-node test (`gateways/kafka/tests/cluster_failover_tests.rs:257`) reported 181 ms. The other coordinator partitions of the stopped gateway hold no group, so nothing observes their hand-over either. The doc comment's "Its groups move to the other gateways at once" is checked for the first step, and for a later one only when the group happened to land on the next gateway in the rotation. ########## .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)' Review Comment: CI runs nextest with `--profile ci`, which retries a failed test three times (`.config/nextest.toml:151`), and this override keeps that. The one-coordinator check samples one round every 200 ms and the stored-once check depends on where a kill lands, so a race that fails one run in four passes CI as flaky. Each attempt can also hold a shard for up to ten minutes, since `threads-required = "num-cpus"` reserves the whole machine. ########## 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, Review Comment: All three crash tests assert only an upper bound, so a takeover anywhere from 0 s to the limit passes. This test's name says the group moves after the session timeout, and its doc comment says it moves only when Iggy ends the dead gateway's sessions, yet a successor that claimed the partition 2 s after the SIGKILL, with that session still open, would pass. The session cannot end before the timeout minus one `consumer_group.heartbeat_interval`, since the metadata primary expires leases once per pass (`core/server/src/consumer_group/lease.rs:332`, `core/server/src/consumer_group/liveness.rs:141`), so an earlier takeover means a broken lease. The short-session test is named `..._within_it`, yet its takeover lands above the 9.6 s timeout (10.0 and 10.4 s locally) and the bound at `gateways/kafka/tests/cluster_failover_tests.rs:305` allows 24.6 s. The figures that README.md and CONSUMER_GROUPS.md now quote from this file, about 39 s and 10 to 11 s, are printed but not enforced, so a ta keover that drifts to 50 s passes while the docs keep the old number. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
