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/07 13:25:19 UTC
(iggy) 01/01: test(gateways): cover Kafka consumer groups end to end
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-groups-e2e
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 48528f587970792632479f6fd2cdc2c12a93169f
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Wed Oct 7 18:09:50 2026 +0530
test(gateways): cover Kafka consumer groups end to end
No test drove real clients through a gateway restart or a group with
two members. A Java consumer group now resumes from its commit on a
new gateway, and when one of two members leaves, the other resumes its
partition at the leaver's commit.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
Cargo.lock | 1 +
gateways/kafka/Cargo.toml | 2 +
gateways/kafka/README.md | 20 ++
gateways/kafka/docs/MANUAL_TESTING.md | 1 +
gateways/kafka/docs/TEST_SUITE.md | 2 +-
gateways/kafka/tests/kafka_client_e2e_tests.rs | 368 ++++++++++++++++++++++++-
6 files changed, 379 insertions(+), 15 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index d63328f26..f13a1cf3c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7031,6 +7031,7 @@ dependencies = [
"lz4",
"secrecy",
"serde",
+ "serde_json",
"serial_test",
"snap",
"socket2 0.6.5",
diff --git a/gateways/kafka/Cargo.toml b/gateways/kafka/Cargo.toml
index 2acec8c75..17161adeb 100644
--- a/gateways/kafka/Cargo.toml
+++ b/gateways/kafka/Cargo.toml
@@ -73,6 +73,8 @@ zstd = { workspace = true }
crc32c = { workspace = true }
# The client half, so a test encodes requests and decodes responses the way a consumer does.
kafka-protocol = { workspace = true, features = ["client"] }
+# Reads the JSON events that kafka-verifiable-consumer.sh prints.
+serde_json = { workspace = true }
serial_test = { workspace = true }
tempfile = { workspace = true }
tokio = { workspace = true, features = [
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 2cee99fdf..796aa39a9 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -211,6 +211,26 @@ Produce request carrying a `transactional_id` gets `UNSUPPORTED_VERSION` (35) on
rather than having its records stored as if they were ordinary ones. None of those closes the
connection. `docs/SCOPE.md`'s Transactions section has the ordering and the reasoning.
+## Consumer groups ([#3544](https://github.com/apache/iggy/issues/3544))
+
+A stock consumer with a `group.id` joins, takes the partitions its own assignor hands it, commits,
+and resumes from its commit. With the bridge on, commits live in Iggy, so a group also resumes after
+a gateway restart. `tests/kafka_client_e2e_tests.rs` checks both with the Java consumer, and a
+member that leaves hands its partition to the other at its commit.
+
+Limits:
+
+- One gateway per bootstrap endpoint. Group membership is gateway memory
+ ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)).
+- Classic protocol only. A consumer with `group.protocol=consumer` (KIP-848) fails.
+- `group.instance.id` is accepted but not honoured. A static member that restarts causes two
+ rebalances.
+- No DescribeGroups or ListGroups yet ([#3548](https://github.com/apache/iggy/issues/3548)), so
+ `kafka-consumer-groups.sh` cannot list or describe a group.
+- No DeleteGroups or OffsetDelete, and offsets never expire. An operator deletes a group's offsets
+ in Iggy ([docs/OFFSET_STORAGE.md](docs/OFFSET_STORAGE.md#limits)).
+- Delivery is at-least-once. A consumer that stops before it commits reads those records again.
+
## Authentication ([#3549](https://github.com/apache/iggy/issues/3549))
Off by default. With `IGGY_KAFKA_SASL_ENABLED=true` the gateway requires SASL/PLAIN before it serves
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index 4569f50fa..301fc3a9c 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -317,6 +317,7 @@ success. This category predates the bridge landing: G2/G3's "fails at metadata"
| G8 | Static member exit | G7 with `--consumer-property group.instance.id=x` on the one stopped | No LeaveGroup is sent; the survivor waits out the session timeout before rebalancing |
| G9 | Group resumes from its commit | Bridge on and topic `orders` made as in Category I. Run `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g9 --topic orders --from-beginning`, stop it, produce one record, run it again | The second run prints only the new record. `--from-beginning` applies only to a group with no committed offset |
| G10 | Resume after a gateway restart | G9 with a gateway restart before the second run | Same as G9, because the offsets live in Iggy ([#3542](https://github.com/apache/iggy/issues/3542)) |
+| G11 | Two members share a topic | Bridge on and a topic with 2 partitions. Run `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g11 --topic <topic> --property print.partition=true --property print.offset=true` in two terminals, produce to both partitions, stop one, produce again | Each prints one partition. After the stop, the other prints the new records of both partitions, and none twice |
Record kcat version and exact error strings in your test log. G1 passing is the minimum bar for client compatibility smoke.
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index bfb73c95e..8ce0421dc 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -78,7 +78,7 @@ file under `tests/` anymore.
| [`server_e2e_tests.rs`](../tests/server_e2e_tests.rs) | Full `KafkaGateway` TCP round-trips | Partial |
| [`listener_robustness_tests.rs`](../tests/listener_robustness_tests.rs) | TCP listener robustness — framing, pipelining, concurrency, connection limits | No |
| [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No |
-| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker and a built `iggy-server` |
+| [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. The only suite that can catch a client-compatibility bug, [...]
| [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases | No (needs the `iggy-server` binary - see Prerequisites) |
| [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, codes 3 and 43 | No (needs the `iggy-server` binary - see Prerequisites) |
| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) |
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index aab4e0868..35cc5d713 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -25,24 +25,29 @@
//!
//! The stack is real on all three sides: a spawned `iggy-server` process, the gateway in-process
//! with a real `IggyAuthenticator`, and a client from a container. They automate categories S and T
-//! of `docs/MANUAL_TESTING.md`, and test G9.
+//! of `docs/MANUAL_TESTING.md`, and tests G9 to G11.
//!
//! The Fetch tests swap the authenticator for a bridge: kcat writes records through Produce, and
-//! kcat and the Java consumer read them back through Fetch.
+//! kcat and the Java consumer read them back through Fetch. The group tests run Java consumers in
+//! one group across a gateway restart and a member leaving.
//!
//! Prerequisites are Docker and an already-built `iggy-server`. Missing either skips, the way the
//! wire-fixture suites do, unless `KAFKA_E2E_REQUIRED=1` is set, which turns a skip into a failure
//! so a broken CI step cannot leave these silently green.
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
+use std::io::{BufRead, BufReader, Read};
use std::net::SocketAddr;
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
-use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, mpsc};
use std::time::{Duration, Instant};
use secrecy::SecretString;
+use serde_json::Value;
use tokio::sync::broadcast;
+use tokio::task::JoinHandle;
use iggy_gateway_kafka::auth::IggyAuthenticator;
use iggy_gateway_kafka::bridge::{
@@ -66,6 +71,9 @@ const USER_PASSWORD: &str = "s3cretpass";
const TOPIC: &str = "orders";
/// What the bridged tests write, as `(key, value)`, and read back in this order.
const RECORDS: [(&str, &str); 3] = [("k1", "alpha"), ("k2", "beta"), ("k3", "gamma")];
+/// The topic the group member test shares. Two partitions, so two members get one each.
+const GROUP_TOPIC: &str = "payments";
+const GROUP_TOPIC_PARTITIONS: u32 = 2;
/// Budget for `iggy-server` to start listening. Generous: it is a cold process start, and a debug
/// build on a loaded machine is not quick.
@@ -80,6 +88,19 @@ const CLIENT_RUN_TIMEOUT: &str = "45s";
/// unanswered request eat most of `CLIENT_RUN_TIMEOUT` retrying.
const JAVA_CLIENT_TIMEOUT_MS: u32 = 10_000;
+/// Cap on a client left running, enforced by `timeout` inside its container. Longer than
+/// `CLIENT_RUN_TIMEOUT`, because a group member lives through several rebalances.
+const RUNNING_CLIENT_TIMEOUT: &str = "120s";
+
+/// How long a test waits for the next line it expects from a running client.
+const CLIENT_EVENT_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// Above the gateway's 25s shutdown drain.
+const GATEWAY_STOP_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Numbers running client containers, so their names differ within one test process.
+static NEXT_CONTAINER: AtomicUsize = AtomicUsize::new(0);
+
/// Reports why the suite cannot run, and whether that is fatal.
///
/// Returns `true` when the caller should skip. `KAFKA_E2E_REQUIRED=1` makes it panic instead, so a
@@ -279,6 +300,33 @@ async fn spawn_gateway(iggy_address: &str) -> SocketAddr {
/// Starts the gateway with a bridge to `server` and SASL off. Creates `TOPIC` first, because
/// neither Produce nor Fetch creates one.
async fn spawn_bridged_gateway(server: &TestServer) -> SocketAddr {
+ let gateway = start_bridged_gateway(server, &[(TOPIC, 1)]).await;
+ // Held for the test's lifetime: dropping the sender shuts the gateway down mid-exchange.
+ std::mem::forget(gateway.shutdown);
+ gateway.addr
+}
+
+/// A bridged gateway the test holds. Dropping it shuts the gateway down.
+struct BridgedGateway {
+ addr: SocketAddr,
+ shutdown: broadcast::Sender<()>,
+ task: JoinHandle<()>,
+}
+
+impl BridgedGateway {
+ /// Shuts the gateway down and waits until it has stopped, so a test can start a new one in
+ /// its place.
+ async fn stop(self) {
+ drop(self.shutdown);
+ tokio::time::timeout(GATEWAY_STOP_TIMEOUT, self.task)
+ .await
+ .expect("the gateway stops within its shutdown drain")
+ .expect("the gateway task does not panic");
+ }
+}
+
+/// As [`spawn_bridged_gateway`], creating each `(topic, partitions)` first.
+async fn start_bridged_gateway(server: &TestServer, topics: &[(&str, u32)]) -> BridgedGateway {
let config = IggyBridgeConfig {
address: server.address.clone(),
username: ROOT_USER.to_string(),
@@ -290,10 +338,12 @@ async fn spawn_bridged_gateway(server: &TestServer) -> SocketAddr {
let bridge = IggyBridge::connect(config)
.await
.expect("the bridge connects to a ready server");
- bridge
- .ensure_stream_and_topic(TOPIC, 1)
- .await
- .expect("create the topic");
+ for &(topic, partitions) in topics {
+ bridge
+ .ensure_stream_and_topic(topic, partitions)
+ .await
+ .expect("create the topic");
+ }
let listener = bind_listener("127.0.0.1:0").expect("bind ephemeral port");
let addr = listener.local_addr().expect("local addr");
let config = GatewayConfig {
@@ -302,12 +352,14 @@ async fn spawn_bridged_gateway(server: &TestServer) -> SocketAddr {
};
let gateway = KafkaGateway::new(config).with_bridge(Some(Arc::new(bridge)));
let (shutdown, receiver) = broadcast::channel(1);
- tokio::spawn(async move {
+ let task = tokio::spawn(async move {
let _ = gateway.run(listener, receiver).await;
});
- // Held for the test's lifetime: dropping the sender shuts the gateway down mid-exchange.
- std::mem::forget(shutdown);
- addr
+ BridgedGateway {
+ addr,
+ shutdown,
+ task,
+ }
}
/// Runs a container against the host network and returns its combined output.
@@ -378,6 +430,134 @@ impl ClientRun {
}
}
+/// A client container left running, so the test can act while it runs. `run_client` returns only
+/// after its container exits.
+///
+/// Lines arrive over a channel as the client prints them. Dropping this removes the container,
+/// which `--rm` does not do while a client outlives a failed assertion. The `timeout` inside the
+/// container bounds it if this process dies before it can drop.
+struct RunningClient {
+ container: String,
+ process: Child,
+ lines: mpsc::Receiver<String>,
+ output: Vec<String>,
+}
+
+impl RunningClient {
+ fn start(image: &str, args: &[&str]) -> Self {
+ let container = format!(
+ "iggy-kafka-e2e-{}-{}",
+ std::process::id(),
+ NEXT_CONTAINER.fetch_add(1, Ordering::Relaxed)
+ );
+ let mut process = Command::new("docker")
+ .args([
+ "run",
+ "--rm",
+ "--pull",
+ "never",
+ "--network",
+ "host",
+ "--name",
+ ])
+ .arg(&container)
+ .arg(image)
+ .args(["timeout", "--kill-after=10s", RUNNING_CLIENT_TIMEOUT])
+ .args(args)
+ .stdout(Stdio::piped())
+ .stderr(Stdio::piped())
+ .spawn()
+ .expect("start client container");
+ let (sender, lines) = mpsc::channel();
+ forward_lines(
+ process.stdout.take().expect("stdout is piped"),
+ sender.clone(),
+ );
+ forward_lines(process.stderr.take().expect("stderr is piped"), sender);
+ Self {
+ container,
+ process,
+ lines,
+ output: Vec::new(),
+ }
+ }
+
+ /// Reads output until `found` returns a value for a line, and returns that value. Fails the
+ /// test if the client exits first or `CLIENT_EVENT_TIMEOUT` passes.
+ fn wait_for<T>(&mut self, what: &str, mut found: impl FnMut(&str) -> Option<T>) -> T {
+ let deadline = Instant::now() + CLIENT_EVENT_TIMEOUT;
+ loop {
+ let Some(line) = self.next_line(what, deadline) else {
+ panic!(
+ "{what}: the client exited first. Output:\n{}",
+ self.output.join("\n")
+ );
+ };
+ if let Some(value) = found(&line) {
+ return value;
+ }
+ }
+ }
+
+ /// Waits for the client to exit on its own, checks that it succeeded, and returns every line
+ /// it printed.
+ fn wait_exit(mut self, what: &str) -> Vec<String> {
+ let deadline = Instant::now() + CLIENT_EVENT_TIMEOUT;
+ while self.next_line(what, deadline).is_some() {}
+ let status = self.process.wait().expect("wait for the client container");
+ assert!(
+ status.success(),
+ "{what}: the client container exited with {status}. Output:\n{}",
+ self.output.join("\n")
+ );
+ std::mem::take(&mut self.output)
+ }
+
+ /// The next line the client prints, also kept in `output`, or `None` once it has exited and
+ /// every line was read. Fails the test at `deadline`.
+ fn next_line(&mut self, what: &str, deadline: Instant) -> Option<String> {
+ match self
+ .lines
+ .recv_timeout(deadline.saturating_duration_since(Instant::now()))
+ {
+ Ok(line) => {
+ self.output.push(line.clone());
+ Some(line)
+ }
+ Err(mpsc::RecvTimeoutError::Disconnected) => None,
+ Err(mpsc::RecvTimeoutError::Timeout) => panic!(
+ "{what}: nothing within {CLIENT_EVENT_TIMEOUT:?}. Output:\n{}",
+ self.output.join("\n")
+ ),
+ }
+ }
+}
+
+impl Drop for RunningClient {
+ fn drop(&mut self) {
+ // After a clean exit `--rm` has already removed it, and this fails harmlessly.
+ let _ = Command::new("docker")
+ .args(["rm", "--force"])
+ .arg(&self.container)
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .status();
+ let _ = self.process.kill();
+ let _ = self.process.wait();
+ }
+}
+
+/// Sends each line of `pipe` over `sender`, on a thread of its own, until either end closes.
+fn forward_lines(pipe: impl Read + Send + 'static, sender: mpsc::Sender<String>) {
+ std::thread::spawn(move || {
+ for line in BufReader::new(pipe).lines().map_while(Result::ok) {
+ if sender.send(line).is_err() {
+ break;
+ }
+ }
+ });
+}
+
fn kcat(addr: SocketAddr, username: &str, password: &str, mechanism: &str) -> ClientRun {
run_client(
KCAT_IMAGE,
@@ -515,6 +695,16 @@ fn ready_server() -> Option<TestServer> {
/// Writes `records` to partition 0 of `TOPIC` through the gateway, with kcat's producer and one
/// header on each record.
fn produce_with_kcat(gateway: SocketAddr, records: &[(&str, &str)]) {
+ produce_to_partition(gateway, TOPIC, 0, records);
+}
+
+/// As [`produce_with_kcat`], to `partition` of `topic`.
+fn produce_to_partition(
+ gateway: SocketAddr,
+ topic: &str,
+ partition: u32,
+ records: &[(&str, &str)],
+) {
let dir = tempfile::tempdir().expect("create records dir");
let path = dir.path().join("records.txt");
let lines: String = records
@@ -529,9 +719,9 @@ fn produce_with_kcat(gateway: SocketAddr, records: &[(&str, &str)]) {
&gateway.to_string(),
"-P",
"-t",
- TOPIC,
+ topic,
"-p",
- "0",
+ &partition.to_string(),
"-K",
":",
"-H",
@@ -988,3 +1178,153 @@ async fn given_records_when_the_java_consumer_reads_the_partition_should_get_the
"the Java consumer must stop after the records it asked for, got: {output}"
);
}
+
+/// G10 of `docs/MANUAL_TESTING.md`. The second gateway has a new bridge and an empty coordinator,
+/// as a restart leaves it, so the group's offset can only come from Iggy.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_a_group_that_committed_when_the_gateway_restarts_should_resume_after_it() {
+ let Some(server) = ready_server() else {
+ return;
+ };
+ let gateway = start_bridged_gateway(&server, &[(TOPIC, 1)]).await;
+ produce_with_kcat(gateway.addr, &RECORDS);
+ let first = consume_as_group(gateway.addr, RECORDS.len());
+ assert_eq!(
+ records_read(&first).len(),
+ RECORDS.len(),
+ "the first run must read every record, got: {first}"
+ );
+ gateway.stop().await;
+
+ let restarted = start_bridged_gateway(&server, &[(TOPIC, 1)]).await;
+ produce_with_kcat(restarted.addr, &[("k4", "delta")]);
+ let second = consume_as_group(restarted.addr, 1);
+ assert_eq!(
+ records_read(&second),
+ vec![format!("Offset:{}\tk4\tdelta", RECORDS.len())],
+ "the group must resume after its commit on a gateway that never saw it, got: {second}"
+ );
+}
+
+/// G11 of `docs/MANUAL_TESTING.md`. Two members split the topic. When one leaves, the other
+/// resumes its partition at the offset it committed: an earlier offset reads records twice, a
+/// later one skips them. Records are written only once both members hold a partition, so each
+/// reads only its own.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_two_members_of_one_group_when_one_leaves_should_let_the_other_resume_its_partition()
+{
+ let Some(server) = ready_server() else {
+ return;
+ };
+ let gateway = start_bridged_gateway(&server, &[(GROUP_TOPIC, GROUP_TOPIC_PARTITIONS)]).await;
+ let round = i64::try_from(RECORDS.len()).expect("a handful of records");
+
+ // One round on its own partition, then one on each.
+ let mut survivor = group_member(gateway.addr, 3 * round);
+ let alone = survivor.wait_for("the first member's assignment", assigned);
+ assert_eq!(alone, [0, 1], "a lone member must hold every partition");
+
+ let mut leaver = group_member(gateway.addr, round);
+ let handed_over = leaver.wait_for("the second member's assignment", assigned_one);
+ let kept = survivor.wait_for("the first member's new assignment", assigned_one);
+ assert_ne!(
+ kept, handed_over,
+ "two members of one group must hold different partitions"
+ );
+
+ produce_round(gateway.addr);
+ let leaver_output = leaver.wait_exit("the leaving member");
+ assert_eq!(
+ records_by_partition(&leaver_output),
+ BTreeMap::from([(handed_over, (0..round).collect::<Vec<_>>())]),
+ "the leaver must read its own partition and nothing else, got:\n{}",
+ leaver_output.join("\n")
+ );
+
+ survivor.wait_for("the survivor taking every partition", |line| {
+ assigned(line).filter(|partitions| *partitions == [0, 1])
+ });
+ produce_round(gateway.addr);
+ let survivor_output = survivor.wait_exit("the surviving member");
+ let read = records_by_partition(&survivor_output);
+ assert_eq!(
+ read.get(&handed_over),
+ Some(&(round..2 * round).collect::<Vec<_>>()),
+ "the survivor must resume partition {handed_over} at the leaver's commit, offset {round}, \
+ got:\n{}",
+ survivor_output.join("\n")
+ );
+ assert_eq!(
+ read.get(&kept),
+ Some(&(0..2 * round).collect::<Vec<_>>()),
+ "the survivor must read all of partition {kept}, got:\n{}",
+ survivor_output.join("\n")
+ );
+}
+
+/// Starts `kafka-verifiable-consumer.sh` as a member of group `e2e-members` on `GROUP_TOPIC`. It
+/// commits each batch synchronously after reading it, prints a JSON event for each step, and one
+/// for each record with `--verbose`. It leaves the group after `count` records.
+fn group_member(gateway: SocketAddr, count: i64) -> RunningClient {
+ RunningClient::start(
+ KAFKA_IMAGE,
+ &[
+ "/opt/kafka/bin/kafka-verifiable-consumer.sh",
+ "--bootstrap-server",
+ &gateway.to_string(),
+ "--topic",
+ GROUP_TOPIC,
+ "--group-id",
+ "e2e-members",
+ "--max-messages",
+ &count.to_string(),
+ "--verbose",
+ ],
+ )
+}
+
+/// Writes `RECORDS` to every partition of `GROUP_TOPIC`.
+fn produce_round(gateway: SocketAddr) {
+ for partition in 0..GROUP_TOPIC_PARTITIONS {
+ produce_to_partition(gateway, GROUP_TOPIC, partition, &RECORDS);
+ }
+}
+
+/// The JSON event a verifiable consumer printed on `line`, if it is named `name`.
+fn event(line: &str, name: &str) -> Option<Value> {
+ let parsed: Value = serde_json::from_str(line).ok()?;
+ (parsed["name"] == name).then_some(parsed)
+}
+
+/// The partitions a `partitions_assigned` event hands out, sorted.
+fn assigned(line: &str) -> Option<Vec<i64>> {
+ let assignment = event(line, "partitions_assigned")?;
+ let mut partitions: Vec<i64> = assignment["partitions"]
+ .as_array()?
+ .iter()
+ .filter_map(|partition| partition["partition"].as_i64())
+ .collect();
+ partitions.sort_unstable();
+ Some(partitions)
+}
+
+/// The partition a `partitions_assigned` event hands out, when it hands out exactly one.
+fn assigned_one(line: &str) -> Option<i64> {
+ match assigned(line)?.as_slice() {
+ [partition] => Some(*partition),
+ _ => None,
+ }
+}
+
+/// The offset of every record a verifiable consumer printed, by partition, in the order read.
+fn records_by_partition(output: &[String]) -> BTreeMap<i64, Vec<i64>> {
+ let mut read: BTreeMap<i64, Vec<i64>> = BTreeMap::new();
+ for record in output.iter().filter_map(|line| event(line, "record_data")) {
+ if let (Some(partition), Some(offset)) =
+ (record["partition"].as_i64(), record["offset"].as_i64())
+ {
+ read.entry(partition).or_default().push(offset);
+ }
+ }
+ read
+}