test(cluster): test the Kafka gateway and the SDK on split primaries

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

core/integration/tests/cluster/partition_primary_routing.rs

@@ -638,6 +638,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,

gateways/kafka/Cargo.toml

@@ -69,6 +69,8 @@ uuid = { workspace = true }
 x509-parser = { workspace = true }
 
 [dev-dependencies]
+# 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.

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
 

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

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
 

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))
+}