You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/10/25 08:36:56 UTC
(iggy) branch kafka-cluster-failover-test created (now 49622badd)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch kafka-cluster-failover-test
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 49622badd test(sdk): keep the session across sends to split partition primaries
This branch includes the following new commits:
new f0ebc0d5f test(gateways): add a 3-node failover test for the Kafka gateway
new 49622badd test(sdk): keep the session across sends to split partition primaries
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 02/02: test(sdk): keep the session across sends to split partition primaries
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-cluster-failover-test
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 49622baddcf6a964efbf67c95dae69eeda2f07f9
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sun Oct 25 13:37:53 2026 +0530
test(sdk): keep the session across sends to split partition primaries
One producer sends in turn to 3 partitions whose primaries sit on different nodes, the case of #4436.
---
core/integration/tests/cluster/chaos_monkey.rs | 2 +-
.../tests/cluster/partition_primary_routing.rs | 95 ++++++++++++++++++++++
2 files changed, 96 insertions(+), 1 deletion(-)
diff --git a/core/integration/tests/cluster/chaos_monkey.rs b/core/integration/tests/cluster/chaos_monkey.rs
index 7f9950ed1..28bf9799d 100644
--- a/core/integration/tests/cluster/chaos_monkey.rs
+++ b/core/integration/tests/cluster/chaos_monkey.rs
@@ -88,7 +88,7 @@
//! | `cascading-crash` | for each node: SIGKILL it, SIGKILL its successor `SECOND_KILL_DELAY_SECS` later during the view change, restart both `d` after the second kill; `persisted` | fails on master with `persisted`: after the restarted pair elected a new view, the surviving node fenced partition 0 as divergent (`partition WAL append does not extend its history`, then a ShardFatal and a process exit) and the partition stayed frozen. With `replicated` on 3 nodes one node stays alive alon [...]
//! | `full-outage` | SIGKILL every node at once after `g`, restart all after `d`; `persisted` | fails on master with `persisted`: the cluster recovers and loses nothing, but every producer stays `disconnected` until the end, the producer stall described below. With `replicated`, unflushed acknowledged writes are lost by design, so that setting is expected to fail |
//! | `rolling-crash` | `crash-restart` on 1 partition at 30k messages/s from 12 producers in batches of 100, `persisted`, a kill every 80 s and 40 s down, for 360 s | #4438: a restarted replica logs `needs state transfer`, never finishes the transfer and keeps serving a stale prefix. Timing dependent; the loss reaches millions of messages at this load. Expected: pass |
-//! | `split-primaries` | `crash-restart`, then a 60 s window without faults, opened once the replica mesh has formed, in which the p99 latency of an acknowledged batch and the longest gap between acknowledgements must stay under `IGGY_TEST_CHAOS_P99_LIMIT_MS` | #4436: once partition primaries sit on different nodes, the SDK moves its session on nearly every send. #4437 is the deterministic test. Expected: pass |
+//! | `split-primaries` | `crash-restart`, then a 60 s window without faults, opened once the replica mesh has formed, in which the p99 latency of an acknowledged batch and the longest gap between acknowledgements must stay under `IGGY_TEST_CHAOS_P99_LIMIT_MS` | #4436: once partition primaries sit on different nodes, the SDK moves its session on nearly every send. `given_partitions_on_different_primaries_when_sending_round_robin_should_keep_the_session` in `partition_primary_routing.rs` [...]
//!
//! `isolate` needs no privileges: the harness relays every replica link over
//! loopback. A cut closes the node's relayed connections, so its peers see EOF
diff --git a/core/integration/tests/cluster/partition_primary_routing.rs b/core/integration/tests/cluster/partition_primary_routing.rs
index 7dd60a161..e4d452574 100644
--- a/core/integration/tests/cluster/partition_primary_routing.rs
+++ b/core/integration/tests/cluster/partition_primary_routing.rs
@@ -389,6 +389,101 @@ async fn assert_partition_payload(
assert_eq!(polled.messages[0].payload.as_ref(), expected.as_bytes());
}
+/// One producer alternates sends between partitions with primaries on
+/// different nodes. No send may wait out a refusal window or sign the
+/// session in again (#4436).
+#[iggy_harness(cluster_nodes = 3, server(sharding.cpu_allocation = "2"))]
+async fn given_partitions_on_different_primaries_when_sending_round_robin_should_keep_the_session(
+ harness: &mut TestHarness,
+) {
+ const PARTITIONS: u32 = 3;
+ const ROUND_ROBIN_SENDS: usize = 12;
+ /// How long the SDK replayed a refused send before it moved the session
+ /// in #4436. It did that twice per send.
+ const REFUSAL_WINDOW: Duration = Duration::from_secs(2);
+
+ let producer = connect_without_login(harness.node(0).tcp_addr().unwrap()).await;
+ producer
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .unwrap();
+ producer
+ .create_stream(STREAM_NAME, &StreamCreateOptions::default())
+ .await
+ .unwrap();
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ producer
+ .create_topic(
+ &stream,
+ TOPIC_NAME,
+ &TopicCreateOptions {
+ partitions_count: Some(PARTITIONS),
+ ..TopicCreateOptions::default()
+ },
+ )
+ .await
+ .unwrap();
+ let mut primaries = Vec::new();
+ for partition_id in 0..PARTITIONS {
+ let route = partition_route(&producer, &stream, &topic, partition_id).await;
+ primaries.push(route.primary_replica);
+ }
+ assert_eq!(
+ primaries,
+ [0, 1, 2],
+ "precondition: a fresh cluster places each partition's primary on its own node"
+ );
+
+ // The first send to each partition opens the link to its primary.
+ for partition_id in 0..PARTITIONS {
+ producer
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(partition_id),
+ &mut [message("warm-up")],
+ )
+ .await
+ .expect("warm-up send");
+ }
+ let identity = producer.session_identity().await.unwrap();
+ let mut events = producer.subscribe_events().await;
+ // A subscription starts with every event the client published before it.
+ while events.try_recv().is_ok() {}
+
+ let mut sends = Vec::with_capacity(ROUND_ROBIN_SENDS);
+ for partition_id in (0..PARTITIONS).cycle().take(ROUND_ROBIN_SENDS) {
+ let started = Instant::now();
+ producer
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(partition_id),
+ &mut [message("round-robin")],
+ )
+ .await
+ .expect("round-robin send");
+ sends.push((partition_id, started.elapsed()));
+ }
+
+ let current = producer.session_identity().await.unwrap();
+ assert_eq!(
+ (current.client_id, current.session),
+ (identity.client_id, identity.session),
+ "the producer must keep its session across sends to different primaries"
+ );
+ let reconnects: Vec<_> = std::iter::from_fn(|| events.try_recv().ok()).collect();
+ assert!(
+ reconnects.is_empty(),
+ "sends to different primaries must not reconnect or sign in again: {reconnects:?}"
+ );
+ assert!(
+ sends.iter().all(|(_, elapsed)| *elapsed < REFUSAL_WINDOW),
+ "no send may wait out a refusal window (partition, send time): {sends:?}"
+ );
+}
+
#[iggy_harness(cluster_nodes = 5, server(sharding.cpu_allocation = "2"))]
async fn given_balanced_primaries_when_a_replica_fails_and_returns_should_rebalance_without_data_loss(
harness: &mut TestHarness,
(iggy) 01/02: test(gateways): add a 3-node failover test for the Kafka gateway
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-cluster-failover-test
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit f0ebc0d5fcbf7f7376bd42850dddca71c2f307a8
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Sun Oct 25 13:37:33 2026 +0530
test(gateways): add a 3-node failover test for the Kafka gateway
The script stops and restarts Iggy nodes under Kafka load, and fails if a record is missing or an offset reads back wrong.
---
Cargo.lock | 1 +
gateways/kafka/Cargo.toml | 2 +
gateways/kafka/README.md | 5 +-
gateways/kafka/docs/MANUAL_TESTING.md | 38 ++
gateways/kafka/docs/OFFSET_STORAGE.md | 12 +-
gateways/kafka/examples/partition_routes.rs | 115 +++++
gateways/kafka/scripts/cluster-failover.sh | 529 +++++++++++++++++++++
.../kafka/tests/java/ReadYourCommitsProbe.java | 73 +++
8 files changed, 761 insertions(+), 14 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index ad0b60756..6c40fa9e7 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7290,6 +7290,7 @@ dependencies = [
"flate2",
"futures",
"iggy",
+ "iggy_binary_protocol",
"kafka-protocol",
"libc",
"lz4",
diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml
index bd4afe645..50f2a24c3 100644
--- a/gateways/kafka/Cargo.toml
+++ b/gateways/kafka/Cargo.toml
@@ -78,6 +78,8 @@ zstd = { workspace = true }
[dev-dependencies]
# Repairs the CRC of a batch header a test rewrote, which the decoder checks first.
crc32c = { workspace = true }
+# Partition route requests for `examples/partition_routes.rs`.
+iggy_binary_protocol = { workspace = true }
# The client half, so a test encodes requests and decodes responses the way a consumer does.
kafka-protocol = { workspace = true, features = ["client"] }
# Certificates for the TLS listener tests, minted per test so nothing in the tree expires.
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 692aa6a2c..2fb8cdc7e 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -44,10 +44,7 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34
> resumes from its last commit, also after a gateway restart
> ([docs/OFFSET_STORAGE.md](docs/OFFSET_STORAGE.md)). Every Iggy server must run a build with the
> external group consumer kind ([#4392](https://github.com/apache/iggy/pull/4392)) before the first
-> commit. In a partly upgraded cluster, one commit can make a partition stop committing. In a
-> cluster, a node that has not applied a group's first commit answers "no offset", so a consumer on
-> `auto.offset.reset=latest` can skip records until
-> [#4409](https://github.com/apache/iggy/issues/4409).
+> commit. In a partly upgraded cluster, one commit can make a partition stop committing.
## Run
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index 45a9dc702..1f9bc5083 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -311,6 +311,44 @@ Run it four ways, with `CA_FILE=/tmp/certs/ca.pem` exported: `go run . 127.0.0.1
and certificate runs also list `User:iggy`'s bindings, and the last reports the broker closing the
connection during `ApiVersions`.
+### Category V — Node failover in a 3-node cluster
+
+Run against a 3-node Iggy cluster, because a single node has no partition primaries to move. A
+partition primary is the node that orders the writes of a partition. `scripts/cluster-failover.sh`
+starts 3 Iggy nodes and one gateway on 127.0.0.1, and creates a topic with 6 partitions. Each node
+holds the primary of 2 partitions. Through the gateway, `kafka-producer-perf-test.sh` sends 2000
+records a second, and a `kafka-verifiable-consumer.sh` group reads them.
+
+Under this load, the script stops a node that holds partition primaries and restarts it. Then it
+does the same to the metadata primary. At the end, `tests/java/ReadYourCommitsProbe.java` commits an
+offset in each partition and reads it back at once, 300 times. The gateway's node is a backup for 4
+of the 6 partitions. Since the fix for [#4409](https://github.com/apache/iggy/issues/4409), offset
+reads go to the partition primary. If a backup answers one instead, the probe gets a wrong offset.
+
+The script needs Linux, Docker, jq, perl and a JDK 17 or later. A run takes about 5 minutes:
+
+```bash
+cargo build --release --bin iggy-server
+cargo build --release -p iggy-gateway-kafka --bin iggy-gateway-kafka --example partition_routes
+docker pull apache/kafka:4.4.0
+./gateways/kafka/scripts/cluster-failover.sh /tmp/failover-term
+STOP_SIGNAL=KILL ./gateways/kafka/scripts/cluster-failover.sh /tmp/failover-kill
+```
+
+The work dir keeps every log. To print the report of a run again, run `cluster-failover.sh --report
+<work dir>`. For each phase, the report shows the slowest send, the time until sends recovered and
+the client retries. It also counts how often the gateway's Iggy session dialed another node.
+
+| ID | Test | Command | Pass criteria | Last run |
+| ---- | ------ | --------- | --------------- | ---------- |
+| V1 | Graceful stop | `cluster-failover.sh` | Prints `pass` and exits 0. No record is missing, and no offset reads back wrong | Pass |
+| V2 | Crash | `STOP_SIGNAL=KILL cluster-failover.sh` | Same as V1 | Pass |
+
+In the last runs, the gateway's session stayed on its node, and the producer got no error. A stop
+paused sends for about 5 s. Then new partition primaries took over, and sends continued. After a
+restart, the restarted node caught up for up to 25 s, and sends through the gateway were slow in
+that time.
+
### Category A — Smoke tests (must pass before check-in)
The gateway binds `:9093` by default; `kafka-message-gen`'s own default is `:9092` (a real Kafka
diff --git a/gateways/kafka/docs/OFFSET_STORAGE.md b/gateways/kafka/docs/OFFSET_STORAGE.md
index ab9f3365d..67f983fd4 100644
--- a/gateways/kafka/docs/OFFSET_STORAGE.md
+++ b/gateways/kafka/docs/OFFSET_STORAGE.md
@@ -142,16 +142,8 @@ The Iggy HTTP API cannot carry kind 3. The gateway uses TCP, so this does not li
## Reads in a cluster
-An offset read goes to the Iggy node that the gateway is connected to. That is not always the
-partition primary (the node that orders the partition writes). So an OffsetFetch right after an
-OffsetCommit can return the older value until that node applies the commit.
-
-- After a later commit, the consumer reads some records again.
-- After the first commit of a group, the node has no offset and answers -1. The consumer then
- applies `auto.offset.reset`. With `latest`, it skips records.
-
-Reads from the primary are [#4409](https://github.com/apache/iggy/issues/4409). Until then, use
-`auto.offset.reset=earliest` for a group that a cluster serves.
+OffsetCommit and OffsetFetch both go to the partition primary, whichever Iggy node the gateway
+dials. So an OffsetFetch right after an OffsetCommit returns the committed offset.
## References
diff --git a/gateways/kafka/examples/partition_routes.rs b/gateways/kafka/examples/partition_routes.rs
new file mode 100644
index 000000000..53a3bd685
--- /dev/null
+++ b/gateways/kafka/examples/partition_routes.rs
@@ -0,0 +1,115 @@
+// 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.
+
+//! Prints the metadata primary and the primary of each partition of one topic, as one JSON line
+//! a second, until stopped. `scripts/cluster-failover.sh` reads it.
+//!
+//! ```bash
+//! cargo run --release -p iggy-gateway-kafka --example partition_routes -- \
+//! 127.0.0.1:28000 kafka orders 6
+//! ```
+//!
+//! `IGGY_USERNAME` and `IGGY_PASSWORD` default to `iggy`.
+
+use std::error::Error;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+use iggy::prelude::{
+ AutoLogin, Client, ClusterClient, ClusterNodeRole, Credentials, IggyClient, IggyClientBuilder,
+};
+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;
+
+const INTERVAL: Duration = Duration::from_secs(1);
+/// A poll that runs past this prints an error line, so the series keeps its pace.
+const POLL_TIMEOUT: Duration = Duration::from_secs(2);
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+ let args: Vec<String> = std::env::args().skip(1).collect();
+ let [address, stream, topic, partitions] = args.as_slice() else {
+ return Err("usage: partition_routes <iggy address> <stream> <topic> <partitions>".into());
+ };
+ let partitions: u32 = partitions.parse()?;
+ let username = std::env::var("IGGY_USERNAME").unwrap_or_else(|_| "iggy".to_string());
+ let password = std::env::var("IGGY_PASSWORD").unwrap_or_else(|_| "iggy".to_string());
+ let client = IggyClientBuilder::new()
+ .with_tcp()
+ .with_server_address(address.clone())
+ .with_auto_sign_in(AutoLogin::Enabled(Credentials::UsernamePassword(
+ username,
+ SecretString::from(password),
+ )))
+ .build()?;
+ client.connect().await?;
+ let requests = (0..partitions)
+ .map(|partition_id| {
+ Ok(GetPartitionRouteRequest {
+ stream_id: WireIdentifier::named(stream.as_str())?,
+ topic_id: WireIdentifier::named(topic.as_str())?,
+ partition_id,
+ }
+ .to_bytes())
+ })
+ .collect::<Result<Vec<_>, Box<dyn Error>>>()?;
+
+ let mut ticks = tokio::time::interval(INTERVAL);
+ loop {
+ ticks.tick().await;
+ let at_ms = SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis();
+ let line = match tokio::time::timeout(POLL_TIMEOUT, routes(&client, &requests)).await {
+ Ok(Ok((metadata_primary, routes))) => {
+ let primaries: Vec<u8> = routes.iter().map(|route| route.primary_replica).collect();
+ let views: Vec<u32> = routes.iter().map(|route| route.view).collect();
+ format!(
+ r#"{{"at_ms":{at_ms},"metadata_primary":{metadata_primary},"primaries":{primaries:?},"views":{views:?}}}"#
+ )
+ }
+ Ok(Err(error)) => {
+ let error = error.to_string().replace(['"', '\\'], "'");
+ format!(r#"{{"at_ms":{at_ms},"error":"{error}"}}"#)
+ }
+ Err(_) => format!(r#"{{"at_ms":{at_ms},"error":"no answer within {POLL_TIMEOUT:?}"}}"#),
+ };
+ println!("{line}");
+ }
+}
+
+/// The metadata primary, then the route of each partition, as the node that answers sees them.
+async fn routes(
+ client: &IggyClient,
+ requests: &[bytes::Bytes],
+) -> Result<(u8, Vec<PartitionRouteResponse>), Box<dyn Error>> {
+ let metadata = client.get_cluster_metadata().await?;
+ let metadata_primary = metadata
+ .nodes
+ .iter()
+ .find(|node| node.role == ClusterNodeRole::Leader)
+ .ok_or("no node is the metadata primary")?
+ .replica_id;
+ let mut routes = Vec::with_capacity(requests.len());
+ for request in requests {
+ let body = client
+ .send_binary_request(GET_PARTITION_ROUTE_CODE, request.clone())
+ .await?;
+ routes.push(PartitionRouteResponse::decode_from(&body)?);
+ }
+ Ok((metadata_primary, routes))
+}
diff --git a/gateways/kafka/scripts/cluster-failover.sh b/gateways/kafka/scripts/cluster-failover.sh
new file mode 100755
index 000000000..27a2c616b
--- /dev/null
+++ b/gateways/kafka/scripts/cluster-failover.sh
@@ -0,0 +1,529 @@
+#!/usr/bin/env bash
+# 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.
+
+# Stops Iggy nodes under Kafka load through one gateway, then reports send latency, the error
+# codes that clients got and the time to recover.
+#
+# Runs 3 Iggy nodes and one gateway on 127.0.0.1, with kafka-producer-perf-test.sh and a
+# kafka-verifiable-consumer.sh group in Docker. Under load it stops, then restarts:
+#
+# 1. a node that holds partition primaries,
+# 2. the metadata primary.
+#
+# The gateway dials GATEWAY_NODE. With the default, no stop hits it. At the end
+# tests/java/ReadYourCommitsProbe.java commits offsets and reads each back at once.
+#
+# Needs Linux for Docker host networking, the KAFKA_IMAGE image pulled, jq, perl and a JDK 17+.
+# Build first, from the workspace root:
+#
+# cargo build --release --bin iggy-server
+# cargo build --release -p iggy-gateway-kafka --bin iggy-gateway-kafka --example partition_routes
+#
+# Usage: gateways/kafka/scripts/cluster-failover.sh [work dir]
+# gateways/kafka/scripts/cluster-failover.sh --report <work dir>
+#
+# The work dir keeps every log, each line stamped in epoch ms. --report prints a run's report
+# again. Both exit 1 if the consumer missed a record or the probe read back a wrong offset.
+
+set -euo pipefail
+
+TARGET_DIR="${TARGET_DIR:-target/release}"
+KAFKA_IMAGE="${KAFKA_IMAGE:-apache/kafka:4.4.0}"
+BASE_PORT="${BASE_PORT:-28000}"
+GATEWAY_NODE="${GATEWAY_NODE:-2}"
+PARTITIONS="${PARTITIONS:-6}"
+# Records per second.
+RATE="${RATE:-2000}"
+WARMUP_SECS="${WARMUP_SECS:-30}"
+DOWN_SECS="${DOWN_SECS:-30}"
+SETTLE_SECS="${SETTLE_SECS:-60}"
+# TERM stops a node. KILL crashes it.
+STOP_SIGNAL="${STOP_SIGNAL:-TERM}"
+SHARDS="${SHARDS:-2}"
+PROBE_ROUNDS="${PROBE_ROUNDS:-300}"
+
+NODES=3
+PROBE_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../tests/java" && pwd)"
+TOPIC="failover"
+GROUP="failover-group"
+STREAM="kafka"
+GATEWAY_PORT=$((BASE_PORT + 100))
+GATEWAY_ADDR="127.0.0.1:${GATEWAY_PORT}"
+PRODUCER="iggy-failover-producer-$$"
+CONSUMER="iggy-failover-consumer-$$"
+RECORDS=$((RATE * (WARMUP_SECS + 2 * (DOWN_SECS + SETTLE_SECS) + 10)))
+
+if [[ "${1:-}" == "--report" ]]; then
+ WORK_DIR="${2:?usage: $0 --report <work dir>}"
+ # shellcheck source=/dev/null
+ source "$WORK_DIR/settings"
+else
+ WORK_DIR="${1:-$(mktemp -d "${TMPDIR:-/tmp}/iggy-cluster-failover.XXXXXX")}"
+fi
+mkdir -p "$WORK_DIR"
+EVENTS="$WORK_DIR/events.log"
+
+NODE_PIDS=()
+GATEWAY_PID=""
+ROUTES_PID=""
+PRODUCER_PID=""
+CONSUMER_PID=""
+
+log() {
+ echo "[$(date +%T)] $*" >&2
+}
+
+# Appends "<epoch ms> <kind> [<node> <note>]" to events.log. Every kind but "stopped" opens a
+# phase of the report.
+event() {
+ echo "$(date +%s%3N) $*" >>"$EVENTS"
+ log "$*"
+}
+
+# Prefixes each line with its time in epoch milliseconds and drops ANSI colors.
+stamp() {
+ perl -MTime::HiRes=time -ne 'BEGIN { $| = 1 } s/\e\[[0-9;]*m//g; printf "%d %s", time * 1000, $_'
+}
+
+tcp_port() {
+ echo $((BASE_PORT + 10 * $1))
+}
+
+replica_port() {
+ echo $((BASE_PORT + 10 * $1 + 1))
+}
+
+cleanup() {
+ docker rm -f "$PRODUCER" "$CONSUMER" >/dev/null 2>&1 || true
+ for pid in "$ROUTES_PID" "$GATEWAY_PID" "${NODE_PIDS[@]}"; do
+ if [[ -n "$pid" ]]; then
+ kill "$pid" 2>/dev/null || true
+ fi
+ done
+ wait 2>/dev/null || true
+}
+
+preflight() {
+ local binary
+ for binary in iggy-server iggy-gateway-kafka examples/partition_routes; do
+ if [[ ! -x "$TARGET_DIR/$binary" ]]; then
+ echo "missing $TARGET_DIR/$binary, see the build commands at the top of $0" >&2
+ exit 1
+ fi
+ done
+ for binary in docker jq perl javac; do
+ command -v "$binary" >/dev/null || {
+ echo "$binary is required" >&2
+ exit 1
+ }
+ done
+ docker image inspect "$KAFKA_IMAGE" >/dev/null 2>&1 || {
+ echo "pull $KAFKA_IMAGE first" >&2
+ exit 1
+ }
+}
+
+# Waits until something accepts on `port`. Fails early if process `pid` exits first.
+wait_for_port() {
+ local port=$1 pid=$2 deadline=$((SECONDS + 60))
+ until (exec 3<>"/dev/tcp/127.0.0.1/$port") 2>/dev/null; do
+ if ! kill -0 "$pid" 2>/dev/null || ((SECONDS >= deadline)); then
+ return 1
+ fi
+ sleep 0.2
+ done
+}
+
+start_node() {
+ local node=$1 peer
+ local env=(
+ "PATH=$PATH"
+ "IGGY_PATH=$WORK_DIR/node-$node"
+ "IGGY_TCP_ADDRESS=127.0.0.1:$(tcp_port "$node")"
+ "IGGY_HTTP_ENABLED=false"
+ "IGGY_QUIC_ENABLED=false"
+ "IGGY_WEBSOCKET_ENABLED=false"
+ "IGGY_SHARDING_CPU_ALLOCATION=$SHARDS"
+ "IGGY_SHARDING_PIN_CORES=false"
+ "IGGY_ROOT_USERNAME=iggy"
+ "IGGY_ROOT_PASSWORD=iggy"
+ "IGGY_CLUSTER_ENABLED=true"
+ "IGGY_CLUSTER_NAME=kafka-failover"
+ )
+ for ((peer = 0; peer < NODES; peer++)); do
+ env+=(
+ "IGGY_CLUSTER_NODES_${peer}_NAME=node-$peer"
+ "IGGY_CLUSTER_NODES_${peer}_IP=127.0.0.1"
+ "IGGY_CLUSTER_NODES_${peer}_REPLICA_ID=$peer"
+ "IGGY_CLUSTER_NODES_${peer}_PORTS_TCP=$(tcp_port "$peer")"
+ "IGGY_CLUSTER_NODES_${peer}_PORTS_TCP_REPLICA=$(replica_port "$peer")"
+ )
+ done
+ env -i "${env[@]}" "$TARGET_DIR/iggy-server" --replica-id "$node" \
+ > >(stamp >>"$WORK_DIR/node-$node.log") 2>&1 &
+ NODE_PIDS[node]=$!
+ wait_for_port "$(tcp_port "$node")" "${NODE_PIDS[node]}" || {
+ echo "node $node did not start, see $WORK_DIR/node-$node.log" >&2
+ exit 1
+ }
+}
+
+stop_node() {
+ local node=$1
+ kill -s "$STOP_SIGNAL" "${NODE_PIDS[node]}"
+ wait "${NODE_PIDS[node]}" 2>/dev/null || true
+}
+
+# The bridge signs in before the listener binds, and fails while the cluster has no metadata
+# primary yet. So a gateway that exits early gets another try.
+start_gateway() {
+ local attempt
+ for attempt in 1 2 3 4 5; do
+ env -i "PATH=$PATH" \
+ "RUST_LOG=info,iggy=info,iggy_gateway_kafka::protocol::handlers=debug,iggy_gateway_kafka::bridge=debug" \
+ "IGGY_KAFKA_BIND_ADDR=$GATEWAY_ADDR" \
+ "IGGY_KAFKA_BRIDGE_ENABLED=true" \
+ "IGGY_KAFKA_IGGY_ADDR=127.0.0.1:$(tcp_port "$GATEWAY_NODE")" \
+ "IGGY_KAFKA_IGGY_PASSWORD=iggy" \
+ "$TARGET_DIR/iggy-gateway-kafka" > >(stamp >>"$WORK_DIR/gateway.log") 2>&1 &
+ GATEWAY_PID=$!
+ if wait_for_port "$GATEWAY_PORT" "$GATEWAY_PID"; then
+ return 0
+ fi
+ log "gateway attempt $attempt failed"
+ sleep 2
+ done
+ echo "the gateway did not start, see $WORK_DIR/gateway.log" >&2
+ exit 1
+}
+
+kafka_tool() {
+ docker run --rm --pull never --network host "$KAFKA_IMAGE" "/opt/kafka/bin/$1" "${@:2}"
+}
+
+latest_routes() {
+ grep '"primaries"' "$WORK_DIR/routes.jsonl" | tail -n 1 || true
+}
+
+# Every node holds primaries, and no node holds two more than another.
+wait_for_spread_primaries() {
+ local deadline=$((SECONDS + 120))
+ until latest_routes | jq -e --argjson nodes "$NODES" \
+ '.primaries | group_by(.) | map(length) | length == $nodes and max - min <= 1' >/dev/null 2>&1; do
+ if ((SECONDS >= deadline)); then
+ echo "partition primaries did not spread: $(latest_routes)" >&2
+ exit 1
+ fi
+ sleep 1
+ done
+}
+
+metadata_primary() {
+ latest_routes | jq '.metadata_primary'
+}
+
+# A node that holds partition primaries, and is neither the metadata primary nor the gateway's.
+primary_holder() {
+ latest_routes | jq --argjson gateway "$GATEWAY_NODE" \
+ '.metadata_primary as $metadata | .primaries | unique
+ | map(select(. != $metadata and . != $gateway)) | first // empty'
+}
+
+partitions_of() {
+ latest_routes | jq -c --argjson node "$1" '[.primaries | to_entries[] | select(.value == $node) | .key]'
+}
+
+start_load() {
+ docker run --rm --pull never --network host --name "$CONSUMER" "$KAFKA_IMAGE" \
+ /opt/kafka/bin/kafka-verifiable-consumer.sh --bootstrap-server "$GATEWAY_ADDR" \
+ --topic "$TOPIC" --group-id "$GROUP" --reset-policy earliest --verbose \
+ > >(stamp >"$WORK_DIR/consumer.out") 2> >(stamp >"$WORK_DIR/consumer.err") &
+ CONSUMER_PID=$!
+ # Keys spread the records over every partition. The monotonic payload is the record number,
+ # so the consumer can tell a lost record from a duplicate.
+ docker run --rm --pull never --network host --name "$PRODUCER" "$KAFKA_IMAGE" \
+ /opt/kafka/bin/kafka-producer-perf-test.sh --bootstrap-server "$GATEWAY_ADDR" \
+ --topic "$TOPIC" --num-records "$RECORDS" --throughput "$RATE" --payload-monotonic \
+ --record-key-range $((PARTITIONS * 1000)) --key-distribution range \
+ --reporting-interval 1000 --command-property acks=all --print-metrics \
+ > >(stamp >"$WORK_DIR/producer.out") 2> >(stamp >"$WORK_DIR/producer.err") &
+ PRODUCER_PID=$!
+}
+
+# Stops `node`, keeps it down for DOWN_SECS, restarts it, then waits SETTLE_SECS.
+outage() {
+ local node=$1 role=$2
+ event "stop $node $role, primary of partitions $(partitions_of "$node")"
+ stop_node "$node"
+ event "stopped $node"
+ sleep "$DOWN_SECS"
+ event "restart $node"
+ start_node "$node"
+ sleep "$SETTLE_SECS"
+}
+
+# Waits for the producer to finish, then for the consumer to read as many records as the
+# producer had acknowledged, or to read nothing more for 30 s.
+finish_load() {
+ local acked read last_read=-1 idle=0 deadline=$((SECONDS + 300))
+ while kill -0 "$PRODUCER_PID" 2>/dev/null && ((SECONDS < deadline)); do
+ sleep 1
+ done
+ docker stop -t 20 "$PRODUCER" >/dev/null 2>&1 || true
+ wait "$PRODUCER_PID" || true
+ acked=$(grep -o '^[0-9]* [0-9]* records sent.*50th' "$WORK_DIR/producer.out" | awk '{ print $2 }')
+ while ((idle < 30)); do
+ read=$(grep -c '"name":"record_data"' "$WORK_DIR/consumer.out" || true)
+ if [[ -n "$acked" ]] && ((read >= acked)); then
+ break
+ fi
+ if ((read == last_read)); then
+ idle=$((idle + 1))
+ else
+ idle=0
+ last_read=$read
+ fi
+ sleep 1
+ done
+ docker stop -t 20 "$CONSUMER" >/dev/null 2>&1 || true
+ wait "$CONSUMER_PID" || true
+ event "end"
+ kafka_tool kafka-consumer-groups.sh --bootstrap-server "$GATEWAY_ADDR" --describe \
+ --group "$GROUP" >"$WORK_DIR/group.txt" 2>&1 || true
+}
+
+# Commits an offset and reads it back at once through the gateway, in every partition. The
+# gateway's node is a backup of the partitions whose primary is elsewhere. The image has no
+# compiler, so the probe compiles here against the image's kafka-clients jar.
+read_your_commits() {
+ local classes="$WORK_DIR/probe"
+ mkdir -p "$classes"
+ docker run --rm --pull never --entrypoint cat "$KAFKA_IMAGE" \
+ "/opt/kafka/libs/kafka-clients-${KAFKA_IMAGE##*:}.jar" >"$classes/kafka-clients.jar"
+ javac --release 17 -cp "$classes/kafka-clients.jar" -d "$classes" \
+ "$PROBE_DIR/ReadYourCommitsProbe.java"
+ log "read your commits, routes $(latest_routes)"
+ docker run --rm --pull never --network host -v "$classes:/probe:ro" "$KAFKA_IMAGE" \
+ java -cp '/opt/kafka/libs/*:/probe' ReadYourCommitsProbe "$GATEWAY_ADDR" "$TOPIC" \
+ "$PARTITIONS" "$PROBE_ROUNDS" read-your-commits >"$WORK_DIR/probe.out" 2>&1 || true
+}
+
+report() {
+ local status=0
+ awk -v rate="$RATE" -v records="$RECORDS" -v gateway_node="127.0.0.1:$(tcp_port "$GATEWAY_NODE")" '
+ function phase_of(at, phase) {
+ for (phase = phases; phase > 0; phase--) if (at >= start[phase]) return phase
+ return 0
+ }
+ function since(at, phase) { return sprintf("%.1f", (at - start[phase]) / 1000) }
+ function count(phase, what) { if (phase > 0) seen_in[phase, what]++ }
+ # The message of a Kafka or a tracing log line, with numbers as N.
+ function message(text) {
+ sub(/^[0-9]+ /, "", text)
+ if (!sub(/^\[[^]]*\] [A-Z]+ (\[[^]]*\] )?/, "", text)) {
+ sub(/^[^ ]+ +[A-Z]+ +[A-Za-z0-9_:]+: /, "", text)
+ }
+ sub(/ \(org[.]apache[.]kafka[^)]*\)$/, "", text)
+ gsub(/[0-9]+/, "N", text)
+ return text
+ }
+ FILENAME ~ /events.log$/ {
+ if ($2 == "stopped") { stopped[phases] = $1; next }
+ phases++
+ start[phases] = $1
+ label[phases] = $2 (NF > 2 ? " node " $3 : "")
+ note[phases] = $0
+ if (!sub(/^[^ ]+ [^ ]+ [^ ]+ /, "", note[phases])) note[phases] = ""
+ next
+ }
+ FILENAME ~ /producer.out$/ && / records sent, / && !/50th/ {
+ p = phase_of($1)
+ split($0, part, ", ")
+ split(part[2], sent, " ")
+ split(part[4], slowest, " ")
+ if (slowest[1] + 0 > max_latency[p]) max_latency[p] = slowest[1] + 0
+ if (sent[1] + 0 < rate / 2 || slowest[1] + 0 > 1000) {
+ impaired[p]++
+ recovered[p] = $1
+ }
+ next
+ }
+ FILENAME ~ /producer.out$/ && /50th/ { summary = substr($0, index($0, " ") + 1); next }
+ FILENAME ~ /producer.(out|err)$/ && /WARN/ {
+ p = phase_of($1)
+ retries[p]++
+ count(p, "producer: " message($0))
+ next
+ }
+ FILENAME ~ /producer.(out|err)$/ && $2 ~ /^org[.]apache[.]kafka[.]common[.]errors[.]/ {
+ p = phase_of($1)
+ failed[p]++
+ count(p, "producer record failed: " message($0))
+ next
+ }
+ FILENAME ~ /consumer.out$/ && /"name":"records_consumed"/ {
+ line = $0
+ while (match(line, /"partition":[0-9]+/)) {
+ partition = substr(line, RSTART + 12, RLENGTH - 12)
+ if (partition in last_read) {
+ p = phase_of($1)
+ if ($1 - last_read[partition] > read_gap[p]) read_gap[p] = $1 - last_read[partition]
+ }
+ last_read[partition] = $1
+ line = substr(line, RSTART + RLENGTH)
+ }
+ next
+ }
+ FILENAME ~ /consumer.out$/ && /"name":"record_data"/ && match($0, /"value":"[0-9]+"/) {
+ value = substr($0, RSTART + 9, RLENGTH - 10)
+ if (++read_count[value] == 1) distinct++
+ else if (read_count[value] == 2) repeated++
+ next
+ }
+ FILENAME ~ /consumer.out$/ && /"name":"offsets_committed"/ && /"success":false/ {
+ p = phase_of($1)
+ retries[p]++
+ count(p, "consumer commit failed")
+ next
+ }
+ FILENAME ~ /consumer.out$/ && /"name":"partitions_assigned"/ {
+ count(phase_of($1), "consumer partitions assigned")
+ next
+ }
+ FILENAME ~ /consumer.err$/ && /WARN/ {
+ p = phase_of($1)
+ retries[p]++
+ count(p, "consumer: " message($0))
+ next
+ }
+ FILENAME ~ /gateway.log$/ && /Connecting to / {
+ if (index($0, "Connecting to " gateway_node) == 0) moves[phase_of($1)]++
+ next
+ }
+ FILENAME ~ /gateway.log$/ && (/ (WARN|ERROR) / || / DEBUG .*(refused|failed)/) {
+ count(phase_of($1), "gateway: " message($0))
+ next
+ }
+ FILENAME ~ /node-[0-9].log$/ && /sim_event="PrimaryElected"/ {
+ p = phase_of($1)
+ if (p == 0) next
+ match($0, /replica_id=[0-9]+/)
+ node = substr($0, RSTART + 11, RLENGTH - 11)
+ if (/plane="metadata"/) {
+ count(p, "metadata primary elected on node " node " at +" since($1, p) " s")
+ } else {
+ elections[p]++
+ last_election[p] = $1
+ }
+ next
+ }
+ END {
+ printf "%-16s %7s %7s %8s %9s %8s %7s %5s\n", "phase", "start s", "max ms", "impaired", \
+ "recovered", "read gap", "retries", "moves"
+ for (p = 1; p < phases; p++) {
+ printf "%-16s %7s %7.0f %8d %9s %8.1f %7d %5d\n", label[p], since(start[p], 1), \
+ max_latency[p], impaired[p], p in recovered ? since(recovered[p], p) : "-", \
+ read_gap[p] / 1000, retries[p], moves[p]
+ }
+ print ""
+ print "max ms: slowest send. impaired: 1 s windows under half the rate or with a send over"
+ print "1 s, and recovered: seconds into the phase when the last of them ended. read gap: the"
+ print "longest time a partition went unread, in seconds. retries: client warnings and failed"
+ print "commits. moves: SDK sessions in the gateway that dialed another Iggy node."
+ for (p = 1; p < phases; p++) {
+ print ""
+ print label[p] (note[p] != "" ? ", " note[p] : "")
+ if (p in stopped) print " the node exited after " since(stopped[p], p) " s"
+ if (elections[p]) {
+ print " " elections[p] " partition primaries elected, the last at +" since(last_election[p], p) " s"
+ }
+ for (key in seen_in) {
+ split(key, at, SUBSEP)
+ if (at[1] == p) print " " seen_in[key] " x " at[2]
+ }
+ }
+ print ""
+ print "producer: " summary
+ missing = 0
+ for (value = 0; value < records; value++) if (!(value in read_count)) missing++
+ printf "consumer: read %d of %d records, %d of them more than once, %d never\n", \
+ distinct, records, repeated, missing
+ exit missing > 0 ? 1 : 0
+ }
+ ' "$EVENTS" "$WORK_DIR"/producer.* "$WORK_DIR"/consumer.* "$WORK_DIR/gateway.log" \
+ "$WORK_DIR"/node-[0-9].log || status=1
+ echo
+ cat "$WORK_DIR/group.txt"
+ echo
+ grep -E '^(read_your_commits|partition=)' "$WORK_DIR/probe.out" || cat "$WORK_DIR/probe.out"
+ grep -q '^read_your_commits .* wrong=0 ' "$WORK_DIR/probe.out" || status=1
+ echo
+ if ((status == 0)); then
+ echo "pass: the consumer read every record, and every offset read back as committed"
+ else
+ echo "fail: the consumer missed a record, or an offset read back wrong"
+ fi
+ return "$status"
+}
+
+main() {
+ preflight
+ trap cleanup EXIT
+ log "work dir $WORK_DIR"
+ printf 'RATE=%s\nRECORDS=%s\nGATEWAY_NODE=%s\nBASE_PORT=%s\n' \
+ "$RATE" "$RECORDS" "$GATEWAY_NODE" "$BASE_PORT" >"$WORK_DIR/settings"
+ local node
+ for ((node = 0; node < NODES; node++)); do
+ start_node "$node"
+ done
+ start_gateway
+ kafka_tool kafka-topics.sh --bootstrap-server "$GATEWAY_ADDR" --create --topic "$TOPIC" \
+ --partitions "$PARTITIONS"
+ env -i "IGGY_USERNAME=iggy" "IGGY_PASSWORD=iggy" "$TARGET_DIR/examples/partition_routes" \
+ "127.0.0.1:$(tcp_port "$GATEWAY_NODE")" "$STREAM" "$TOPIC" "$PARTITIONS" \
+ >"$WORK_DIR/routes.jsonl" 2>&1 &
+ ROUTES_PID=$!
+ wait_for_spread_primaries
+ log "routes $(latest_routes)"
+
+ event "load"
+ start_load
+ sleep "$WARMUP_SECS"
+ local holder
+ holder=$(primary_holder)
+ if [[ -z "$holder" ]]; then
+ echo "no node outside the gateway's and the metadata primary holds primaries: $(latest_routes)" >&2
+ exit 1
+ fi
+ outage "$holder" "a node"
+ local metadata
+ metadata=$(metadata_primary)
+ if ((metadata == GATEWAY_NODE)); then
+ log "the metadata primary is the gateway's node $metadata, so new Iggy clients fail while it is down"
+ fi
+ outage "$metadata" "the metadata primary"
+ finish_load
+ read_your_commits
+ report | tee "$WORK_DIR/report.txt"
+}
+
+if [[ "${1:-}" == "--report" ]]; then
+ report
+else
+ main
+fi
diff --git a/gateways/kafka/tests/java/ReadYourCommitsProbe.java b/gateways/kafka/tests/java/ReadYourCommitsProbe.java
new file mode 100644
index 000000000..02e8e0b75
--- /dev/null
+++ b/gateways/kafka/tests/java/ReadYourCommitsProbe.java
@@ -0,0 +1,73 @@
+/*
+ * 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.
+ */
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+
+/**
+ * Commits an offset and reads it back at once, in every partition, round after round. A read
+ * that a lagging replica answers returns the previous offset, or none after the first commit.
+ * Prints one line per wrong read and a summary, and exits with 1 if any read was wrong.
+ *
+ * <p>Arguments: bootstrap server, topic, partition count, rounds, group id.
+ */
+public final class ReadYourCommitsProbe {
+ private ReadYourCommitsProbe() {}
+
+ public static void main(String[] args) {
+ Properties props = new Properties();
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, args[0]);
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, args[4]);
+ props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+ int partitions = Integer.parseInt(args[2]);
+ int rounds = Integer.parseInt(args[3]);
+ int wrong = 0;
+ long slowestNanos = 0;
+ try (KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props)) {
+ for (int round = 1; round <= rounds; round++) {
+ for (int partition = 0; partition < partitions; partition++) {
+ TopicPartition topicPartition = new TopicPartition(args[1], partition);
+ long started = System.nanoTime();
+ consumer.commitSync(Map.of(topicPartition, new OffsetAndMetadata(round)));
+ OffsetAndMetadata read =
+ consumer.committed(Set.of(topicPartition), Duration.ofSeconds(30))
+ .get(topicPartition);
+ slowestNanos = Math.max(slowestNanos, System.nanoTime() - started);
+ if (read == null || read.offset() != round) {
+ wrong++;
+ System.out.println("partition=" + partition + " committed=" + round
+ + " read=" + (read == null ? "none" : read.offset()));
+ }
+ }
+ }
+ }
+ System.out.println("read_your_commits rounds=" + rounds + " partitions=" + partitions
+ + " wrong=" + wrong + " slowest_ms=" + slowestNanos / 1_000_000);
+ System.exit(wrong == 0 ? 0 : 1);
+ }
+}