krishvishal commented on code in PR #4702: URL: https://github.com/apache/iggy/pull/4702#discussion_r4441643629
########## 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. Every gateway now starts with `IGGY_KAFKA_IDEMPOTENT_PRODUCERS=true`. Each crash test also kills the coordinator during a burst. A second idempotent producer writes 100,000 records through it, and the SIGKILL lands after 10,000 acknowledgements. If that producer logged no resend of a batch in flight to the crashed gateway, the test fails. The test also proves a deduplicated resend: a record acknowledged through another gateway at an offset that Iggy stored before the crash. Only a resend that Iggy answers with the first offset of its batch gets that. Every run so far found some, for example partitions 3 and 4. ########## 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: Fixed. The clients bootstrap from the coordinator alone, and keep its Metadata for the life of the container (`metadata.max.age.ms=600000`). So their Produce and Fetch requests go to the gateway that crashes. The clients log `NetworkClient` at INFO. If the producer, a member or the burst did not log `Node N disconnected.` for the crashed gateway, the crash test fails. After a crash of the gateway alone, the producer must write on within 8 s, below the earliest takeover (8.6 s with the short session). A const assert keeps it there. A crash of the metadata primary stalls every partition until the next primary, so that test waits up to 90 s. The members cannot fetch before the takeover. `kafka-verifiable-consumer.sh` polls with no timeout, and the Java consumer then blocks in `poll` until it finds a coordinator. So the test checks that each member lost its connection to the crashed gateway. It also checks that the members read on through another gateway after the takeover. ########## 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: Fixed. Each crash test now checks a window: from the session timeout minus one `consumer_group.heartbeat_interval`, to the timeout plus that interval and 4 s. A crash that takes the metadata primary adds 10 s. That is 31 to 45 s by default, and 8.6 to 14.6 s with the short session. The probe times a move by the end of the round that saw it, so an early takeover cannot hide inside a round. The short-session tests are now named `..._after_it`. README.md and CONSUMER_GROUPS.md give these windows in place of the measured figures. As long as the test passes, the docs stay true. Measured over four runs: 36.2 to 41.8 s by default, 10.4 to 11.4 s with the short session. ########## 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: Fixed. The suite parses `offsets_committed` and fails on any commit with `success: false`. After each SIGTERM hand-over, each node stop and each crash takeover, it reads the committed offsets with OffsetFetch from the current coordinator. When one is below an acknowledged commit, the test fails. At the end the coordinator must return exactly the last commit of each partition. That must also be the offset after the last record read there. ########## 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 now holds a member in a group of each of the 16 coordinator partitions. Before each stop, the test reads which partitions the gateway answers for. The step then waits until another gateway answers for every one of them, so `HANDOVER_LIMIT` checks the slowest move of every step. Each step also picks the current coordinator of the Java group first, among the gateways that did not stop yet. So that group moves more than once. ########## .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: Fixed. A `profile.ci` override sets `retries = 0` for this binary. A race that fails one run in four now fails CI. The 10 minute cap is gone. The slowest test takes about 160 s, inside the default 5 minute cap, and each test fails itself at 270 s. -- 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]
