From f0ebc0d5fcbf7f7376bd42850dddca71c2f307a8 Mon Sep 17 00:00:00 2001 From: Krishna Vishal Date: Sun, 25 Oct 2026 13:37:33 +0530 Subject: [PATCH 1/2] 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 ++++++++++++++++++ .../tests/java/ReadYourCommitsProbe.java | 73 +++ 8 files changed, 761 insertions(+), 14 deletions(-) create mode 100644 gateways/kafka/examples/partition_routes.rs create mode 100755 gateways/kafka/scripts/cluster-failover.sh create mode 100644 gateways/kafka/tests/java/ReadYourCommitsProbe.java diff --git a/Cargo.lock b/Cargo.lock index ad0b60756f..6c40fa9e79 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 bd4afe645d..50f2a24c34 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 692aa6a2cb..2fb8cdc7ea 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 45a9dc702b..1f9bc50835 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 +`. 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 ab9f3365d6..67f983fd4e 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 0000000000..53a3bd6851 --- /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> { + let args: Vec = std::env::args().skip(1).collect(); + let [address, stream, topic, partitions] = args.as_slice() else { + return Err("usage: partition_routes ".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::, Box>>()?; + + 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 = routes.iter().map(|route| route.primary_replica).collect(); + let views: Vec = 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), Box> { + 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 0000000000..27a2c616b1 --- /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 +# +# 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 }" + # 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 " [ ]" 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 0000000000..02e8e0b75c --- /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. + * + *

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 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); + } +} From 49622baddcf6a964efbf67c95dae69eeda2f07f9 Mon Sep 17 00:00:00 2001 From: Krishna Vishal Date: Sun, 25 Oct 2026 13:37:53 +0530 Subject: [PATCH 2/2] 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. --- .../integration/tests/cluster/chaos_monkey.rs | 2 +- .../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 7f9950ed15..28bf9799dd 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 alone for `d`, so acknowledged writes held only by the killed pair are lost by design, the same mechanism as `full-outage`. #4476 and #4500 cover the fenced replica and the survivor election, not that loss | //! | `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` is the deterministic test. Expected: pass | //! //! `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 7dd60a1612..e4d4525744 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,