This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch kafka-empty-groups in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 6068db0b4a300596c6f924b0946cc32661b168a0 Author: Krishna Vishal <[email protected]> AuthorDate: Wed Oct 14 14:47:33 2026 +0530 feat(gateways): report groups that only hold offsets as Empty After its consumers leave, a group with offsets stays visible in ListGroups and DescribeGroups. --- gateways/kafka/README.md | 9 +- gateways/kafka/docs/CONSUMER_GROUPS.md | 64 +++- gateways/kafka/docs/MANUAL_TESTING.md | 7 +- gateways/kafka/docs/SCOPE.md | 4 +- gateways/kafka/docs/TEST_SUITE.md | 1 + gateways/kafka/docs/kafka_api_keys_reference.md | 3 +- .../kafka/src/bridge/iggy_bridge/group_index.rs | 322 ++++++++++++++++ .../kafka/src/bridge/iggy_bridge/group_offsets.rs | 111 +++++- gateways/kafka/src/bridge/iggy_bridge/mod.rs | 7 + gateways/kafka/src/bridge/mod.rs | 4 +- gateways/kafka/src/group/mod.rs | 36 ++ gateways/kafka/src/group/state.rs | 93 ++++- .../kafka/src/protocol/handlers/describe_groups.rs | 74 +++- .../kafka/src/protocol/handlers/list_groups.rs | 70 +++- gateways/kafka/src/protocol/handlers/mod.rs | 6 +- .../kafka/tests/offset_groups_real_bridge_tests.rs | 426 +++++++++++++++++++++ 16 files changed, 1167 insertions(+), 70 deletions(-) diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index 3c83aa194..324509e35 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -22,9 +22,10 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34 > Consumer group coordination is not a stub either: `FindCoordinator`, > `JoinGroup`, `Heartbeat`, > `LeaveGroup`, `SyncGroup`, `DescribeGroups` and `ListGroups` are real, with > real membership, > rebalances, graceful leave, session expiry and admin views of that membership -> ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). `kafka-consumer-groups.sh --list`, -> `--describe --members` and `--describe --state` use those views. The default `--describe` -> calls OffsetFetch (API key 9), which is not implemented, so that mode does not work. With the bridge off, +> ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). `kafka-consumer-groups.sh --list` and +> `--describe` use those views. With the bridge on, they also show a group whose consumers stopped, +> as `Empty`, while it holds offsets. The default `--describe` then prints each committed offset +> and its lag. With the bridge off, > Metadata reports every topic unknown, so a consumer joins a group and is > assigned 0 partitions. > With the bridge on, OffsetCommit and OffsetFetch keep each group's offsets > in Iggy, so a group > resumes from its last commit, also after a gateway restart @@ -234,8 +235,6 @@ Limits: - 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. diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md index 166f6acac..528e440bd 100644 --- a/gateways/kafka/docs/CONSUMER_GROUPS.md +++ b/gateways/kafka/docs/CONSUMER_GROUPS.md @@ -14,8 +14,8 @@ in Iggy ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). | 12 | Heartbeat | 0-4 | Refreshes a session; `REBALANCE_IN_PROGRESS` is how a follower learns to rejoin | | 13 | LeaveGroup | 0-5 | Removes members; survivors' parked joins and syncs are released at once | | 14 | SyncGroup | 0-5 | Relays the leader's assignment blobs; a follower parks until the leader syncs | -| 15 | DescribeGroups | 0-6 | Members, assignment and state. A group that is not here is `Dead`; v6 also returns `GROUP_ID_NOT_FOUND` (69) | -| 16 | ListGroups | 0-5 | Groups currently on this coordinator. v4+ can filter by state, v5 by type `classic` | +| 15 | DescribeGroups | 0-6 | Members, assignment and state. A group that is not here but holds offsets in Iggy is `Empty`. Otherwise it is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69) | +| 16 | ListGroups | 0-5 | Groups currently on this coordinator, plus the groups that hold offsets in Iggy as `Empty`. v4+ can filter by state, v5 by type `classic` | `kafka-protocol` can encode FindCoordinator v5 and v6 as well, and they are byte-identical to v4. They are not advertised because `SCOPE.md`'s governance model only admits a version once it has @@ -25,16 +25,14 @@ been manually tested. `DescribeGroups` and `ListGroups` read the same in-memory map the join path writes. State strings are `Empty`, `PreparingRebalance`, `CompletingRebalance` and `Stable`. `Empty` means the group -has no joined members: a v4+ client has been handed a member id and has not rejoined with it yet. -`Dead` is only the describe answer for an id that is not in the map. - -A group with neither members nor pending member ids is removed. Kafka keeps an `Empty` group -that only has committed offsets. Offsets are not stored here yet. -TODO(#3542): once offset storage lands, keep those groups and report `Empty`. +has no joined members. Either a v4+ client got a member id and did not rejoin with it yet, or +the group only holds offsets +([below](#groups-that-only-hold-offsets)). `Dead` is only the describe answer for an id +that is not in the map and holds no offsets. `ListGroups` does not tick. It leaves sessions, pending ids and phases untouched, and it omits a group whose member sessions and pending-id deadlines are all already past. `DescribeGroups` does -tick each named group, so that same id is then `Dead`. +tick each named group, so that same id is then `Dead`, or `Empty` for a group that holds offsets. While a group is not `Stable`, `DescribeGroups` leaves `protocol_data`, member metadata and member assignment empty. Kafka does the same: during `PreparingRebalance` the stored assignment @@ -48,14 +46,45 @@ second snapshot of the same members would otherwise multiply the encoded respons count. `kafka-consumer-groups.sh --list`, `--describe --members` and `--describe --state` use these -two APIs. The default `--describe` calls `OffsetFetch` (API key 9), which is not implemented, -so that mode does not work. `--reset-offsets` needs the same API. +two APIs. With a bridge, the default `--describe` also reads each committed offset with +`OffsetFetch` (9) and each end offset with `ListOffsets` (2), and prints the lag. +`--reset-offsets` writes the new offsets with `OffsetCommit` (8). Both work for a group whose +consumers stopped, because that group is `Empty` while it holds offsets. Member `client_id` and `client_host` are empty. The coordinator does not keep the request header's client id or the connection's peer address. `include_authorized_operations` is answered with the omitted sentinel. There is no group ACL bitfield. `ListGroups` v5 sets `group_type` to `classic`, which is the only protocol this coordinator runs. +## Groups that only hold offsets + +A group with neither members nor pending member ids is removed from the map. Its offsets stay in +Iggy, and Kafka keeps such a group and reports it as `Empty`. So both APIs also report each group +that holds offsets in Iggy and has no members here. Such a group is `Empty`, with protocol type +`consumer` and no members. Iggy stores no protocol type. Kafka keeps `consumer` for a group whose +consumers left, and the Java admin client takes `consumer` or an empty type as a consumer group. + +The gateway finds these groups with a scan. A scan lists the Kafka topics, then calls +`get_consumer_groups` once per topic and keeps the Iggy groups named `kafka.cg.<group>` +([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). It runs on the offset read slots, so it never waits +behind Produce. One scan answers every request for 10 seconds, because kafka-ui lists groups every +30 seconds and a scan costs one Iggy call per topic. At most one scan runs at a time, and a request +that comes while it runs waits for it. A commit through this gateway adds its group at once, so a +new group does not wait for the next scan. A group that another gateway created appears with the +next scan. + +A request waits for a scan until its offset deadline, 3 seconds by default. The scan keeps running +after that, for up to 30 seconds, and serves the next request. If a scan fails, the groups of the +last scan that found them stay the answer. If no scan has found them yet, `ListGroups` answers +`COORDINATOR_LOAD_IN_PROGRESS` (14), and `DescribeGroups` answers 14 for each named group that is +not in the map. The Java admin client retries that code. A `ListGroups` state filter without +`Empty`, or a type filter without `classic`, needs no scan. `DescribeGroups` scans only for a named +group that is not in the map. + +A negative commit deletes an offset but not the Iggy group. So a group whose offsets were all +deleted that way still shows as `Empty`. Without a bridge, no group holds offsets, and both APIs +report only the groups in the map. + ## Assignment is the client's job The gateway elects a leader, hands it every member's subscription metadata, and fans the blobs the @@ -120,12 +149,13 @@ survivors keep the current generation and learn about the rebalance through `REBALANCE_IN_PROGRESS` (27) on their next heartbeat, which the Java client handles as a graceful rejoin rather than as lost partitions. -When a group empties, by leave or by expiry, it is dropped, so the next group under that name -starts again at generation 1. The pending-member-id window above is the one case that stays and -reports `Empty`. Kafka also keeps an `Empty` group that only has committed offsets, and its -generation with it. This coordinator does not, until offset storage lands (#3542). This is -safe because member ids carry a UUID and every generation check is preceded by a membership -check, so a stale client is told `UNKNOWN_MEMBER_ID` before its generation is ever compared. +When a group empties, by leave or by expiry, it is dropped, so the next group under that name starts +again at generation 1. The pending-member-id window above is the one case that stays and reports +`Empty`. Kafka also keeps an `Empty` group that only holds committed offsets, and its generation +with it. This gateway reports that group as `Empty` from its offsets in Iggy ([Groups that only hold +offsets](#groups-that-only-hold-offsets)), but keeps no generation for it. This is safe because +member ids carry a UUID and every generation check is preceded by a membership check. So a stale +client is told `UNKNOWN_MEMBER_ID` before its generation is ever compared. ## Leaving a group diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md index 36a6602cc..85b51d17a 100644 --- a/gateways/kafka/docs/MANUAL_TESTING.md +++ b/gateways/kafka/docs/MANUAL_TESTING.md @@ -326,9 +326,10 @@ success. This category predates the bridge landing: G2/G3's "fails at metadata" | 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 | -| G12 | List groups | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --list` after G5 has joined | Prints the group. `--list --state Empty` matches a group that was issued a member id and has not joined yet. A group whose members have all expired is absent | +| G12 | List groups | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --list` after G5 joined | Prints the group. `--list --state Empty` matches a group that was issued a member id and did not join yet. With the bridge on, it also matches a group whose members left or expired and that holds offsets. A group with neither members nor offsets is absent | | G13 | Describe members | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --members --group g2` | Prints member ids. Assignment and protocol are filled only while the group is Stable | -| G14 | Describe state | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --state --group g2` | Prints the state string (`Stable`, `PreparingRebalance`, `CompletingRebalance`, or `Empty`). The default `--describe` (no `--members` or `--state`) calls OffsetFetch and fails, because that API is not implemented | +| G14 | Describe state | `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --describe --state --group g2` | Prints the state string (`Stable`, `PreparingRebalance`, `CompletingRebalance`, or `Empty`). With the bridge on, the default `--describe` (no `--members` or `--state`) prints each committed offset and its lag | +| G15 | Group with no live members | Bridge on. After G9's second run ends, `kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9093 --list --state`, then `--describe --group g9` | The group is listed as `Empty`. `--describe` prints `Consumer group 'g9' has no active members.` and then each committed offset and its lag | Record kcat version and exact error strings in your test log. G1 passing is the minimum bar for client compatibility smoke. @@ -458,7 +459,7 @@ kcat version (if used): ___________ [ ] D1–D10 Flexible vs legacy encoding [ ] E1–E4 Metadata stub semantics [ ] F1–F6 TCP / connection behavior -[ ] G1–G11 kcat / Java client / kafka-consumer-groups.sh (record errors for G2/G3; G11's default `--describe` is expected to fail) +[ ] G1–G15 kcat / Java client / kafka-consumer-groups.sh (record errors for G2/G3) [ ] H1–H6 Adversarial input [ ] I1–I6 Docker Compose quick start (record wall-clock time to I3; I5 is restart durability) diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md index b8cd2d962..b38ff3099 100644 --- a/gateways/kafka/docs/SCOPE.md +++ b/gateways/kafka/docs/SCOPE.md @@ -63,8 +63,8 @@ it knows the server supports flexible encoding. | 12 | Heartbeat | 0 | 4 | 0, 1, 2, 3, 4 | Refreshes a session; `REBALANCE_IN_PROGRESS` (27) drives a rejoin; flexible encoding at v4+ | | 13 | LeaveGroup | 0 | 5 | 0, 1, 2, 3, 4, 5 | Removes members, per-member errors from v3; flexible encoding at v4+ | | 14 | SyncGroup | 0 | 5 | 0, 1, 2, 3, 4, 5 | Relays the leader's assignment blobs; flexible encoding at v4+ | -| 15 | DescribeGroups | 0 | 6 | 0, 1, 2, 3, 4, 5, 6 | Members, assignment and state of groups on this coordinator; a missing group is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69); flexible encoding at v5+ | -| 16 | ListGroups | 0 | 5 | 0, 1, 2, 3, 4, 5 | Groups on this coordinator; v4+ state filter and v5 type filter; flexible encoding at v3+ | +| 15 | DescribeGroups | 0 | 6 | 0, 1, 2, 3, 4, 5, 6 | Members, assignment and state of groups on this coordinator. A group that is not there but holds offsets in Iggy is `Empty`. Any other missing group is `Dead`, and v6 also returns `GROUP_ID_NOT_FOUND` (69). Flexible encoding at v5+ | +| 16 | ListGroups | 0 | 5 | 0, 1, 2, 3, 4, 5 | Groups on this coordinator, plus the groups that hold offsets in Iggy as `Empty` (see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)). v4+ state filter and v5 type filter. Flexible encoding at v3+ | | 22 | InitProducerId | 0 | 5 | 0, 1, 2, 3, 4, 5 | Allocate a producer id (epoch 0); a `transactional_id` gets `UNSUPPORTED_VERSION` (35); flexible encoding at v2+ | | 32 | DescribeConfigs | 1 | 4 | 1, 2, 3, 4 | Topic resources only. Returns `retention.ms` and `cleanup.policy`. Flexible encoding at v4. See [`CONFIGS.md`](CONFIGS.md) | | 33 | AlterConfigs | 0 | 2 | 0, 1, 2 | Persists `retention.ms` only. Flexible encoding at v2. See [`CONFIGS.md`](CONFIGS.md) | diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index 0a62468fd..292354edc 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -84,6 +84,7 @@ file under `tests/` anymore. | [`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) | | [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition | No (needs the `iggy-server` binary - see Prerequisites) | | [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) | +| [`offset_groups_real_bridge_tests.rs`](../tests/offset_groups_real_bridge_tests.rs) | ListGroups (key 16) and DescribeGroups (key 15) through the whole handler against a real, spawned `iggy-server`. A group with offsets and no members is `Empty`, and a live group is listed once. Also the filters, a commit after the last scan, a new gateway, a native Iggy group, and `COORDINATOR_LOAD_IN_PROGRESS` (14) before any scan | No (needs the `iggy-server` binary - see Prerequisites) | | [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) | [#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics → Metadata → Produce → ListOffsets over real TCP against the real compiled `iggy-gateway-kafka` binary bridged to a real spawned `iggy-server` - the only suite that spawns the gateway as a process rather than calling `KafkaGateway::run` in-process | No (needs both binaries - see Prerequisites) | `tests/common/` holds shared helpers (`codec.rs`, `fixtures.rs`, `gateway_process.rs`, `scope.rs`, diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md index 59f4f2f9d..b75568210 100644 --- a/gateways/kafka/docs/kafka_api_keys_reference.md +++ b/gateways/kafka/docs/kafka_api_keys_reference.md @@ -92,7 +92,8 @@ Key new minimums: | 16 | **ListGroups** | 1 | 5 | v3 | Implemented | | 42 | **DeleteGroups** | 1 | 2 | v2 | 🟡 Optional Stub | -> DescribeGroups and ListGroups are implemented. They read this coordinator's in-memory groups. +> DescribeGroups and ListGroups are implemented. They read this coordinator's in-memory groups, and +> report a group that only holds offsets in Iggy as `Empty`. > See [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md). --- diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_index.rs b/gateways/kafka/src/bridge/iggy_bridge/group_index.rs new file mode 100644 index 000000000..bf6023db2 --- /dev/null +++ b/gateways/kafka/src/bridge/iggy_bridge/group_index.rs @@ -0,0 +1,322 @@ +// 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. + +//! The Kafka groups that hold offsets in Iggy, for `ListGroups` and `DescribeGroups`. Finding them +//! costs one Iggy call per topic, so one scan answers every request for a while. + +use std::collections::BTreeSet; +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::{Semaphore, watch}; +use tokio::time::{Instant, timeout_at}; +use tracing::warn; + +use crate::bridge::error::BridgeError; + +/// How long one scan answers for. kafka-ui lists groups every 30 s, and a scan costs one Iggy call +/// per topic. +const SCAN_MAX_AGE: Duration = Duration::from_secs(10); + +/// The Kafka groups one scan found holding offsets in Iggy. +#[derive(Debug, Clone)] +pub struct OffsetGroups { + /// When the scan began. + started: Instant, + groups: BTreeSet<String>, +} + +impl OffsetGroups { + #[must_use] + pub fn contains(&self, group_id: &str) -> bool { + self.groups.contains(group_id) + } + + /// The group ids, in order. + pub fn iter(&self) -> impl Iterator<Item = &str> { + self.groups.iter().map(String::as_str) + } + + fn is_fresh(&self) -> bool { + self.started.elapsed() < SCAN_MAX_AGE + } +} + +/// The groups of the last scan that ended, and at most one scan that runs. +pub(super) struct GroupIndex { + published: Arc<watch::Sender<Published>>, + /// One permit, held by the scan that runs. + scanning: Arc<Semaphore>, +} + +#[derive(Default)] +struct Published { + /// Scans that ended, with groups or without. A caller waits for this count to move. + ended: u64, + /// The groups of the last scan that found them. + groups: Option<Arc<OffsetGroups>>, +} + +impl GroupIndex { + pub(super) fn new() -> Self { + Self { + published: Arc::new(watch::Sender::new(Published::default())), + scanning: Arc::new(Semaphore::new(1)), + } + } + + /// The groups of a scan that began less than `SCAN_MAX_AGE` ago. Otherwise starts a scan with + /// `scan`, unless one runs, and waits for the scan that runs until `deadline`. + /// + /// A scan that fails, or that does not end by `deadline`, leaves the groups of the last scan + /// that found them, and those are the answer. `None` when no scan found groups yet. + pub(super) async fn groups<F, S>(&self, deadline: Instant, scan: F) -> Option<Arc<OffsetGroups>> + where + F: FnOnce() -> S, + S: Future<Output = Result<BTreeSet<String>, BridgeError>> + Send + 'static, + { + let mut published = self.published.subscribe(); + let ended = { + let current = published.borrow_and_update(); + if let Some(groups) = current.groups.as_ref().filter(|groups| groups.is_fresh()) { + return Some(Arc::clone(groups)); + } + current.ended + }; + self.start(scan); + // A timeout leaves the groups of the last scan as the answer. + let _ = timeout_at( + deadline, + published.wait_for(|current| current.ended > ended), + ) + .await; + published.borrow().groups.clone() + } + + /// Adds `group` to the groups of the last scan. A commit of this gateway created its offsets, + /// so it shows before the next scan. + pub(super) fn note(&self, group: &str) { + self.published.send_if_modified(|current| { + let Some(groups) = current.groups.as_mut() else { + return false; + }; + if groups.contains(group) { + return false; + } + Arc::make_mut(groups).groups.insert(group.to_owned()); + true + }); + } + + /// Runs `scan` in a task of its own, so it ends even when no caller waits for it any more. + /// Does nothing while another scan runs. + fn start<F, S>(&self, scan: F) + where + F: FnOnce() -> S, + S: Future<Output = Result<BTreeSet<String>, BridgeError>> + Send + 'static, + { + let Ok(permit) = Arc::clone(&self.scanning).try_acquire_owned() else { + return; + }; + let published = Arc::clone(&self.published); + let started = Instant::now(); + let scanning = scan(); + tokio::spawn(async move { + let found = scanning.await; + if let Err(error) = &found { + warn!(%error, "could not list the Kafka groups that hold offsets in Iggy"); + } + published.send_modify(|current| { + current.ended += 1; + if let Ok(groups) = found { + current.groups = Some(Arc::new(OffsetGroups { started, groups })); + } + }); + drop(permit); + }); + } +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use iggy::prelude::IggyError; + use tokio::sync::oneshot; + + use super::*; + + fn later() -> Instant { + Instant::now() + Duration::from_secs(5) + } + + fn found(groups: &[&str]) -> BTreeSet<String> { + groups.iter().map(|group| (*group).to_owned()).collect() + } + + fn names(groups: Option<Arc<OffsetGroups>>) -> Vec<String> { + groups + .expect("a scan found groups") + .iter() + .map(str::to_owned) + .collect() + } + + /// A scan that counts its runs and finds `groups`. + fn counted( + runs: &Arc<AtomicUsize>, + groups: &'static [&'static str], + ) -> impl FnOnce() -> std::future::Ready<Result<BTreeSet<String>, BridgeError>> { + let runs = Arc::clone(runs); + move || { + runs.fetch_add(1, Ordering::SeqCst); + std::future::ready(Ok(found(groups))) + } + } + + #[tokio::test(start_paused = true)] + async fn given_a_fresh_scan_should_answer_from_it_without_scanning_again() { + let index = GroupIndex::new(); + let runs = Arc::new(AtomicUsize::new(0)); + + let first = index.groups(later(), counted(&runs, &["orders"])).await; + tokio::time::advance(SCAN_MAX_AGE.saturating_sub(Duration::from_millis(1))).await; + let second = index.groups(later(), counted(&runs, &["payments"])).await; + + assert_eq!(names(first), ["orders"]); + assert_eq!(names(second), ["orders"]); + assert_eq!(runs.load(Ordering::SeqCst), 1); + } + + #[tokio::test(start_paused = true)] + async fn given_a_scan_older_than_its_max_age_should_scan_again() { + let index = GroupIndex::new(); + let runs = Arc::new(AtomicUsize::new(0)); + + index.groups(later(), counted(&runs, &["orders"])).await; + tokio::time::advance(SCAN_MAX_AGE).await; + let second = index.groups(later(), counted(&runs, &["payments"])).await; + + assert_eq!(names(second), ["payments"]); + assert_eq!(runs.load(Ordering::SeqCst), 2); + } + + /// Callers that come while a scan runs wait for that scan rather than start their own. + #[tokio::test(start_paused = true)] + async fn given_callers_during_a_scan_should_share_that_scan() { + let index = GroupIndex::new(); + let runs = Arc::new(AtomicUsize::new(0)); + let (release, released) = oneshot::channel::<()>(); + let first_runs = Arc::clone(&runs); + let blocked = move || { + first_runs.fetch_add(1, Ordering::SeqCst); + async move { + let _ = released.await; + Ok(found(&["orders"])) + } + }; + + let (first, second, ()) = tokio::join!( + index.groups(later(), blocked), + index.groups(later(), counted(&runs, &["payments"])), + async { + tokio::task::yield_now().await; + release.send(()).expect("the scan waits for its release"); + }, + ); + + assert_eq!(names(first), ["orders"]); + assert_eq!(names(second), ["orders"]); + assert_eq!(runs.load(Ordering::SeqCst), 1); + } + + #[tokio::test(start_paused = true)] + async fn given_a_failed_scan_should_answer_the_groups_of_the_last_scan_that_found_them() { + let index = GroupIndex::new(); + let runs = Arc::new(AtomicUsize::new(0)); + index.groups(later(), counted(&runs, &["orders"])).await; + tokio::time::advance(SCAN_MAX_AGE).await; + + let failed = index + .groups(later(), || async { + Err(BridgeError::Iggy(IggyError::CannotEstablishConnection)) + }) + .await; + + assert_eq!(names(failed), ["orders"]); + } + + #[tokio::test(start_paused = true)] + async fn given_no_scan_that_found_groups_when_a_scan_fails_should_answer_none() { + let index = GroupIndex::new(); + + let failed = index + .groups(later(), || async { + Err(BridgeError::Iggy(IggyError::CannotEstablishConnection)) + }) + .await; + + assert!(failed.is_none()); + } + + /// A scan past the deadline of its caller keeps running, and the next caller gets its groups + /// without another scan. + #[tokio::test(start_paused = true)] + async fn given_a_scan_past_the_deadline_should_finish_for_the_next_caller() { + let index = GroupIndex::new(); + let runs = Arc::new(AtomicUsize::new(0)); + let (release, released) = oneshot::channel::<()>(); + + let late = index + .groups( + Instant::now() + Duration::from_millis(20), + move || async move { + let _ = released.await; + Ok(found(&["orders"])) + }, + ) + .await; + assert!(late.is_none(), "no scan found groups by the deadline"); + + release.send(()).expect("the scan still runs"); + tokio::task::yield_now().await; + let next = index.groups(later(), counted(&runs, &["payments"])).await; + + assert_eq!(names(next), ["orders"]); + assert_eq!( + runs.load(Ordering::SeqCst), + 0, + "the scan that ended is fresh" + ); + } + + #[tokio::test(start_paused = true)] + async fn given_a_noted_group_should_answer_it_before_the_next_scan() { + let index = GroupIndex::new(); + let runs = Arc::new(AtomicUsize::new(0)); + index.note("before-any-scan"); + index.groups(later(), counted(&runs, &["orders"])).await; + + index.note("payments"); + index.note("orders"); + let groups = index.groups(later(), counted(&runs, &[])).await; + + assert_eq!(names(groups), ["orders", "payments"]); + assert_eq!(runs.load(Ordering::SeqCst), 1); + } +} diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs index 3f71a0949..a22d8cc9d 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs @@ -17,9 +17,11 @@ //! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`. +use std::collections::BTreeSet; use std::future::Future; use std::hash::{DefaultHasher, Hash, Hasher}; use std::sync::Arc; +use std::time::Duration; use futures::future::join_all; use iggy::prelude::{ @@ -27,17 +29,24 @@ use iggy::prelude::{ }; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use tokio::time::{Instant, timeout, timeout_at}; -use tracing::info; +use tracing::{debug, info}; use super::{ - IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget, in_slot, permit_by, + IggyBridge, KafkaTopicMetadata, LazyClient, OffsetGroups, SLOT_LIMIT, TopicTarget, in_slot, + permit_by, }; use crate::bridge::error::BridgeError; +use crate::group::is_valid_group_id; +use crate::protocol::api::ERROR_UNKNOWN_TOPIC_OR_PARTITION; /// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group /// apart from a native Iggy group of the same name. pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg."; +/// How long one scan for the groups that hold offsets may take. It keeps running after the request +/// that started it stops waiting. +const GROUP_SCAN_LIMIT: Duration = Duration::from_secs(30); + /// Offset calls that run at once, each slot on clients of its own, so offset calls never wait /// behind a Produce send. const OFFSET_SLOTS: usize = 4; @@ -90,7 +99,8 @@ impl IggyBridge { /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already /// applied counts as done. /// - /// Commits on one slot start one at a time, in the order given. + /// Commits on one slot start one at a time, in the order given. A stored offset adds `group` + /// to [`Self::offset_groups`] at once. /// /// # Errors /// @@ -109,7 +119,7 @@ impl IggyBridge { }; // `join_all` polls each call once, in order, before any again, so calls on one slot // queue for it in order. - join_all(commits.iter().map(|&(topic, partition, offset)| { + let results = join_all(commits.iter().map(|&(topic, partition, offset)| { let (names, stream_id, topic_id) = ( Arc::clone(&names), topic.stream_id.clone(), @@ -120,7 +130,15 @@ impl IggyBridge { }; self.commit_in_slot(slot_of(topic, partition), calls, commit) })) - .await + .await; + let stored = commits + .iter() + .zip(&results) + .any(|(&(_, _, offset), result)| offset >= 0 && result.is_ok()); + if stored { + self.group_index.note(group); + } + results } /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in @@ -200,6 +218,75 @@ impl IggyBridge { self.read_in_slot(LISTING_SLOT, calls, listing).await } + /// The Kafka groups that hold offsets in Iggy, from a scan that began less than 10 s ago, plus + /// the groups that this gateway stored offsets for since. A scan lists the Kafka topics, then + /// the Iggy groups of each topic on the offset read slots, so it costs one call per topic. + /// + /// Callers that come while a scan runs wait for it, until `deadline`. The scan keeps running + /// after that and serves the next caller. A scan that fails, or does not end in time, leaves + /// the groups of the last scan that found them as the answer. `None` when no scan found them + /// yet. + pub async fn offset_groups(self: &Arc<Self>, deadline: Instant) -> Option<Arc<OffsetGroups>> { + let bridge = Arc::clone(self); + let scan = move || async move { + let calls = OffsetCalls::new(Instant::now() + GROUP_SCAN_LIMIT); + bridge.scan_offset_groups(&calls).await + }; + self.group_index.groups(deadline, scan).await + } + + /// Every Kafka group with offsets on a Kafka topic. A topic deleted since the listing holds + /// none, and so does one the bridge user may not read. + async fn scan_offset_groups( + &self, + calls: &OffsetCalls, + ) -> Result<BTreeSet<String>, BridgeError> { + let topics = self.list_kafka_topics_in_slot(calls).await?; + let listings = join_all( + topics + .iter() + .map(|topic| self.offset_groups_on(&topic.kafka_topic, calls)), + ) + .await; + let mut groups = BTreeSet::new(); + for (topic, listing) in topics.iter().zip(listings) { + match listing { + Ok(found) => groups.extend(found), + Err(BridgeError::Iggy(IggyError::Unauthorized)) => { + debug!( + topic = topic.kafka_topic.as_str(), + "skipping a topic the bridge user may not read in the scan for groups" + ); + } + Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => {} + Err(error) => return Err(error), + } + } + Ok(groups) + } + + /// The Kafka groups with offsets on `kafka_topic`. + async fn offset_groups_on( + &self, + kafka_topic: &str, + calls: &OffsetCalls, + ) -> Result<Vec<String>, BridgeError> { + let topic = self.topic_target(kafka_topic)?; + // Any slot will do. Partition 0's spreads the topics over the slots. + let slot = slot_of(&topic, 0); + let listing = |client: Arc<IggyClient>| async move { + let groups = client + .get_consumer_groups(&topic.stream_id, &topic.topic_id) + .await + .map_err(BridgeError::Iggy)?; + Ok(groups + .iter() + .filter_map(|group| kafka_group_of(&group.name)) + .collect()) + }; + self.read_in_slot(slot, calls, listing).await + } + /// Runs `call` with the commit client of `slot`, and waits for it until the deadline of /// `calls`. The call keeps the slot until it ends, or for at most `SLOT_LIMIT`, even after its /// caller stops waiting. A call that has not started by the deadline never starts. @@ -336,6 +423,13 @@ impl OffsetNames { } } +/// The Kafka group whose offsets the Iggy group `name` holds. `None` for any other Iggy group. +fn kafka_group_of(name: &str) -> Option<String> { + name.strip_prefix(OFFSET_GROUP_PREFIX) + .filter(|group| is_valid_group_id(group)) + .map(str::to_owned) +} + /// One commit on `client`. A store that finds no Iggy group creates it and stores again. async fn commit_on( client: &IggyClient, @@ -481,6 +575,13 @@ mod tests { ); } + #[test] + fn given_an_iggy_group_name_only_an_offset_group_should_name_a_kafka_group() { + assert_eq!(kafka_group_of("kafka.cg.orders").as_deref(), Some("orders")); + assert_eq!(kafka_group_of("orders"), None, "a native Iggy group"); + assert_eq!(kafka_group_of("kafka.cg."), None, "an empty Kafka group id"); + } + #[test] fn given_a_kafka_group_the_key_should_be_its_prefixed_external_group() { let names = OffsetNames::new("orders").unwrap(); diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs index fc068e8ec..d9bbc94c0 100644 --- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs +++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs @@ -32,6 +32,7 @@ use crate::bridge::error::BridgeError; use crate::bridge::topic_map::validate_kafka_topic_name; mod fetch; +mod group_index; mod group_offsets; mod offsets; mod produce; @@ -39,6 +40,8 @@ mod topics; use fetch::{FetchPool, LazyClient}; pub(crate) use fetch::{FetchSlot, PartitionProbe, TopicProbe}; +use group_index::GroupIndex; +pub use group_index::OffsetGroups; use group_offsets::OffsetPool; pub use group_offsets::{OFFSET_GROUP_PREFIX, OffsetCalls}; use produce::ProducePool; @@ -150,6 +153,8 @@ pub struct IggyBridge { probe_client: LazyClient, /// Group offset slots, each with a commit client and a read client. See `OffsetPool`. offset_pool: OffsetPool, + /// The Kafka groups that hold offsets. See `offset_groups`. + group_index: GroupIndex, } /// The Iggy stream and topic one Kafka topic maps to. Resolve once per topic, use many times. @@ -190,6 +195,7 @@ impl IggyBridge { fetch_pool: Arc::new(FetchPool::new()), probe_client: LazyClient::default(), offset_pool: OffsetPool::new(), + group_index: GroupIndex::new(), }) } @@ -289,6 +295,7 @@ impl IggyBridge { fetch_pool: Arc::new(FetchPool::new()), probe_client: LazyClient::default(), offset_pool: OffsetPool::new(), + group_index: GroupIndex::new(), } } } diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs index 8b25a8ae7..5bb822400 100644 --- a/gateways/kafka/src/bridge/mod.rs +++ b/gateways/kafka/src/bridge/mod.rs @@ -29,7 +29,7 @@ pub mod topic_map; pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig, ProduceSessionLimits}; pub use error::BridgeError; pub use iggy_bridge::{ - IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, ProduceTurn, Producer, - StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget, + IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, OffsetCalls, OffsetGroups, ProduceTurn, + Producer, StreamTopicCache, TopicCreationOutcome, TopicLoad, TopicTarget, }; pub use topic_map::{TopicMapping, TopicOverride}; diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs index 4af5d481a..5b6ed3f0d 100644 --- a/gateways/kafka/src/group/mod.rs +++ b/gateways/kafka/src/group/mod.rs @@ -53,6 +53,13 @@ pub const GROUP_STATE_COMPLETING_REBALANCE: &str = "CompletingRebalance"; /// `DescribeGroups` / `ListGroups` state string for a group whose assignment is in effect. pub const GROUP_STATE_STABLE: &str = "Stable"; +/// Protocol type of a group that only holds offsets in Iggy. +/// +/// Kafka keeps `consumer` for a group whose consumers left, and Iggy stores no protocol type to say +/// otherwise. Kafka's admin client takes a group as a consumer group with `consumer` or an empty +/// type. +pub const OFFSETS_ONLY_PROTOCOL_TYPE: &str = "consumer"; + /// v6 `error_message` when one `DescribeGroups` group does not fit the response budget. /// /// `UNKNOWN_SERVER_ERROR` is not retried by `kafka-consumer-groups.sh`, so the tool prints this @@ -358,6 +365,19 @@ pub struct GroupListing { pub state: &'static str, } +impl GroupListing { + /// A group with no members here that holds offsets in Iggy. Kafka keeps a group whose + /// consumers left, and lists it as `Empty`. + #[must_use] + pub const fn offsets_only(group_id: StrBytes) -> Self { + Self { + group_id, + protocol_type: StrBytes::from_static_str(OFFSETS_ONLY_PROTOCOL_TYPE), + state: GROUP_STATE_EMPTY, + } + } +} + /// One group the coordinator currently holds, as `DescribeGroups` reports it. #[derive(Debug, Clone)] pub struct GroupDescription { @@ -374,6 +394,22 @@ pub struct GroupDescription { pub members: Vec<MemberDescription>, } +impl GroupDescription { + /// [`GroupListing::offsets_only`] as `DescribeGroups` reports it: `Empty`, with no protocol + /// and no members. + #[must_use] + pub const fn offsets_only(group_id: StrBytes) -> Self { + Self { + group_id, + error: ERROR_NONE, + state: GROUP_STATE_EMPTY, + protocol_type: StrBytes::from_static_str(OFFSETS_ONLY_PROTOCOL_TYPE), + protocol_name: None, + members: Vec::new(), + } + } +} + /// The answer for one `LeavingMember`. #[derive(Debug, Clone)] pub struct LeftMember { diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs index 9106e6daa..b7237b035 100644 --- a/gateways/kafka/src/group/state.rs +++ b/gateways/kafka/src/group/state.rs @@ -23,8 +23,8 @@ //! across an `.await`. //! //! A group with no joined members is reported as Kafka's `Empty`. One with neither members nor -//! pending member ids is dropped: offsets live in Iggy, so nothing here is worth keeping. -//! TODO(#3542): keep an offsets-only group and report `Empty` once those offsets are stored here. +//! pending member ids is dropped. Its offsets stay in Iggy, and the `ListGroups` and +//! `DescribeGroups` handlers report it as `Empty` from there. use std::collections::{BTreeMap, HashMap, HashSet}; use std::time::Duration; @@ -39,8 +39,8 @@ use crate::group::{ CommitRequest, DESCRIBE_RESPONSE_TOO_LARGE, GROUP_STATE_COMPLETING_REBALANCE, GROUP_STATE_EMPTY, GROUP_STATE_PREPARING_REBALANCE, GROUP_STATE_STABLE, GroupCoordinatorConfig, GroupDescription, GroupListing, GroupMember, JoinRequest, JoinResult, JoinedMember, - LeaveRequest, LeaveResult, LeavingMember, LeftMember, MemberDescription, SyncRequest, - SyncResult, owned_str, + LeaveRequest, LeaveResult, LeavingMember, LeftMember, MemberDescription, + OFFSETS_ONLY_PROTOCOL_TYPE, SyncRequest, SyncResult, owned_str, }; use crate::protocol::api::{ API_KEY_DESCRIBE_GROUPS, ERROR_COORDINATOR_NOT_AVAILABLE, ERROR_FENCED_INSTANCE_ID, @@ -1335,7 +1335,8 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) -> /// How one requested id is answered. Priced before [`GroupState::description`] copies blobs. #[derive(Clone, Copy)] enum DescribedChoice { - /// Not in the map after the tick. Encoded as `Dead`. + /// Not in the map after the tick. Encoded as `Empty` when the group holds offsets in Iggy, + /// and as `Dead` otherwise. Missing, /// `summary` is Kafka's `summaryNoMetadata` shape. Present { summary: bool }, @@ -1519,9 +1520,16 @@ fn too_large_described_len(version: i16, group_id: &StrBytes) -> usize { total } -/// A group that is not in the map. v6 carries `GROUP_ID_NOT_FOUND` and `Group {id} not found.`; -/// older versions encode error 0 and omit `error_message`. Matches `missing_group` in the handler. +/// A group that is not in the map. The handler answers it after this lock is released, `Empty` +/// when it holds offsets in Iggy and `Dead` otherwise, so it is priced as the larger answer. A +/// `COORDINATOR_LOAD_IN_PROGRESS` answer, for when the handler cannot tell, is smaller than both. fn missing_described_len(version: i16, group_id: &StrBytes) -> usize { + dead_described_len(version, group_id).max(offsets_only_described_len(version, group_id)) +} + +/// v6 carries `GROUP_ID_NOT_FOUND` and `Group {id} not found.`. Older versions encode error 0 and +/// omit `error_message`. Matches `missing_group` in the handler. +fn dead_described_len(version: i16, group_id: &StrBytes) -> usize { let flexible = version >= 5; let mut total = 2; if version >= 6 { @@ -1542,6 +1550,28 @@ fn missing_described_len(version: i16, group_id: &StrBytes) -> usize { total } +/// Error 0 with a null v6 `error_message`, `Empty`, [`OFFSETS_ONLY_PROTOCOL_TYPE`], no protocol +/// and no members. Matches [`GroupDescription::offsets_only`] through `described_group`. +fn offsets_only_described_len(version: i16, group_id: &StrBytes) -> usize { + let flexible = version >= 5; + let mut total = 2; + if version >= 6 { + total += kafka_nullable_string_len(true, None); + } + total += kafka_string_len(flexible, group_id.len()); + total += kafka_string_len(flexible, GROUP_STATE_EMPTY.len()); + total += kafka_string_len(flexible, OFFSETS_ONLY_PROTOCOL_TYPE.len()); + total += kafka_string_len(flexible, 0); + total += kafka_array_prefix(flexible, 0); + if version >= 3 { + total += 4; + } + if flexible { + total += 1; + } + total +} + fn kafka_response_header_len(api_key: i16, version: i16) -> usize { if response_header_version(api_key, version) >= 1 { 5 @@ -1714,6 +1744,8 @@ fn generate_member_id(group_instance_id: Option<&StrBytes>) -> StrBytes { #[cfg(test)] mod tests { + use kafka_protocol::messages::describe_groups_response::DescribedGroup; + use super::*; use crate::protocol::handlers::describe_groups::{ described_group, encode_response, missing_group, @@ -2375,6 +2407,9 @@ mod tests { } } + /// A missing id is answered `Dead`, or `Empty` when it holds offsets in Iggy. One of the two + /// is larger for every id at a given version, `Empty` up to v5 and `Dead` at v6, where it + /// carries a message. So the price is the larger of the two whole responses. fn assert_priced_response( version: i16, groups: &mut Groups, @@ -2384,28 +2419,35 @@ mod tests { ) { let priced = describe_frame_len(version, groups, ids, max_frame_size); let described = describe_groups(groups, ids, version, max_frame_size, now); - let encoded = encode_described(version, &dedup_first_seen(ids), &described); + let ids = dedup_first_seen(ids); let header = kafka_response_header_len(API_KEY_DESCRIBE_GROUPS, version); + let dead = header + encode_described(version, &ids, &described, missing_group).len(); + let offsets_only = + header + encode_described(version, &ids, &described, offsets_only_group).len(); assert_eq!( priced, - header + encoded.len(), - "version {version} ids {} priced {priced} encoded {}", + dead.max(offsets_only), + "version {version} ids {} priced {priced} dead {dead} offsets only {offsets_only}", ids.len(), - header + encoded.len() ); } + fn offsets_only_group(version: i16, group_id: &StrBytes) -> DescribedGroup { + described_group(version, GroupDescription::offsets_only(group_id.clone())) + } + fn encode_described( version: i16, ids: &[StrBytes], described: &[Option<GroupDescription>], + missing: fn(i16, &StrBytes) -> DescribedGroup, ) -> Bytes { let groups = ids .iter() .zip(described) .map(|(group_id, group)| { group.clone().map_or_else( - || missing_group(version, group_id), + || missing(version, group_id), |group| described_group(version, group), ) }) @@ -2413,6 +2455,33 @@ mod tests { encode_response(version, groups).expect("described group encodes") } + /// Each answer to a missing id on its own, so neither hides behind the other in the maximum. + /// The id is past the compact varint boundary. + #[test] + fn given_a_missing_group_when_pricing_each_answer_should_match_the_encoder() { + let group_id = StrBytes::from_string("m".repeat(200)); + for version in 0..=6 { + let header = kafka_response_header_len(API_KEY_DESCRIBE_GROUPS, version); + let fixed = describe_fixed_len(version, 1); + let encoded = |group: DescribedGroup| { + header + + encode_response(version, vec![group]) + .expect("described group encodes") + .len() + }; + assert_eq!( + fixed + dead_described_len(version, &group_id), + encoded(missing_group(version, &group_id)), + "version {version} dead" + ); + assert_eq!( + fixed + offsets_only_described_len(version, &group_id), + encoded(offsets_only_group(version, &group_id)), + "version {version} offsets only" + ); + } + } + fn join_wide(groups: &mut Groups, config: &GroupCoordinatorConfig, now: Instant) -> JoinResult { let wide = "x".repeat(130); let mut request = request("", &["range"]); diff --git a/gateways/kafka/src/protocol/handlers/describe_groups.rs b/gateways/kafka/src/protocol/handlers/describe_groups.rs index e8f59fd57..c4aa0ba63 100644 --- a/gateways/kafka/src/protocol/handlers/describe_groups.rs +++ b/gateways/kafka/src/protocol/handlers/describe_groups.rs @@ -18,28 +18,35 @@ //! `DescribeGroups` (API key 15). //! //! Reports the members, assignment and state of groups this coordinator holds. A named group -//! that is not here is `Dead`: v6 also carries `GROUP_ID_NOT_FOUND`, and earlier versions do -//! not, matching Kafka 4.0's `describeGroups`. A v6 client treats error 0 as a real group. +//! that is not here is `Empty` when it holds offsets in Iggy, as Kafka keeps a group whose +//! consumers left. Otherwise it is `Dead`: v6 also carries `GROUP_ID_NOT_FOUND`, and earlier +//! versions do not, matching Kafka 4.0's `describeGroups`. A v6 client treats error 0 as a real +//! group. When no scan has found the groups that hold offsets yet (`IggyBridge::offset_groups`), +//! a group that is not here gets `COORDINATOR_LOAD_IN_PROGRESS` (14), which clients retry. //! //! `include_authorized_operations` is not answered. The field stays at the omitted sentinel //! (`i32::MIN`): there is no group ACL bitfield to report, and any other value fails the //! encoder below v3. +use std::sync::Arc; + use bytes::Bytes; use kafka_protocol::messages::describe_groups_response::{DescribedGroup, DescribedGroupMember}; use kafka_protocol::messages::{DescribeGroupsRequest, DescribeGroupsResponse, GroupId}; use kafka_protocol::protocol::StrBytes; +use crate::bridge::OffsetGroups; use crate::error::Result; use crate::group::{DESCRIBE_RESPONSE_TOO_LARGE, GroupDescription, MemberDescription}; use crate::protocol::api::{ - API_KEY_DESCRIBE_GROUPS, ApiVersionRange, ERROR_GROUP_ID_NOT_FOUND, ERROR_NONE, - ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome, is_supported_version, + API_KEY_DESCRIBE_GROUPS, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS, + ERROR_GROUP_ID_NOT_FOUND, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome, + is_supported_version, }; use crate::protocol::bounds_guard::validate_describe_groups_shape; use crate::protocol::dedup::dedup_first_seen; use crate::protocol::handlers::{ - decode_guarded, encode_message, respond_or_close, unsupported_version_response, + decode_guarded, encode_message, offset_deadline, respond_or_close, unsupported_version_response, }; pub const RANGE: ApiVersionRange = ApiVersionRange { @@ -54,7 +61,12 @@ const DEAD: &str = "Dead"; /// `GROUP_ID_NOT_FOUND` for a missing group. const FIRST_ERROR_MESSAGE_VERSION: i16 = 6; -pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { +pub async fn handle( + state: &GatewayState, + connection: &ConnectionState, + api_version: i16, + body: Bytes, +) -> HandleOutcome { if !is_supported_version(API_KEY_DESCRIBE_GROUPS, api_version) { // No top-level error field to carry a code on, so this answers with an empty group // list rather than a dedicated error response. Versions outside 0-6 fail in the @@ -90,12 +102,17 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand .groups .describe_groups(&group_ids, api_version, state.max_frame_size) .await; + let absent = if described.iter().any(Option::is_none) { + Absent::look_up(state, connection).await + } else { + Absent::Dead + }; let groups = group_ids .iter() .zip(described) .map(|(group_id, group)| { group.map_or_else( - || missing_group(api_version, group_id), + || absent.answer(api_version, group_id), |group| described_group(api_version, group), ) }) @@ -103,6 +120,39 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand respond_or_close(encode_response(api_version, groups), "DescribeGroups") } +/// How one request answers a group that is not on this coordinator. +enum Absent { + /// Without a bridge no group holds offsets, so each one is `Dead`. + Dead, + /// `Empty` for a group that holds offsets, `Dead` for any other. + Known(Arc<OffsetGroups>), + /// No scan has found the groups that hold offsets yet. + Unknown, +} + +impl Absent { + async fn look_up(state: &GatewayState, connection: &ConnectionState) -> Self { + let Some(bridge) = &state.bridge else { + return Self::Dead; + }; + let deadline = offset_deadline(state, connection).await; + bridge + .offset_groups(deadline) + .await + .map_or(Self::Unknown, Self::Known) + } + + fn answer(&self, version: i16, group_id: &StrBytes) -> DescribedGroup { + match self { + Self::Known(offset_groups) if offset_groups.contains(group_id.as_str()) => { + described_group(version, GroupDescription::offsets_only(group_id.clone())) + } + Self::Dead | Self::Known(_) => missing_group(version, group_id), + Self::Unknown => loading_group(group_id), + } + } +} + pub(crate) fn described_group(version: i16, group: GroupDescription) -> DescribedGroup { if group.error != ERROR_NONE { return too_large_group(version, &group.group_id); @@ -131,6 +181,16 @@ fn too_large_group(version: i16, group_id: &StrBytes) -> DescribedGroup { group } +/// The same empty shape as [`too_large_group`], with a code the Java admin retries. +fn loading_group(group_id: &StrBytes) -> DescribedGroup { + DescribedGroup::default() + .with_error_code(ERROR_COORDINATOR_LOAD_IN_PROGRESS) + .with_group_id(GroupId(group_id.clone())) + .with_group_state(StrBytes::from_static_str("")) + .with_protocol_type(StrBytes::from_static_str("")) + .with_protocol_data(StrBytes::from_static_str("")) +} + fn described_member(member: MemberDescription) -> DescribedGroupMember { // `client_id` and `client_host` stay empty: the coordinator does not retain the request // header's client id, and handlers are not given the connection's peer address. diff --git a/gateways/kafka/src/protocol/handlers/list_groups.rs b/gateways/kafka/src/protocol/handlers/list_groups.rs index bf29264cc..81828c279 100644 --- a/gateways/kafka/src/protocol/handlers/list_groups.rs +++ b/gateways/kafka/src/protocol/handlers/list_groups.rs @@ -17,26 +17,36 @@ //! `ListGroups` (API key 16). //! -//! Lists the groups this coordinator holds. v4+ `states_filter` and v5 `types_filter` keep a -//! group only when the filter is empty or one entry matches, case-insensitively, after trim. -//! A filter value this coordinator does not use matches nothing, so the group is left out and -//! the response error stays 0. Kafka 4.0's `listGroups` does the same ("if invalid, no groups -//! are returned"). +//! Lists the groups this coordinator holds, and as `Empty` each group with no members here that +//! holds offsets in Iggy. Kafka keeps a group whose consumers left while it has offsets. v4+ +//! `states_filter` and v5 `types_filter` keep a group only when the filter is empty or one entry +//! matches, case-insensitively, after trim. A filter value this coordinator does not use matches +//! nothing, so the group is left out and the response error stays 0. Kafka 4.0's `listGroups` +//! does the same ("if invalid, no groups are returned"). +//! +//! The groups that hold offsets come from a scan that costs one Iggy call per topic, and one scan +//! answers for 10 s (`IggyBridge::offset_groups`). A filter that leaves out `Empty` or `classic` +//! needs no scan. When no scan has found the groups yet, the answer is +//! `COORDINATOR_LOAD_IN_PROGRESS` (14), which clients retry. + +use std::collections::HashSet; use bytes::Bytes; use kafka_protocol::messages::list_groups_response::ListedGroup; use kafka_protocol::messages::{GroupId, ListGroupsRequest, ListGroupsResponse}; use kafka_protocol::protocol::StrBytes; +use crate::bridge::OffsetGroups; use crate::error::Result; -use crate::group::GroupListing; +use crate::group::{GROUP_STATE_EMPTY, GroupListing}; use crate::protocol::api::{ - API_KEY_LIST_GROUPS, ApiVersionRange, ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, - GatewayState, HandleOutcome, is_supported_version, + API_KEY_LIST_GROUPS, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS, + ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome, + is_supported_version, }; use crate::protocol::bounds_guard::validate_list_groups_shape; use crate::protocol::handlers::{ - decode_guarded, encode_message, respond_or_close, unsupported_version_response, + decode_guarded, encode_message, offset_deadline, respond_or_close, unsupported_version_response, }; use crate::protocol::header::response_header_version; use crate::protocol::wire_len::{kafka_compact_prefix, kafka_string_len}; @@ -50,7 +60,12 @@ pub const RANGE: ApiVersionRange = ApiVersionRange { /// `ListGroups` v5 `group_type` for the classic protocol this coordinator runs. const CLASSIC_GROUP_TYPE: &str = "classic"; -pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome { +pub async fn handle( + state: &GatewayState, + connection: &ConnectionState, + api_version: i16, + body: Bytes, +) -> HandleOutcome { if !is_supported_version(API_KEY_LIST_GROUPS, api_version) { return unsupported_version_response(API_KEY_LIST_GROUPS, api_version, |version| { encode_error_response(version, ERROR_UNSUPPORTED_VERSION) @@ -70,10 +85,21 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand } }; - let listings: Vec<GroupListing> = state - .groups - .list_groups() - .await + let mut listings = state.groups.list_groups().await; + let lists_offsets_only = selected(&request.states_filter, GROUP_STATE_EMPTY) + && selected(&request.types_filter, CLASSIC_GROUP_TYPE); + if lists_offsets_only && let Some(bridge) = &state.bridge { + let deadline = offset_deadline(state, connection).await; + let Some(offset_groups) = bridge.offset_groups(deadline).await else { + tracing::debug!("ListGroups found no scan of the groups that hold offsets yet"); + return respond_or_close( + encode_error_response(api_version, ERROR_COORDINATOR_LOAD_IN_PROGRESS), + "ListGroups", + ); + }; + add_offsets_only(&mut listings, &offset_groups); + } + let listings: Vec<GroupListing> = listings .into_iter() .filter(|group| { selected(&request.states_filter, group.state) @@ -96,6 +122,22 @@ fn selected(filter: &[StrBytes], value: &str) -> bool { .any(|wanted| wanted.as_str().trim().eq_ignore_ascii_case(value)) } +/// Adds each group of `offset_groups` that is not live here, as `Empty`. Keeps the list in group +/// id order. +fn add_offsets_only(listings: &mut Vec<GroupListing>, offset_groups: &OffsetGroups) { + let live: HashSet<&str> = listings + .iter() + .map(|group| group.group_id.as_str()) + .collect(); + let offsets_only: Vec<GroupListing> = offset_groups + .iter() + .filter(|group_id| !live.contains(group_id)) + .map(|group_id| GroupListing::offsets_only(StrBytes::from_string(group_id.to_owned()))) + .collect(); + listings.extend(offsets_only); + listings.sort_by(|left, right| left.group_id.cmp(&right.group_id)); +} + /// Keeps as many listed groups as fit `max_frame_size`, dropping the rest rather than closing /// the connection - same approach as `metadata::truncate_all_topics_to_frame_budget` for the /// identical problem: this coordinator's own state, not a request the client chose, is what diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs index 941fe8e74..5d2dd1598 100644 --- a/gateways/kafka/src/protocol/handlers/mod.rs +++ b/gateways/kafka/src/protocol/handlers/mod.rs @@ -127,8 +127,10 @@ pub async fn dispatch( API_KEY_HEARTBEAT => heartbeat::handle(state, connection, api_version, body).await, API_KEY_LEAVE_GROUP => leave_group::handle(state, api_version, body).await, API_KEY_SYNC_GROUP => sync_group::handle(state, api_version, body).await, - API_KEY_DESCRIBE_GROUPS => describe_groups::handle(state, api_version, body).await, - API_KEY_LIST_GROUPS => list_groups::handle(state, api_version, body).await, + API_KEY_DESCRIBE_GROUPS => { + describe_groups::handle(state, connection, api_version, body).await + } + API_KEY_LIST_GROUPS => list_groups::handle(state, connection, api_version, body).await, API_KEY_INIT_PRODUCER_ID => init_producer_id::handle(state, api_version, body).await, _ => HandleOutcome::Close, } diff --git a/gateways/kafka/tests/offset_groups_real_bridge_tests.rs b/gateways/kafka/tests/offset_groups_real_bridge_tests.rs new file mode 100644 index 000000000..57b3a8bc6 --- /dev/null +++ b/gateways/kafka/tests/offset_groups_real_bridge_tests.rs @@ -0,0 +1,426 @@ +// 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. + +//! `ListGroups` and `DescribeGroups` against a real `iggy-server` process, for groups with no +//! members on the coordinator whose offsets are in Iggy. Kafka keeps such a group and reports it +//! as `Empty`. Responses are decoded by hand for the reason that +//! `create_topics_real_bridge_tests.rs` gives. + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/iggy_server.rs"] +mod iggy_server; +#[path = "common/wire.rs"] +mod wire; + +use std::sync::Arc; +use std::time::Duration; + +use iggy::prelude::{ConsumerGroupClient, Identifier}; +use serial_test::serial; +use tokio_util::sync::CancellationToken; + +use iggy_gateway_kafka::bridge::IggyBridge; +use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig}; +use iggy_gateway_kafka::protocol::api::{ + API_KEY_DESCRIBE_GROUPS, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP, API_KEY_LIST_GROUPS, + API_KEY_OFFSET_COMMIT, BrokerAdvertise, ERROR_COORDINATOR_LOAD_IN_PROGRESS, + ERROR_GROUP_ID_NOT_FOUND, ERROR_NONE, GatewayState, handle_request_bounded, +}; + +use codec::Decoder; +use iggy_server::TestServer; +use wire::{ + JoinGroupParams, build_describe_groups_request, build_join_group_request, + build_leave_group_request, build_list_groups_request, build_offset_commit_request, +}; + +const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024; +const COMMIT_VERSION: i16 = 8; +/// The first version with both filters. +const LIST_VERSION: i16 = 5; +/// The last version without `error_message`, so a missing group is `Dead` with error 0. +const DESCRIBE_VERSION: i16 = 5; +/// v3 admits a member without the `MEMBER_ID_REQUIRED` round trip. +const JOIN_VERSION: i16 = 3; +const LEAVE_VERSION: i16 = 1; +const GROUP: &str = "orders-app"; + +/// One `ListGroups` entry: `(group_id, protocol_type, state, group_type)`. +type Listed = (String, String, String, String); + +/// One `DescribeGroups` entry with the fields an offsets-only group sets. +#[derive(Debug, PartialEq, Eq)] +struct Described { + error: i16, + error_message: Option<String>, + group_id: String, + state: String, + protocol_type: String, + protocol_data: String, + members: usize, +} + +async fn connected_state(server: &TestServer) -> GatewayState { + let bridge = IggyBridge::connect(server.test_config()) + .await + .expect("bridge should connect to a ready server"); + // A new group's first join completes at once, so a test can hold a live member. + let groups = GroupCoordinatorConfig { + initial_rebalance_delay: Duration::ZERO, + ..GroupCoordinatorConfig::default() + }; + GatewayState::new( + BrokerAdvertise::default(), + Some(Arc::new(bridge)), + TEST_MAX_FRAME_SIZE, + false, + 0, + GroupCoordinator::new(groups, CancellationToken::new()), + ) +} + +async fn seed_topic(server: &TestServer, topic: &str) { + IggyBridge::connect(server.test_config()) + .await + .expect("seed bridge should connect") + .ensure_stream_and_topic(topic, 1) + .await + .expect("seed the topic"); +} + +/// Commits `offset` for partition 0 of `topic` outside the group protocol, as admin tools do. +async fn commit(state: &GatewayState, group: &str, topic: &str, offset: i64) { + let body = build_offset_commit_request(COMMIT_VERSION, group, topic, offset); + let response = handle_request_bounded(state, API_KEY_OFFSET_COMMIT, COMMIT_VERSION, body) + .await + .expect_response("OffsetCommit answers"); + let mut d = Decoder::new(response); + let _throttle_time_ms = d.read_i32().expect("throttle_time_ms"); + assert_eq!(d.read_varint().expect("topics"), 2, "one topic"); + let _name = d.read_compact_nullable_string().expect("topic name"); + assert_eq!(d.read_varint().expect("partitions"), 2, "one partition"); + let _index = d.read_i32().expect("partition_index"); + assert_eq!(d.read_i16().expect("error_code"), ERROR_NONE, "the commit"); +} + +/// The top-level error and every group of a v5 `ListGroups` answer. +async fn list(state: &GatewayState, states: &[&str], group_types: &[&str]) -> (i16, Vec<Listed>) { + let body = build_list_groups_request(LIST_VERSION, states, group_types); + let response = handle_request_bounded(state, API_KEY_LIST_GROUPS, LIST_VERSION, body) + .await + .expect_response("ListGroups answers"); + let mut d = Decoder::new(response); + let _throttle_time_ms = d.read_i32().expect("throttle_time_ms"); + let error = d.read_i16().expect("error_code"); + let mut groups = Vec::new(); + for _ in 1..d.read_varint().expect("groups") { + let mut string = || { + d.read_compact_nullable_string() + .expect("a string") + .expect("not null") + }; + groups.push((string(), string(), string(), string())); + d.read_tagged_fields().expect("group tagged fields"); + } + d.read_tagged_fields().expect("tagged fields"); + assert_eq!(d.remaining(), 0, "ListGroups trailing bytes"); + (error, groups) +} + +async fn describe(state: &GatewayState, version: i16, groups: &[&str]) -> Vec<Described> { + let body = build_describe_groups_request(version, groups, false); + let response = handle_request_bounded(state, API_KEY_DESCRIBE_GROUPS, version, body) + .await + .expect_response("DescribeGroups answers"); + let mut d = Decoder::new(response); + let _throttle_time_ms = d.read_i32().expect("throttle_time_ms"); + let mut described = Vec::new(); + for _ in 1..d.read_varint().expect("groups") { + let error = d.read_i16().expect("error_code"); + let error_message = if version >= 6 { + d.read_compact_nullable_string().expect("error_message") + } else { + None + }; + let mut string = || { + d.read_compact_nullable_string() + .expect("a string") + .expect("not null") + }; + let (group_id, state, protocol_type, protocol_data) = + (string(), string(), string(), string()); + let members = usize::try_from(d.read_varint().expect("members") - 1).expect("a count"); + assert_eq!(members, 0, "only groups with no members are described here"); + let _authorized_operations = d.read_i32().expect("authorized_operations"); + d.read_tagged_fields().expect("group tagged fields"); + described.push(Described { + error, + error_message, + group_id, + state, + protocol_type, + protocol_data, + members, + }); + } + d.read_tagged_fields().expect("tagged fields"); + assert_eq!(d.remaining(), 0, "DescribeGroups trailing bytes"); + described +} + +/// Joins a member to `group` and returns its id. The group is then `CompletingRebalance`. +async fn join(state: &GatewayState, group: &str) -> String { + let params = JoinGroupParams { + group_id: group, + ..JoinGroupParams::default() + }; + let body = build_join_group_request(JOIN_VERSION, ¶ms); + let response = handle_request_bounded(state, API_KEY_JOIN_GROUP, JOIN_VERSION, body) + .await + .expect_response("JoinGroup answers"); + let mut d = Decoder::new(response); + let _throttle_time_ms = d.read_i32().expect("throttle_time_ms"); + assert_eq!(d.read_i16().expect("error_code"), ERROR_NONE, "the join"); + let _generation_id = d.read_i32().expect("generation_id"); + let _protocol_name = d.read_nullable_string().expect("protocol_name"); + let _leader = d.read_nullable_string().expect("leader"); + d.read_nullable_string() + .expect("member_id") + .expect("a member id") +} + +async fn leave(state: &GatewayState, group: &str, member_id: &str) { + let body = build_leave_group_request(LEAVE_VERSION, group, &[(member_id, None, None)]); + let response = handle_request_bounded(state, API_KEY_LEAVE_GROUP, LEAVE_VERSION, body) + .await + .expect_response("LeaveGroup answers"); + let mut d = Decoder::new(response); + let _throttle_time_ms = d.read_i32().expect("throttle_time_ms"); + assert_eq!(d.read_i16().expect("error_code"), ERROR_NONE, "the leave"); +} + +fn listed(group: &str, state: &str) -> Listed { + let group_type = "classic".to_string(); + ( + group.to_string(), + "consumer".to_string(), + state.to_string(), + group_type, + ) +} + +fn offsets_only(group: &str) -> Described { + Described { + error: ERROR_NONE, + error_message: None, + group_id: group.to_string(), + state: "Empty".to_string(), + protocol_type: "consumer".to_string(), + protocol_data: String::new(), + members: 0, + } +} + +/// A group whose consumers committed and left has no members here. Kafka still lists and +/// describes it, as `Empty`, while it holds offsets. +#[tokio::test] +#[serial] +async fn given_offsets_and_no_members_list_and_describe_should_report_an_empty_group() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + seed_topic(&server, "orders").await; + let state = connected_state(&server).await; + commit(&state, GROUP, "orders", 7).await; + + assert_eq!( + list(&state, &[], &[]).await, + (ERROR_NONE, vec![listed(GROUP, "Empty")]) + ); + assert_eq!( + describe(&state, DESCRIBE_VERSION, &[GROUP, "never-committed"]).await, + vec![ + offsets_only(GROUP), + Described { + state: "Dead".to_string(), + protocol_type: String::new(), + ..offsets_only("never-committed") + }, + ] + ); + assert_eq!( + describe(&state, 6, &[GROUP, "never-committed"]).await, + vec![ + offsets_only(GROUP), + Described { + error: ERROR_GROUP_ID_NOT_FOUND, + error_message: Some("Group never-committed not found.".to_string()), + state: "Dead".to_string(), + protocol_type: String::new(), + ..offsets_only("never-committed") + }, + ] + ); +} + +/// A live group is listed once, with its live state. After its last member leaves, the group is +/// `Empty`, because its offsets are still in Iggy. +#[tokio::test] +#[serial] +async fn given_a_live_group_with_offsets_should_list_it_once_and_empty_after_it_leaves() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + seed_topic(&server, "orders").await; + let state = connected_state(&server).await; + commit(&state, GROUP, "orders", 7).await; + + let member_id = join(&state, GROUP).await; + assert_eq!( + list(&state, &[], &[]).await, + (ERROR_NONE, vec![listed(GROUP, "CompletingRebalance")]) + ); + + leave(&state, GROUP, &member_id).await; + assert_eq!( + list(&state, &[], &[]).await, + (ERROR_NONE, vec![listed(GROUP, "Empty")]) + ); + assert_eq!( + describe(&state, DESCRIBE_VERSION, &[GROUP]).await, + vec![offsets_only(GROUP)] + ); +} + +/// The filters apply to a group that only holds offsets as to any other: its state is `Empty` and +/// its type is `classic`. +#[tokio::test] +#[serial] +async fn given_filters_should_keep_an_offsets_only_group_only_when_they_match() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + seed_topic(&server, "orders").await; + let state = connected_state(&server).await; + commit(&state, GROUP, "orders", 7).await; + let empty = (ERROR_NONE, vec![listed(GROUP, "Empty")]); + + assert_eq!(list(&state, &[" empty "], &[]).await, empty); + assert_eq!(list(&state, &[], &["Classic"]).await, empty); + assert_eq!(list(&state, &["Stable"], &[]).await, (ERROR_NONE, vec![])); + assert_eq!(list(&state, &[], &["consumer"]).await, (ERROR_NONE, vec![])); +} + +/// Groups come from a scan that answers for a while. A commit through this gateway adds its group +/// at once, so a group does not wait for the next scan. +#[tokio::test] +#[serial] +async fn given_a_commit_after_the_last_scan_should_list_its_group_at_once() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + seed_topic(&server, "orders").await; + let state = connected_state(&server).await; + commit(&state, "first-app", "orders", 1).await; + assert_eq!( + list(&state, &[], &[]).await, + (ERROR_NONE, vec![listed("first-app", "Empty")]) + ); + + commit(&state, "second-app", "orders", 2).await; + + assert_eq!( + list(&state, &[], &[]).await, + ( + ERROR_NONE, + vec![listed("first-app", "Empty"), listed("second-app", "Empty")] + ) + ); +} + +/// A gateway that just started has no memory of the groups. Its scan finds the groups that another +/// gateway, or an earlier run, committed for, on every topic. +#[tokio::test] +#[serial] +async fn given_a_new_gateway_should_find_the_groups_of_every_topic() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + for topic in ["orders", "payments"] { + seed_topic(&server, topic).await; + } + let first = connected_state(&server).await; + commit(&first, "first-app", "orders", 1).await; + commit(&first, "second-app", "payments", 2).await; + commit(&first, "first-app", "payments", 3).await; + drop(first); + + let restarted = connected_state(&server).await; + + assert_eq!( + list(&restarted, &[], &[]).await, + ( + ERROR_NONE, + vec![listed("first-app", "Empty"), listed("second-app", "Empty")] + ) + ); +} + +/// Only an Iggy group named `kafka.cg.<group>` holds Kafka offsets. A native Iggy group on the +/// same topic is not a Kafka group. +#[tokio::test] +#[serial] +async fn given_a_native_iggy_group_should_not_list_it() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + seed_topic(&server, "orders").await; + let raw = iggy_server::raw_client(&server).await; + let iggy_id = |name: &str| Identifier::named(name).expect("valid Iggy name"); + raw.create_consumer_group(&iggy_id("kafka"), &iggy_id("orders"), "native-app") + .await + .expect("create a native Iggy group"); + let state = connected_state(&server).await; + + assert_eq!(list(&state, &[], &[]).await, (ERROR_NONE, vec![])); +} + +/// Without a scan that found the groups, the gateway cannot tell `Empty` from `Dead`. It answers +/// `COORDINATOR_LOAD_IN_PROGRESS`, which clients retry, rather than a list that leaves groups out. +#[tokio::test] +#[serial] +async fn given_iggy_unreachable_before_any_scan_should_answer_coordinator_load_in_progress() { + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let state = connected_state(&server).await; + drop(server); + + assert_eq!( + list(&state, &[], &[]).await, + (ERROR_COORDINATOR_LOAD_IN_PROGRESS, vec![]) + ); + assert_eq!( + list(&state, &["Stable"], &[]).await, + (ERROR_NONE, vec![]), + "a filter that leaves out Empty needs no scan" + ); + let unknown = describe(&state, DESCRIBE_VERSION, &[GROUP]).await; + assert_eq!( + unknown, + vec![Described { + error: ERROR_COORDINATOR_LOAD_IN_PROGRESS, + state: String::new(), + protocol_type: String::new(), + ..offsets_only(GROUP) + }] + ); +}
