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/21 07:10:50 UTC
(iggy) 02/02: address review
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 82aed0ed70c50b4452d6fd06c01261c6fbb74aa6
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Fri Oct 16 14:36:41 2026 +0530
address review
---
gateways/kafka/docs/CONSUMER_GROUPS.md | 26 +-
gateways/kafka/docs/OFFSET_STORAGE.md | 3 +
.../kafka/src/bridge/iggy_bridge/group_index.rs | 22 +-
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 178 +++++++++-----
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 11 +-
gateways/kafka/src/bridge/iggy_bridge/topics.rs | 24 +-
gateways/kafka/src/group/mod.rs | 197 ++++++++++++++-
gateways/kafka/src/group/offset_work.rs | 266 +++++++++++++++++++++
.../kafka/src/protocol/handlers/delete_groups.rs | 66 ++++-
.../kafka/src/protocol/handlers/describe_groups.rs | 42 ++--
.../kafka/src/protocol/handlers/offset_commit.rs | 12 +-
.../kafka/src/protocol/handlers/offset_delete.rs | 64 ++---
.../kafka/tests/offset_groups_real_bridge_tests.rs | 111 ++++++++-
gateways/kafka/tools/kafka-tool/src/response.rs | 49 +++-
14 files changed, 908 insertions(+), 163 deletions(-)
diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md
index 55eef29a3..3cc5f7507 100644
--- a/gateways/kafka/docs/CONSUMER_GROUPS.md
+++ b/gateways/kafka/docs/CONSUMER_GROUPS.md
@@ -71,10 +71,11 @@ consumers left, and the Java admin client takes `consumer` or an empty type as a
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
+([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). It runs on an Iggy client of its own and makes one
+call at a time. So it never waits behind Produce, and no offset call waits behind it. 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.
@@ -112,6 +113,17 @@ Iggy deletes a group on each partition of the topic, and lists it until every pa
gateway looks up the group on the topic every 5 to 100 milliseconds. When the group is gone, the
gateway answers. Then `OffsetFetch` returns -1 for the group, and the admin APIs no longer find it.
+While a deletion runs, the group takes no join and no commit. `JoinGroup` and `OffsetCommit` answer
+`COORDINATOR_LOAD_IN_PROGRESS` (14), and clients retry. The deletion waits for each commit that
+passed its check before the deletion began. So a commit lands whole before the deletion or whole
+after it. Also, no member subscribes to a topic between the check of `OffsetDelete` and its
+deletions. If one of those commits still runs at the offset deadline, the deletion answers 14.
+
+A gateway does not see the members on another gateway. Through a gateway where a group has no
+members, `DeleteGroups` deletes the group. `OffsetDelete` there also deletes offsets of the topics
+that those members subscribe to. Point admin tools at the gateway that the group's consumers use
+([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md#more-than-one-gateway)).
+
`OffsetDelete` reads the topics that the members subscribe to from their metadata for the selected
protocol, as Kafka does. It reads only the topic list of the consumer protocol's v0 prefix, which
every later version keeps. If a member's metadata does not decode, or the group has not chosen a
@@ -123,7 +135,11 @@ A call to Iggy that a retry can fix answers `COORDINATOR_LOAD_IN_PROGRESS` (14),
client retries it. `DeleteGroups` answers it for the group, and `OffsetDelete` for the whole
request. A second delete of an offset or a group does no harm. A deletion that does not end by the
offset deadline also answers 14. When Iggy refuses the bridge user, `DeleteGroups` answers
-`GROUP_AUTHORIZATION_FAILED` (30).
+`GROUP_AUTHORIZATION_FAILED` (30). When the bridge user has no permission to list the Iggy groups
+of a Kafka topic, `DeleteGroups` deletes nothing. It answers 30 for each group to delete. The same
+applies to an override topic that the bridge user has no permission to read. `Metadata` leaves such
+a topic out. A group can hold offsets there, so the gateway cannot delete all of them. The log names
+the topics.
A SASL listener names principals, and so does an `SSL` listener that verifies client certificates.
When a listener names principals, both APIs require the principal to hold the Iggy permission
diff --git a/gateways/kafka/docs/OFFSET_STORAGE.md b/gateways/kafka/docs/OFFSET_STORAGE.md
index 58b55d403..ab9f3365d 100644
--- a/gateways/kafka/docs/OFFSET_STORAGE.md
+++ b/gateways/kafka/docs/OFFSET_STORAGE.md
@@ -114,6 +114,9 @@ Offsets are shared, since the key depends only on the Kafka group id. Membership
- A commit that a gateway gave up on can still land after a newer commit made through another
gateway. It then puts back the older offset, or deletes the key. Iggy cannot refuse the older
write yet.
+- `DeleteGroups` and `OffsetDelete` see only the members on their own gateway. Through a gateway
+ with no members in a group, they delete offsets of a group that is live on another gateway. That
+ includes offsets of the topics that its members subscribe to.
## Server side
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_index.rs b/gateways/kafka/src/bridge/iggy_bridge/group_index.rs
index c7898b5dc..0783a3dfd 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/group_index.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_index.rs
@@ -18,7 +18,7 @@
//! The Kafka groups that hold offsets in Iggy, for the group admin APIs. Finding them costs one
//! Iggy call per topic, so one scan answers every request for a while.
-use std::collections::BTreeSet;
+use std::collections::{BTreeSet, HashMap};
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
@@ -70,10 +70,10 @@ struct Published {
ended: u64,
/// The groups of the last scan that found them.
groups: Option<Arc<OffsetGroups>>,
- /// The groups this gateway stored offsets for (`true`) or deleted (`false`) while a scan
- /// runs, in order. `None` while none runs. The scan can list a topic before such a change,
- /// so the changes go on top of what it finds.
- changes: Option<Vec<(String, bool)>>,
+ /// The last change of each group while a scan runs: this gateway stored offsets for it
+ /// (`true`) or deleted them (`false`). `None` while none runs. The scan can list a topic
+ /// before such a change, so the changes go on top of what it finds.
+ changes: Option<HashMap<String, bool>>,
}
impl GroupIndex {
@@ -127,7 +127,11 @@ impl GroupIndex {
fn change(&self, group: &str, holds: bool) {
self.published.send_if_modified(|current| {
if let Some(changes) = current.changes.as_mut() {
- changes.push((group.to_owned(), holds));
+ if let Some(held) = changes.get_mut(group) {
+ *held = holds;
+ } else {
+ changes.insert(group.to_owned(), holds);
+ }
}
let Some(groups) = current.groups.as_mut() else {
return false;
@@ -152,7 +156,7 @@ impl GroupIndex {
};
let published = Arc::clone(&self.published);
published.send_if_modified(|current| {
- current.changes = Some(Vec::new());
+ current.changes = Some(HashMap::new());
false
});
let started = Instant::now();
@@ -362,13 +366,15 @@ mod tests {
let (release, released) = oneshot::channel::<()>();
let stale = move || async move {
let _ = released.await;
- Ok(found(&["deleted", "kept"]))
+ Ok(found(&["deleted", "kept", "flipped"]))
};
let (groups, ()) = tokio::join!(index.groups(later(), stale), async {
tokio::task::yield_now().await;
index.forget("deleted");
index.note("stored");
+ index.note("flipped");
+ index.forget("flipped");
release.send(()).expect("the scan waits for its release");
});
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
index f94a11006..326c34ae1 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
@@ -29,7 +29,7 @@ use iggy::prelude::{
};
use tokio::sync::{OwnedSemaphorePermit, Semaphore};
use tokio::time::{Instant, timeout, timeout_at};
-use tracing::{debug, info};
+use tracing::{debug, error, info};
use super::{
IggyBridge, KafkaTopicMetadata, LazyClient, OffsetGroups, SLOT_LIMIT, TopicTarget, in_slot,
@@ -44,7 +44,7 @@ use crate::protocol::api::ERROR_UNKNOWN_TOPIC_OR_PARTITION;
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.
+/// that started it stops waiting, on a client of its own, so no offset call waits behind it.
const GROUP_SCAN_LIMIT: Duration = Duration::from_secs(30);
/// The first pause, and the longest, between two checks of whether Iggy finished deleting a group.
@@ -213,33 +213,43 @@ impl IggyBridge {
///
/// Lists the topics, then the Iggy groups of each topic, so it costs one call per topic, plus
/// at least two per deletion. Iggy deletes a group on each partition of the topic, and lists
- /// it until every partition did, so a deletion waits for that. A topic that the bridge user
- /// may not read, or that was deleted since the listing, holds none. A group whose deletions
- /// all ended leaves [`Self::offset_groups`] at once.
+ /// it until every partition did, so a deletion waits for that. A topic deleted since the
+ /// listing holds none. A group whose deletions all ended leaves [`Self::offset_groups`] at
+ /// once.
///
/// # Errors
///
- /// The outer error holds for every group: the listing failed, and nothing was deleted. Per
- /// group, as [`Self::commit_group_offsets`], and [`BridgeError::Timeout`] also when Iggy has
- /// not finished a deletion by the deadline of `calls`.
+ /// The outer error holds for every group, and nothing was deleted: the listing failed, or
+ /// [`IggyError::Unauthorized`] when the bridge user may not list the groups of a Kafka topic,
+ /// or may not read an override topic, which the listing then leaves out. A group can hold
+ /// offsets there. Per group, as [`Self::commit_group_offsets`], and [`BridgeError::Timeout`]
+ /// also when Iggy has not finished a deletion by the deadline of `calls`.
pub async fn delete_offset_groups(
&self,
groups: &[&str],
calls: &OffsetCalls,
) -> Result<Vec<Result<bool, BridgeError>>, BridgeError> {
- let topics = self.list_kafka_topics_in_slot(calls).await?;
+ let listing =
+ |client: Arc<IggyClient>| async move { self.kafka_topic_listing_on(&client).await };
+ let listing = self.read_in_slot(LISTING_SLOT, calls, listing).await?;
let listings = join_all(
- topics
+ listing
+ .topics
.iter()
.map(|topic| self.offset_groups_on(&topic.kafka_topic, calls)),
)
.await;
+ let mut unexamined = listing.unreadable;
// The index in `groups` and the Kafka topic of each Iggy group to delete.
let mut deletions = Vec::new();
- for (topic, listing) in topics.iter().zip(listings) {
- let found = match listing {
+ for (topic, listed) in listing.topics.iter().zip(listings) {
+ let found = match listed {
Ok(found) => found,
- Err(error) if holds_no_offset_groups(&error) => continue,
+ Err(error) if topic_deleted(&error) => continue,
+ Err(BridgeError::Iggy(IggyError::Unauthorized)) => {
+ unexamined.push(topic.kafka_topic.clone());
+ continue;
+ }
Err(error) => return Err(error),
};
for (at, group) in groups.iter().enumerate() {
@@ -248,6 +258,14 @@ impl IggyBridge {
}
}
}
+ if !unexamined.is_empty() {
+ error!(
+ kafka_topics = ?unexamined,
+ "the bridge user may not list the Iggy groups of these Kafka topics, so DeleteGroups \
+ deletes nothing"
+ );
+ return Err(BridgeError::Iggy(IggyError::Unauthorized));
+ }
let deleted =
join_all(deletions.iter().map(|&(at, kafka_topic)| {
self.delete_offset_group_on(groups[at], kafka_topic, calls)
@@ -270,8 +288,8 @@ impl IggyBridge {
Ok(results)
}
- /// The partition count of `kafka_topic`, read on an offset slot. `None` when the topic does
- /// not exist.
+ /// The Iggy topic of `kafka_topic` and its partition count, read on an offset slot. `None`
+ /// when the topic does not exist.
///
/// # Errors
///
@@ -280,14 +298,15 @@ impl IggyBridge {
&self,
kafka_topic: &str,
calls: &OffsetCalls,
- ) -> Result<Option<u32>, BridgeError> {
+ ) -> Result<Option<(TopicTarget, u32)>, BridgeError> {
let topic = self.topic_target(kafka_topic)?;
let lookup = |client: Arc<IggyClient>| async move {
let details = self.get_kafka_topic_on(&client, kafka_topic).await?;
Ok(details.map(|details| details.partitions_count))
};
// Any slot will do. Partition 0's keeps the choice fixed.
- self.read_in_slot(slot_of(&topic, 0), calls, lookup).await
+ let count = self.read_in_slot(slot_of(&topic, 0), calls, lookup).await?;
+ Ok(count.map(|count| (topic, count)))
}
/// [`Self::list_kafka_topics`] on an offset slot's read client, so it never waits behind
@@ -307,7 +326,8 @@ impl IggyBridge {
/// 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.
+ /// the Iggy groups of each topic, so it costs one call per topic. It runs on a client of its
+ /// own, one call at a time, so no offset call waits behind it.
///
/// 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
@@ -315,44 +335,41 @@ impl IggyBridge {
/// 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
- };
+ let scan = move || async move { bridge.scan_offset_groups().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(error) if holds_no_offset_groups(&error) => {
- debug!(
- topic = topic.kafka_topic.as_str(),
- %error,
- "skipping a topic in the scan for groups"
- );
+ /// none. A topic whose groups the bridge user may not list is left out, so that one such
+ /// topic does not hide the groups of every other.
+ async fn scan_offset_groups(&self) -> Result<BTreeSet<String>, BridgeError> {
+ let scan = |client: Arc<IggyClient>| async move {
+ let mut groups = BTreeSet::new();
+ for topic in self.list_kafka_topics_on(&client).await? {
+ let target = self.topic_target(&topic.kafka_topic)?;
+ match kafka_groups_on(&client, &target).await {
+ Ok(found) => groups.extend(found),
+ Err(error)
+ if topic_deleted(&error)
+ || matches!(error, BridgeError::Iggy(IggyError::Unauthorized)) =>
+ {
+ debug!(
+ topic = topic.kafka_topic.as_str(),
+ %error,
+ "skipping a topic in the scan for groups"
+ );
+ }
+ Err(error) => return Err(error),
}
- Err(error) => return Err(error),
}
- }
- Ok(groups)
+ Ok(groups)
+ };
+ self.scan_client
+ .until(&self.config, Instant::now() + GROUP_SCAN_LIMIT, scan)
+ .await
}
- /// The Kafka groups with offsets on `kafka_topic`.
+ /// The Kafka groups with offsets on `kafka_topic`, listed on an offset slot.
async fn offset_groups_on(
&self,
kafka_topic: &str,
@@ -361,16 +378,8 @@ impl IggyBridge {
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())
- };
+ let listing =
+ |client: Arc<IggyClient>| async move { kafka_groups_on(&client, &topic).await };
self.read_in_slot(slot, calls, listing).await
}
@@ -410,7 +419,7 @@ impl IggyBridge {
match self.holds_offset_group(group, &topic, calls).await {
Ok(false) => return Ok(()),
Ok(true) => {}
- Err(error) if holds_no_offset_groups(&error) => return Ok(()),
+ Err(error) if topic_deleted(&error) => return Ok(()),
Err(error) => return Err(error),
}
if Instant::now() + pause >= calls.deadline() {
@@ -530,11 +539,10 @@ fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
(first + step) % OFFSET_SLOTS
}
-/// Whether a failed listing of the Iggy groups of a topic means that the topic holds no Kafka
-/// offsets here: the bridge user may not read it, or it was deleted since the topic listing.
-const fn holds_no_offset_groups(error: &BridgeError) -> bool {
- matches!(error, BridgeError::Iggy(IggyError::Unauthorized))
- || error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION
+/// Whether a call on a listed topic failed because the topic was deleted since the listing. Iggy
+/// deleted its groups with it.
+const fn topic_deleted(error: &BridgeError) -> bool {
+ error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION
}
/// One refusal per call, for a group id that names no valid Iggy group.
@@ -563,6 +571,21 @@ impl OffsetNames {
}
}
+/// The Kafka groups with offsets on `topic`, listed on `client`.
+async fn kafka_groups_on(
+ client: &IggyClient,
+ topic: &TopicTarget,
+) -> Result<Vec<String>, BridgeError> {
+ 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())
+}
+
/// 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)
@@ -908,6 +931,33 @@ mod tests {
);
}
+ /// A scan may take 30 s, so it runs on a client of its own. On an offset read lane, one slow
+ /// scan call would make the `OffsetFetch` of that lane time out behind it.
+ #[tokio::test]
+ async fn given_every_offset_read_lane_busy_a_scan_should_still_run() {
+ let bridge = Arc::new(IggyBridge {
+ scan_client: LazyClient::holding(offline_client()),
+ ..offline_bridge(offline_clients(), offline_clients())
+ });
+ let held = OffsetCalls::new(later());
+ let mut lanes = Vec::with_capacity(OFFSET_SLOTS);
+ for slot in 0..OFFSET_SLOTS {
+ let turn = held
+ .take(slot, &bridge.offset_pool.slots[slot].reads)
+ .await
+ .expect("a free lane");
+ lanes.push(turn);
+ }
+
+ let scanned = timeout(Duration::from_secs(1), bridge.offset_groups(later())).await;
+
+ assert!(
+ matches!(scanned, Ok(None)),
+ "the scan ended without the offset lanes, and an offline client finds no groups"
+ );
+ drop(lanes);
+ }
+
#[test]
fn given_a_replay_that_already_landed_the_write_should_count_as_done() {
assert!(landed(Ok(())).is_ok());
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index bd917391a..d4ebe5afd 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -142,8 +142,8 @@ async fn permit_by(turn: &Arc<Semaphore>, deadline: Instant) -> Option<OwnedSema
/// Each producer sends on a client of its own, one request at a time. Producers past the session
/// limit share one more. Metadata, `CreateTopics`, `DeleteTopics`, `DescribeConfigs` and
/// `AlterConfigs` share another. Topic probes, certificate user lookups and `Metadata`'s auto topic
-/// creations each use a client of their own, Fetch polls one per read slot, and offset calls two
-/// per offset slot.
+/// creations each use a client of their own, Fetch polls one per read slot, offset calls two per
+/// offset slot, and the scan for the groups that hold offsets one more.
///
/// Each client signs in at the metadata leader, and a view change or a refused call can move it to
/// another node. So in a cluster, two clients can read replicas that are not at the same offset.
@@ -166,6 +166,8 @@ pub struct IggyBridge {
offset_pool: OffsetPool,
/// The Kafka groups that hold offsets. See `offset_groups`.
group_index: GroupIndex,
+ /// The client of the scan that finds those groups, so no offset call waits behind it.
+ scan_client: LazyClient,
}
/// The Iggy stream and topic one Kafka topic maps to. Resolve once per topic, use many times.
@@ -209,6 +211,7 @@ impl IggyBridge {
auto_create_client: LazyClient::default(),
offset_pool: OffsetPool::new(),
group_index: GroupIndex::new(),
+ scan_client: LazyClient::default(),
})
}
@@ -290,6 +293,7 @@ impl IggyBridge {
lookup_client,
auto_create_client,
offset_clients,
+ scan_client,
shared_client,
) = tokio::join!(
self.produce_pool.close(),
@@ -298,6 +302,7 @@ impl IggyBridge {
self.lookup_client.close(),
self.auto_create_client.close(),
self.offset_pool.close(),
+ self.scan_client.close(),
sign_out_and_shut_down(&self.client),
);
produce_clients
@@ -306,6 +311,7 @@ impl IggyBridge {
.and(lookup_client)
.and(auto_create_client)
.and(offset_clients)
+ .and(scan_client)
.and(shared_client)
}
@@ -393,6 +399,7 @@ impl IggyBridge {
lookup_client: LazyClient::default(),
offset_pool: OffsetPool::new(),
group_index: GroupIndex::new(),
+ scan_client: LazyClient::default(),
}
}
}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/topics.rs b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
index e71d02013..11f7a24e3 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/topics.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/topics.rs
@@ -74,6 +74,14 @@ pub struct KafkaTopicMetadata {
pub partitions_count: u32,
}
+/// [`IggyBridge::list_kafka_topics`], with the override topics that it leaves out because the
+/// bridge user may not read them.
+pub(super) struct KafkaTopicListing {
+ pub(super) topics: Vec<KafkaTopicMetadata>,
+ /// The Kafka names of those override topics.
+ pub(super) unreadable: Vec<String>,
+}
+
impl IggyBridge {
/// Ensures the Iggy stream and topic backing `kafka_topic` exist, creating either or both if
/// missing. Resolves `kafka_topic` through the configured [`TopicMapping`](crate::bridge::topic_map::TopicMapping).
@@ -547,10 +555,20 @@ impl IggyBridge {
&self,
client: &IggyClient,
) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ Ok(self.kafka_topic_listing_on(client).await?.topics)
+ }
+
+ /// [`Self::list_kafka_topics_on`], with the override topics that it leaves out because the
+ /// bridge user may not read them.
+ pub(super) async fn kafka_topic_listing_on(
+ &self,
+ client: &IggyClient,
+ ) -> Result<KafkaTopicListing, BridgeError> {
let default_stream = self.config.topic_mapping.default_stream();
let mut default_stream_override_targets: HashSet<&str> = HashSet::new();
let mut override_keys: HashSet<&str> = HashSet::new();
let mut results = Vec::new();
+ let mut unreadable = Vec::new();
for (kafka_topic, over) in self.config.topic_mapping.overrides() {
override_keys.insert(kafka_topic);
@@ -574,6 +592,7 @@ impl IggyBridge {
unauthorized to read it",
over.stream, over.topic
);
+ unreadable.push(kafka_topic.to_string());
}
Err(err) => return Err(err),
}
@@ -610,7 +629,10 @@ impl IggyBridge {
});
}
- Ok(results)
+ Ok(KafkaTopicListing {
+ topics: results,
+ unreadable,
+ })
}
}
diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs
index 1e27ec003..cae713b91 100644
--- a/gateways/kafka/src/group/mod.rs
+++ b/gateways/kafka/src/group/mod.rs
@@ -26,8 +26,10 @@
//! Membership is process memory, not Iggy state. Two gateway instances fronting one Iggy cluster
//! therefore coordinate two independent groups under one name; see `docs/CONSUMER_GROUPS.md`.
+mod offset_work;
mod state;
+pub use offset_work::{RunningCommit, RunningDeletion};
pub use state::{MAX_GROUP_ID_BYTES, is_valid_group_id};
use std::collections::{HashMap, HashSet};
@@ -42,8 +44,11 @@ use tokio::sync::{Mutex, watch};
use tokio::time::Instant;
use tokio_util::sync::CancellationToken;
+use crate::group::offset_work::OffsetWork;
use crate::group::state::{GroupState, Step};
-use crate::protocol::api::{ERROR_NONE, ERROR_NOT_COORDINATOR, ERROR_UNKNOWN_MEMBER_ID};
+use crate::protocol::api::{
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_NONE, ERROR_NOT_COORDINATOR, ERROR_UNKNOWN_MEMBER_ID,
+};
/// `DescribeGroups` / `ListGroups` state string for a group with no joined members.
pub const GROUP_STATE_EMPTY: &str = "Empty";
@@ -521,6 +526,9 @@ impl From<&HeartbeatRequest> for GroupMember {
pub struct GroupCoordinator {
config: GroupCoordinatorConfig,
groups: Mutex<HashMap<StrBytes, GroupState>>,
+ /// The commits and the offset deletions that run for each group. Changed under `groups`, so
+ /// a check of a group and its deletion's start are one step.
+ offset_work: OffsetWork,
/// Resolves parked waiters on shutdown drain instead of holding it open for a full rebalance
/// timeout.
shutdown: CancellationToken,
@@ -532,11 +540,16 @@ impl GroupCoordinator {
Self {
config,
groups: Mutex::new(HashMap::new()),
+ offset_work: OffsetWork::new(),
shutdown,
}
}
/// Joins `request`'s member, parking until the group's join barrier completes.
+ ///
+ /// While a deletion of the group's offsets runs, answers `COORDINATOR_LOAD_IN_PROGRESS`,
+ /// which clients retry. A member that joined then could subscribe to a topic whose offsets
+ /// the deletion removes, and commit offsets that it wipes.
pub async fn join(&self, request: &JoinRequest) -> JoinResult {
let mut parked: Option<StrBytes> = None;
loop {
@@ -544,6 +557,12 @@ impl GroupCoordinator {
let mut groups = self.groups.lock().await;
let now = Instant::now();
let step = match parked.as_ref() {
+ None if self.offset_work.deleting(&request.group_id) => {
+ Step::Respond(JoinResult::error(
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ request.member_id.clone(),
+ ))
+ }
None => state::join_step(&mut groups, &self.config, request, now),
Some(member_id) => {
state::join_resume_step(&mut groups, &request.group_id, member_id, now)
@@ -648,28 +667,58 @@ impl GroupCoordinator {
)
}
- /// Whether `request` may commit offsets for its group. A member's commit also refreshes its
- /// session. Never parks.
- pub async fn validate_commit(&self, request: &CommitRequest) -> i16 {
+ /// The commit of `request`, counted until it drops, when `request` may commit offsets for its
+ /// group. A member's commit also refreshes its session. Never parks.
+ ///
+ /// # Errors
+ ///
+ /// The Kafka code that refuses the commit. `COORDINATOR_LOAD_IN_PROGRESS` while a deletion of
+ /// the group's offsets runs, which clients retry.
+ pub async fn validate_commit(&self, request: &CommitRequest) -> Result<RunningCommit<'_>, i16> {
let mut groups = self.groups.lock().await;
- state::commit_step(&mut groups, request, Instant::now())
+ match state::commit_step(&mut groups, request, Instant::now()) {
+ ERROR_NONE => self
+ .offset_work
+ .begin_commit(&request.group_id)
+ .ok_or(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ error => Err(error),
+ }
}
/// Removes each group of `group_ids` that has no joined members, as `DeleteGroups` does. One
- /// answer per id, in order. Never parks.
- pub async fn remove_empty(&self, group_ids: &[StrBytes]) -> Vec<GroupRemoval> {
+ /// answer per id, in order. The deletion of the offsets of each group that is not
+ /// [`GroupRemoval::Live`] runs until the returned value drops. Never parks.
+ pub async fn remove_empty(
+ &self,
+ group_ids: &[StrBytes],
+ ) -> (Vec<GroupRemoval>, RunningDeletion<'_>) {
let mut groups = self.groups.lock().await;
let now = Instant::now();
- group_ids
+ let removals: Vec<GroupRemoval> = group_ids
.iter()
.map(|group_id| state::remove_empty(&mut groups, group_id, now))
- .collect()
+ .collect();
+ let deleted = group_ids
+ .iter()
+ .zip(&removals)
+ .filter(|(_, removal)| **removal != GroupRemoval::Live)
+ .map(|(group_id, _)| group_id);
+ let deletion = self.offset_work.begin_deletion(deleted);
+ drop(groups);
+ (removals, deletion)
}
- /// What `OffsetDelete` may do with the offsets of `group_id`. Never parks.
- pub async fn offset_deletion(&self, group_id: &StrBytes) -> OffsetDeletion {
+ /// What `OffsetDelete` may do with the offsets of `group_id`. The deletion runs until the
+ /// returned value drops. Never parks.
+ pub async fn offset_deletion(
+ &self,
+ group_id: &StrBytes,
+ ) -> (OffsetDeletion, RunningDeletion<'_>) {
let mut groups = self.groups.lock().await;
- state::offset_deletion(&mut groups, group_id, Instant::now())
+ let allowed = state::offset_deletion(&mut groups, group_id, Instant::now());
+ let deletion = self.offset_work.begin_deletion([group_id]);
+ drop(groups);
+ (allowed, deletion)
}
/// How long an `OffsetCommit` or `OffsetFetch` may hold its connection. A heartbeat there
@@ -842,6 +891,130 @@ mod tests {
assert_eq!(identities, vec![("m-1", None), ("", Some("i-2"))]);
}
+ /// No initial rebalance delay, so a first join completes at once.
+ fn coordinator() -> GroupCoordinator {
+ let config = GroupCoordinatorConfig {
+ initial_rebalance_delay: Duration::ZERO,
+ ..GroupCoordinatorConfig::default()
+ };
+ GroupCoordinator::new(config, CancellationToken::new())
+ }
+
+ fn id(group: &str) -> StrBytes {
+ StrBytes::from_string(group.to_owned())
+ }
+
+ fn join_request(group: &str) -> JoinRequest {
+ JoinRequest {
+ group_id: id(group),
+ session_timeout: Duration::from_secs(10),
+ rebalance_timeout: Duration::from_secs(10),
+ member_id: StrBytes::new(),
+ group_instance_id: None,
+ protocol_type: StrBytes::from_static_str(CONSUMER_PROTOCOL_TYPE),
+ protocols: vec![(StrBytes::from_static_str("range"), Bytes::new())],
+ require_known_member_id: false,
+ }
+ }
+
+ /// A commit made outside the group protocol, as admin tools send.
+ fn standalone_commit(group: &str) -> CommitRequest {
+ CommitRequest {
+ group_id: id(group),
+ generation_id: -1,
+ member_id: StrBytes::new(),
+ group_instance_id: None,
+ }
+ }
+
+ /// Kafka's coordinator refuses a join or a commit of a group that `DeleteGroups` marked
+ /// `Dead`. A member that joined during the deletion would see its commits wiped.
+ #[tokio::test]
+ async fn given_a_deletion_that_runs_should_refuse_joins_and_commits_of_its_groups() {
+ let coordinator = coordinator();
+ let (removals, deletion) = coordinator.remove_empty(&[id("orders")]).await;
+ assert_eq!(removals, [GroupRemoval::Absent]);
+
+ assert_eq!(
+ coordinator.join(&join_request("orders")).await.error,
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS
+ );
+ assert_eq!(
+ coordinator
+ .validate_commit(&standalone_commit("orders"))
+ .await
+ .err(),
+ Some(ERROR_COORDINATOR_LOAD_IN_PROGRESS)
+ );
+ assert!(
+ coordinator
+ .validate_commit(&standalone_commit("payments"))
+ .await
+ .is_ok(),
+ "another group commits"
+ );
+
+ drop(deletion);
+ assert!(
+ coordinator
+ .validate_commit(&standalone_commit("orders"))
+ .await
+ .is_ok()
+ );
+ assert_eq!(
+ coordinator.join(&join_request("orders")).await.error,
+ ERROR_NONE
+ );
+ }
+
+ /// A group with members is not deleted, so its members go on committing.
+ #[tokio::test]
+ async fn given_a_live_group_when_deleting_should_not_refuse_its_commits() {
+ let coordinator = coordinator();
+ let joined = coordinator.join(&join_request("orders")).await;
+ assert_eq!(joined.error, ERROR_NONE);
+ let synced = coordinator
+ .sync(SyncRequest {
+ group_id: id("orders"),
+ generation_id: joined.generation_id,
+ member_id: joined.member_id.clone(),
+ protocol_type: None,
+ protocol_name: None,
+ assignments: vec![(joined.member_id.clone(), Bytes::new())],
+ })
+ .await;
+ assert_eq!(synced.error, ERROR_NONE, "the group is stable");
+
+ let (removals, _deletion) = coordinator.remove_empty(&[id("orders")]).await;
+ assert_eq!(removals, [GroupRemoval::Live]);
+
+ let commit = CommitRequest {
+ generation_id: joined.generation_id,
+ member_id: joined.member_id,
+ ..standalone_commit("orders")
+ };
+ assert!(coordinator.validate_commit(&commit).await.is_ok());
+ }
+
+ /// A member that joined during `OffsetDelete` could subscribe to a topic whose offsets go.
+ #[tokio::test]
+ async fn given_an_offset_deletion_that_runs_should_refuse_joins_of_the_group() {
+ let coordinator = coordinator();
+ let (deletable, deletion) = coordinator.offset_deletion(&id("orders")).await;
+ assert_eq!(deletable, OffsetDeletion::Absent);
+
+ assert_eq!(
+ coordinator.join(&join_request("orders")).await.error,
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS
+ );
+
+ drop(deletion);
+ assert_eq!(
+ coordinator.join(&join_request("orders")).await.error,
+ ERROR_NONE
+ );
+ }
+
#[test]
fn given_a_decoded_sync_when_normalizing_should_not_point_into_the_frame() {
let frame = Bytes::from_static(b"g m consumer range f blob");
diff --git a/gateways/kafka/src/group/offset_work.rs b/gateways/kafka/src/group/offset_work.rs
new file mode 100644
index 000000000..d11d85906
--- /dev/null
+++ b/gateways/kafka/src/group/offset_work.rs
@@ -0,0 +1,266 @@
+// 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.
+
+//! Keeps each deletion of a group's offsets apart from the group's joins and commits.
+//!
+//! Kafka's coordinator takes the requests of one group one at a time. Here a deletion makes many
+//! Iggy calls, so a join or a commit of the group could land in the middle. While a deletion runs,
+//! the coordinator refuses the group's joins and commits, and the deletion waits for each commit
+//! that passed its check before the deletion began. So a commit lands whole before the deletion or
+//! whole after it, and no member subscribes to a topic between the check of `OffsetDelete` and its
+//! deletions.
+
+use std::collections::HashMap;
+use std::sync::{Mutex, MutexGuard, PoisonError};
+
+use kafka_protocol::protocol::StrBytes;
+use tokio::sync::watch;
+use tokio::time::{Instant, timeout_at};
+
+/// The commits and the offset deletions that run for each group.
+pub(super) struct OffsetWork {
+ /// A group with no commit and no deletion that runs has no entry. Updated under the
+ /// coordinator's lock, and by a drop, which cannot wait for that lock.
+ running: Mutex<HashMap<StrBytes, Running>>,
+ /// Sent when the last commit of a group ends while a deletion of its offsets waits.
+ drained: watch::Sender<()>,
+}
+
+#[derive(Default)]
+struct Running {
+ commits: usize,
+ deletions: usize,
+}
+
+impl OffsetWork {
+ pub(super) fn new() -> Self {
+ Self {
+ running: Mutex::new(HashMap::new()),
+ drained: watch::Sender::new(()),
+ }
+ }
+
+ /// Whether a deletion of the offsets of `group_id` runs.
+ pub(super) fn deleting(&self, group_id: &StrBytes) -> bool {
+ self.lock()
+ .get(group_id)
+ .is_some_and(|running| running.deletions > 0)
+ }
+
+ /// Counts a commit of `group_id` until the returned value drops. `None` while a deletion of
+ /// the group's offsets runs.
+ pub(super) fn begin_commit(&self, group_id: &StrBytes) -> Option<RunningCommit<'_>> {
+ let group_id = super::owned_str(group_id);
+ let mut running = self.lock();
+ let group = running.entry(group_id.clone()).or_default();
+ if group.deletions > 0 {
+ return None;
+ }
+ group.commits += 1;
+ drop(running);
+ Some(RunningCommit {
+ work: self,
+ group_id,
+ })
+ }
+
+ /// Counts a deletion of the offsets of each group of `group_ids` until the returned value
+ /// drops.
+ pub(super) fn begin_deletion<'g>(
+ &self,
+ group_ids: impl IntoIterator<Item = &'g StrBytes>,
+ ) -> RunningDeletion<'_> {
+ let group_ids: Vec<StrBytes> = group_ids.into_iter().map(super::owned_str).collect();
+ let mut running = self.lock();
+ for group_id in &group_ids {
+ running.entry(group_id.clone()).or_default().deletions += 1;
+ }
+ drop(running);
+ RunningDeletion {
+ work: self,
+ group_ids,
+ }
+ }
+
+ fn committing(&self, group_ids: &[StrBytes]) -> bool {
+ let running = self.lock();
+ group_ids
+ .iter()
+ .any(|group_id| running.get(group_id).is_some_and(|group| group.commits > 0))
+ }
+
+ /// Each update is whole, so the map stays sound after a panic elsewhere.
+ fn lock(&self) -> MutexGuard<'_, HashMap<StrBytes, Running>> {
+ self.running.lock().unwrap_or_else(PoisonError::into_inner)
+ }
+}
+
+/// A commit of one group, counted until it drops. Keep it until the commit's Iggy calls end.
+pub struct RunningCommit<'a> {
+ work: &'a OffsetWork,
+ group_id: StrBytes,
+}
+
+impl Drop for RunningCommit<'_> {
+ fn drop(&mut self) {
+ let mut running = self.work.lock();
+ let Some(group) = running.get_mut(&self.group_id) else {
+ return;
+ };
+ group.commits = group.commits.saturating_sub(1);
+ let drained = group.commits == 0 && group.deletions > 0;
+ if group.commits == 0 && group.deletions == 0 {
+ running.remove(&self.group_id);
+ }
+ drop(running);
+ if drained {
+ self.work.drained.send_replace(());
+ }
+ }
+}
+
+/// A deletion of the offsets of some groups, counted until it drops. Keep it until the deletion's
+/// Iggy calls end.
+pub struct RunningDeletion<'a> {
+ work: &'a OffsetWork,
+ group_ids: Vec<StrBytes>,
+}
+
+impl RunningDeletion<'_> {
+ /// Waits until no commit of the groups runs. `false` when one still runs at `deadline`.
+ pub async fn commits_ended(&self, deadline: Instant) -> bool {
+ let mut drained = self.work.drained.subscribe();
+ while self.work.committing(&self.group_ids) {
+ if !matches!(timeout_at(deadline, drained.changed()).await, Ok(Ok(()))) {
+ return false;
+ }
+ }
+ true
+ }
+}
+
+impl Drop for RunningDeletion<'_> {
+ fn drop(&mut self) {
+ let mut running = self.work.lock();
+ for group_id in &self.group_ids {
+ let Some(group) = running.get_mut(group_id) else {
+ continue;
+ };
+ group.deletions = group.deletions.saturating_sub(1);
+ if group.commits == 0 && group.deletions == 0 {
+ running.remove(group_id);
+ }
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use super::*;
+
+ fn id(group: &str) -> StrBytes {
+ StrBytes::from_string(group.to_owned())
+ }
+
+ fn later() -> Instant {
+ Instant::now() + Duration::from_secs(5)
+ }
+
+ #[test]
+ fn given_a_deletion_should_refuse_commits_of_its_groups_until_it_ends() {
+ let work = OffsetWork::new();
+ let deletion = work.begin_deletion(&[id("orders"), id("payments")]);
+
+ assert!(work.deleting(&id("orders")));
+ assert!(work.begin_commit(&id("orders")).is_none());
+ assert!(work.begin_commit(&id("payments")).is_none());
+ assert!(
+ work.begin_commit(&id("audit")).is_some(),
+ "another group commits"
+ );
+
+ drop(deletion);
+ assert!(!work.deleting(&id("orders")));
+ assert!(work.begin_commit(&id("orders")).is_some());
+ }
+
+ #[test]
+ fn given_two_deletions_of_a_group_should_refuse_commits_until_both_end() {
+ let work = OffsetWork::new();
+ let first = work.begin_deletion(&[id("orders")]);
+ let second = work.begin_deletion(&[id("orders")]);
+
+ drop(first);
+ assert!(work.begin_commit(&id("orders")).is_none());
+ drop(second);
+ assert!(work.begin_commit(&id("orders")).is_some());
+ }
+
+ #[test]
+ fn given_every_commit_and_deletion_ended_should_keep_no_entry() {
+ let work = OffsetWork::new();
+ let commits = [
+ work.begin_commit(&id("orders")),
+ work.begin_commit(&id("orders")),
+ ];
+ let deletion = work.begin_deletion(&[id("orders"), id("payments")]);
+
+ drop(commits);
+ drop(deletion);
+
+ assert!(work.lock().is_empty());
+ }
+
+ /// A commit that passed its check before the deletion began lands whole before it.
+ #[tokio::test(start_paused = true)]
+ async fn given_a_commit_that_runs_should_start_the_deletion_after_it_ends() {
+ let work = OffsetWork::new();
+ let commit = work.begin_commit(&id("orders"));
+ let deletion = work.begin_deletion(&[id("orders")]);
+
+ let (ended, ()) = tokio::join!(deletion.commits_ended(later()), async {
+ tokio::time::sleep(Duration::from_millis(10)).await;
+ drop(commit);
+ });
+
+ assert!(ended);
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_a_commit_past_the_deadline_should_report_that_it_still_runs() {
+ let work = OffsetWork::new();
+ let _commit = work.begin_commit(&id("orders"));
+ let deletion = work.begin_deletion(&[id("orders")]);
+
+ assert!(
+ !deletion
+ .commits_ended(Instant::now() + Duration::from_millis(20))
+ .await
+ );
+ }
+
+ #[tokio::test(start_paused = true)]
+ async fn given_commits_of_other_groups_should_not_wait_for_them() {
+ let work = OffsetWork::new();
+ let _commit = work.begin_commit(&id("payments"));
+ let deletion = work.begin_deletion(&[id("orders")]);
+
+ assert!(deletion.commits_ended(later()).await);
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/delete_groups.rs b/gateways/kafka/src/protocol/handlers/delete_groups.rs
index cc6c57d09..a7e2e76b7 100644
--- a/gateways/kafka/src/protocol/handlers/delete_groups.rs
+++ b/gateways/kafka/src/protocol/handlers/delete_groups.rs
@@ -24,8 +24,16 @@
//! offsets in Iggy. The gateway deletes the Iggy group `kafka.cg.<group>` on every Kafka topic,
//! and Iggy deletes the offsets with it (`IggyBridge::delete_offset_groups`).
//!
+//! While the deletion runs, the group takes no join and no commit: both get
+//! `COORDINATOR_LOAD_IN_PROGRESS` (14), which clients retry. The deletion waits for each commit
+//! that passed its check before the deletion began, and answers 14 when one still runs at the
+//! deadline. A member on another gateway is not seen here, so a group whose members all use
+//! another gateway is deleted (`docs/OFFSET_STORAGE.md`).
+//!
//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30) for
-//! every group, and nothing is deleted.
+//! every group, and nothing is deleted. Each group to delete also gets 30, and nothing is deleted,
+//! when the bridge user may not list the Iggy groups of a Kafka topic: a group can hold offsets
+//! there.
//!
//! An id repeated in one request is answered once, as in `DescribeGroups`. The response has no
//! top-level error, so a request that cannot be decoded closes the connection.
@@ -40,10 +48,10 @@ use crate::bridge::{BridgeError, IggyBridge, OffsetCalls};
use crate::error::Result;
use crate::group::{GroupRemoval, is_valid_group_id};
use crate::protocol::api::{
- API_KEY_DELETE_GROUPS, ApiVersionRange, ConnectionState, ERROR_GROUP_AUTHORIZATION_FAILED,
- ERROR_GROUP_ID_NOT_FOUND, ERROR_INVALID_GROUP_ID, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
- ERROR_TOPIC_AUTHORIZATION_FAILED, ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome,
- is_supported_version,
+ API_KEY_DELETE_GROUPS, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_GROUP_AUTHORIZATION_FAILED, ERROR_GROUP_ID_NOT_FOUND, ERROR_INVALID_GROUP_ID,
+ ERROR_NON_EMPTY_GROUP, ERROR_NONE, ERROR_TOPIC_AUTHORIZATION_FAILED,
+ ERROR_UNKNOWN_SERVER_ERROR, GatewayState, HandleOutcome, is_supported_version,
};
use crate::protocol::bounds_guard::validate_delete_groups_shape;
use crate::protocol::dedup::dedup_first_seen;
@@ -119,7 +127,9 @@ async fn delete(
.filter(|group_id| is_valid_group_id(group_id))
.cloned()
.collect();
- let mut removals = state.groups.remove_empty(&valid).await.into_iter();
+ // Until this drops, the groups whose offsets go take no join and no commit.
+ let (removals, deletion) = state.groups.remove_empty(&valid).await;
+ let mut removals = removals.into_iter();
// Per group: its answer, or whether the coordinator held it when its offsets are to go.
let plans: Vec<core::result::Result<bool, i16>> = group_ids
.iter()
@@ -142,11 +152,16 @@ async fn delete(
let mut deleted = match &state.bridge {
Some(bridge) if !deletable.is_empty() => {
let calls = OffsetCalls::new(offset_deadline(state, connection).await);
- delete_offsets(bridge, &deletable, &calls).await
+ if deletion.commits_ended(calls.deadline()).await {
+ delete_offsets(bridge, &deletable, &calls).await
+ } else {
+ vec![Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS); deletable.len()]
+ }
}
_ => Vec::new(),
}
.into_iter();
+ drop(deletion);
plans
.into_iter()
.map(|plan| match plan {
@@ -231,6 +246,7 @@ pub fn encode_response(version: i16, results: Vec<DeletableGroupResult>) -> Resu
#[cfg(test)]
mod tests {
+ use std::sync::Arc;
use std::time::Duration;
use bytes::BytesMut;
@@ -238,7 +254,7 @@ mod tests {
use tokio_util::sync::CancellationToken;
use super::*;
- use crate::group::{GroupCoordinator, GroupCoordinatorConfig, JoinRequest};
+ use crate::group::{CommitRequest, GroupCoordinator, GroupCoordinatorConfig, JoinRequest};
use crate::protocol::acl::PrincipalPermissions;
use crate::protocol::api::ERROR_MEMBER_ID_REQUIRED;
@@ -377,6 +393,40 @@ mod tests {
);
}
+ /// A commit that passed its check before the deletion began must land whole before it. So
+ /// the deletion waits for it, and answers 14 while it still runs at the deadline.
+ #[tokio::test]
+ async fn given_a_commit_that_runs_past_the_deadline_should_answer_coordinator_load_in_progress()
+ {
+ // A 50 ms offset deadline.
+ let groups = GroupCoordinatorConfig {
+ min_session_timeout: Duration::from_millis(100),
+ ..GroupCoordinatorConfig::default()
+ };
+ let state = GatewayState::new(
+ Some(Arc::new(IggyBridge::offline())),
+ 8 * 1024 * 1024,
+ 0,
+ GroupCoordinator::new(groups, CancellationToken::new()),
+ );
+ let commit = CommitRequest {
+ group_id: StrBytes::from_static_str("orders-app"),
+ generation_id: -1,
+ member_id: StrBytes::new(),
+ group_instance_id: None,
+ };
+ let _running = state
+ .groups
+ .validate_commit(&commit)
+ .await
+ .expect("a commit outside the group protocol passes");
+
+ assert_eq!(
+ codes(&state, None, 2, &["orders-app"]).await,
+ vec![("orders-app".to_string(), ERROR_COORDINATOR_LOAD_IN_PROGRESS)]
+ );
+ }
+
/// The response has no top-level error, so an unreadable request has no answer.
#[tokio::test]
async fn given_an_unreadable_request_should_close_the_connection() {
diff --git a/gateways/kafka/src/protocol/handlers/describe_groups.rs b/gateways/kafka/src/protocol/handlers/describe_groups.rs
index c4aa0ba63..3fb7d394c 100644
--- a/gateways/kafka/src/protocol/handlers/describe_groups.rs
+++ b/gateways/kafka/src/protocol/handlers/describe_groups.rs
@@ -148,14 +148,21 @@ impl Absent {
described_group(version, GroupDescription::offsets_only(group_id.clone()))
}
Self::Dead | Self::Known(_) => missing_group(version, group_id),
- Self::Unknown => loading_group(group_id),
+ Self::Unknown => {
+ error_group(version, group_id, ERROR_COORDINATOR_LOAD_IN_PROGRESS, None)
+ }
}
}
}
pub(crate) fn described_group(version: i16, group: GroupDescription) -> DescribedGroup {
if group.error != ERROR_NONE {
- return too_large_group(version, &group.group_id);
+ return error_group(
+ version,
+ &group.group_id,
+ ERROR_UNKNOWN_SERVER_ERROR,
+ Some(DESCRIBE_RESPONSE_TOO_LARGE),
+ );
}
DescribedGroup::default()
.with_group_id(GroupId(group.group_id))
@@ -166,29 +173,26 @@ pub(crate) fn described_group(version: i16, group: GroupDescription) -> Describe
}
/// Empty member list and empty strings, matching Kafka's `DescribeGroupsResponse.groupError`.
-/// The code is not retried by the admin tool, unlike closing the connection.
-fn too_large_group(version: i16, group_id: &StrBytes) -> DescribedGroup {
- let mut group = DescribedGroup::default()
- .with_error_code(ERROR_UNKNOWN_SERVER_ERROR)
+/// `message` goes out from v6. A group that does not fit the response gets
+/// `UNKNOWN_SERVER_ERROR`, which the admin tool does not retry, unlike closing the connection.
+/// One that waits for the first scan gets `COORDINATOR_LOAD_IN_PROGRESS`, which it retries.
+fn error_group(
+ version: i16,
+ group_id: &StrBytes,
+ code: i16,
+ message: Option<&'static str>,
+) -> DescribedGroup {
+ let group = DescribedGroup::default()
+ .with_error_code(code)
.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(""));
if version >= FIRST_ERROR_MESSAGE_VERSION {
- group =
- group.with_error_message(Some(StrBytes::from_static_str(DESCRIBE_RESPONSE_TOO_LARGE)));
+ group.with_error_message(message.map(StrBytes::from_static_str))
+ } else {
+ group
}
- 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 {
diff --git a/gateways/kafka/src/protocol/handlers/offset_commit.rs b/gateways/kafka/src/protocol/handlers/offset_commit.rs
index c21d36952..2b9e9eb76 100644
--- a/gateways/kafka/src/protocol/handlers/offset_commit.rs
+++ b/gateways/kafka/src/protocol/handlers/offset_commit.rs
@@ -71,17 +71,19 @@ pub async fn handle(
}
};
- let error = state
+ let validated = state
.groups
.validate_commit(&CommitRequest::from(&request))
.await;
- let topics = match (&state.bridge, error) {
- (Some(bridge), ERROR_NONE) => {
+ let topics = match (&state.bridge, validated) {
+ // The commit stays counted until its Iggy calls end, so a deletion of the group's offsets
+ // waits for it.
+ (Some(bridge), Ok(_commit)) => {
let calls = OffsetCalls::new(offset_deadline(state, connection).await);
commit_all(bridge, &request, &calls).await
}
- (None, ERROR_NONE) => same_code(&request, ERROR_COORDINATOR_LOAD_IN_PROGRESS),
- (_, error) => same_code(&request, error),
+ (None, Ok(_)) => same_code(&request, ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ (_, Err(error)) => same_code(&request, error),
};
respond_or_close(encode_response(api_version, topics), "OffsetCommit")
}
diff --git a/gateways/kafka/src/protocol/handlers/offset_delete.rs b/gateways/kafka/src/protocol/handlers/offset_delete.rs
index f8b0c0231..ad242b85c 100644
--- a/gateways/kafka/src/protocol/handlers/offset_delete.rs
+++ b/gateways/kafka/src/protocol/handlers/offset_delete.rs
@@ -30,6 +30,12 @@
//! `COORDINATOR_LOAD_IN_PROGRESS` (14) for the whole group, which the Java admin client retries.
//! Deleting an offset twice does no harm. v1 is not in a Kafka release, so only v0 is advertised.
//!
+//! While the deletion runs, the group takes no join and no commit: both get 14, which clients
+//! retry. So no member subscribes to a topic between the check and the deletions, and a commit
+//! lands whole before or after them. The deletions wait for each commit that passed its check
+//! before them, and the group gets 14 when one still runs at the deadline. A member on another
+//! gateway is not seen here, so its topics count as unsubscribed (`docs/OFFSET_STORAGE.md`).
+//!
//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30), and
//! nothing is deleted.
@@ -43,7 +49,7 @@ use kafka_protocol::messages::offset_delete_response::{
use kafka_protocol::messages::{OffsetDeleteRequest, OffsetDeleteResponse};
use crate::auth::AuthenticatedPrincipal;
-use crate::bridge::{BridgeError, IggyBridge, OffsetCalls};
+use crate::bridge::{BridgeError, IggyBridge, OffsetCalls, TopicTarget};
use crate::error::Result;
use crate::group::{OffsetDeletion, Subscribed, is_valid_group_id};
use crate::protocol::api::{
@@ -117,17 +123,18 @@ async fn delete_offsets(
if !is_valid_group_id(group) {
return Err(ERROR_INVALID_GROUP_ID);
}
- let deletion = state.groups.offset_deletion(group).await;
+ // Until this drops, the group takes no join and no commit.
+ let (deletable, deletion) = state.groups.offset_deletion(group).await;
let Some(bridge) = &state.bridge else {
// No group holds offsets, and no topic exists.
- return match deletion {
+ return match deletable {
OffsetDeletion::Absent => Err(ERROR_GROUP_ID_NOT_FOUND),
OffsetDeletion::Refused => Err(ERROR_NON_EMPTY_GROUP),
OffsetDeletion::Allowed(_) => Ok(same_code(request, ERROR_UNKNOWN_TOPIC_OR_PARTITION)),
};
};
let calls = OffsetCalls::new(offset_deadline(state, connection).await);
- let subscribed = match deletion {
+ let subscribed = match deletable {
OffsetDeletion::Refused => return Err(ERROR_NON_EMPTY_GROUP),
OffsetDeletion::Allowed(subscribed) => subscribed,
// No members here, so the group exists only while it holds offsets.
@@ -137,17 +144,20 @@ async fn delete_offsets(
Some(_) => return Err(ERROR_GROUP_ID_NOT_FOUND),
},
};
- let partitions = partition_counts(bridge, request, &calls).await?;
- delete_unsubscribed(bridge, request, &partitions, &subscribed, &calls).await
+ let topics = requested_topics(bridge, request, &calls).await?;
+ if !deletion.commits_ended(calls.deadline()).await {
+ return Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ delete_unsubscribed(bridge, request, &topics, &subscribed, &calls).await
}
-/// The partition count of each requested topic, or the code for each of its partitions. A lookup
-/// that a retry can fix fails the group.
-async fn partition_counts<'a>(
+/// The Iggy topic and the partition count of each requested topic, or the code for each of its
+/// partitions. A lookup that a retry can fix fails the group.
+async fn requested_topics<'a>(
bridge: &IggyBridge,
request: &'a OffsetDeleteRequest,
calls: &OffsetCalls,
-) -> core::result::Result<HashMap<&'a str, core::result::Result<u32, i16>>, i16> {
+) -> core::result::Result<HashMap<&'a str, core::result::Result<(TopicTarget, u32), i16>>, i16> {
let mut names: Vec<&str> = request
.topics
.iter()
@@ -163,8 +173,8 @@ async fn partition_counts<'a>(
.await;
let mut found = HashMap::with_capacity(names.len());
for (name, count) in names.into_iter().zip(counts) {
- let count = match count {
- Ok(Some(count)) => Ok(count),
+ let topic = match count {
+ Ok(Some(topic)) => Ok(topic),
// Kafka answers a topic that it does not know with 3, whatever its name.
Ok(None) | Err(BridgeError::InvalidKafkaTopicName { .. }) => {
Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION)
@@ -176,7 +186,7 @@ async fn partition_counts<'a>(
code => Err(code),
},
};
- found.insert(name, count);
+ found.insert(name, topic);
}
Ok(found)
}
@@ -186,31 +196,26 @@ async fn partition_counts<'a>(
async fn delete_unsubscribed(
bridge: &IggyBridge,
request: &OffsetDeleteRequest,
- partitions: &HashMap<&str, core::result::Result<u32, i16>>,
+ requested: &HashMap<&str, core::result::Result<(TopicTarget, u32), i16>>,
subscribed: &Subscribed,
calls: &OffsetCalls,
) -> core::result::Result<Vec<OffsetDeleteResponseTopic>, i16> {
let group = request.group_id.0.as_str();
- let targets: Vec<_> = request
- .topics
- .iter()
- .map(|topic| bridge.topic_target(topic.name.0.as_str()))
- .collect();
let mut topics = same_code(request, ERROR_NONE);
let mut deletions = Vec::new();
// Where each deletion's code goes: (topic, partition) in `topics`.
let mut positions = Vec::new();
- for (topic_at, (topic, target)) in request.topics.iter().zip(&targets).enumerate() {
+ for (topic_at, topic) in request.topics.iter().enumerate() {
let name = topic.name.0.as_str();
- let count = partitions
- .get(name)
- .copied()
- .unwrap_or(Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION));
+ let found = match requested.get(name) {
+ Some(Ok(found)) => Ok(found),
+ Some(Err(code)) => Err(*code),
+ None => Err(ERROR_UNKNOWN_TOPIC_OR_PARTITION),
+ };
for (partition_at, partition) in topic.partitions.iter().enumerate() {
- let index = u32::try_from(partition.partition_index).ok();
- let code = match (count, index, target) {
- (Err(code), _, _) => code,
- (Ok(count), Some(index), Ok(target)) if index < count => {
+ let code = match (found, u32::try_from(partition.partition_index)) {
+ (Err(code), _) => code,
+ (Ok((target, count)), Ok(index)) if index < *count => {
if subscribed.contains(name) {
ERROR_GROUP_SUBSCRIBED_TO_TOPIC
} else {
@@ -219,8 +224,7 @@ async fn delete_unsubscribed(
continue;
}
}
- (Ok(_), _, Err(error)) => error.to_offset_error_code(),
- (Ok(_), _, Ok(_)) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ (Ok(_), _) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
};
topics[topic_at].partitions[partition_at].error_code = code;
}
diff --git a/gateways/kafka/tests/offset_groups_real_bridge_tests.rs b/gateways/kafka/tests/offset_groups_real_bridge_tests.rs
index 735f81b32..0f5c12f66 100644
--- a/gateways/kafka/tests/offset_groups_real_bridge_tests.rs
+++ b/gateways/kafka/tests/offset_groups_real_bridge_tests.rs
@@ -27,21 +27,26 @@ mod iggy_server;
#[path = "common/wire.rs"]
mod wire;
+use std::collections::{BTreeMap, HashMap};
use std::sync::Arc;
use std::time::Duration;
-use iggy::prelude::{ConsumerGroupClient, Identifier};
+use iggy::prelude::{
+ ConsumerGroupClient, Identifier, Permissions, StreamClient, StreamPermissions, UserClient,
+ UserStatus,
+};
+use secrecy::SecretString;
use serial_test::serial;
use tokio_util::sync::CancellationToken;
-use iggy_gateway_kafka::bridge::IggyBridge;
+use iggy_gateway_kafka::bridge::{IggyBridge, IggyBridgeConfig, TopicMapping, TopicOverride};
use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
use iggy_gateway_kafka::protocol::api::{
API_KEY_DELETE_GROUPS, API_KEY_DESCRIBE_GROUPS, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
API_KEY_LIST_GROUPS, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_DELETE, API_KEY_OFFSET_FETCH,
- ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_GROUP_ID_NOT_FOUND, ERROR_GROUP_SUBSCRIBED_TO_TOPIC,
- ERROR_NON_EMPTY_GROUP, ERROR_NONE, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState,
- handle_request_bounded,
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_GROUP_AUTHORIZATION_FAILED, ERROR_GROUP_ID_NOT_FOUND,
+ ERROR_GROUP_SUBSCRIBED_TO_TOPIC, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, handle_request_bounded,
};
use codec::Decoder;
@@ -82,7 +87,11 @@ struct Described {
}
async fn connected_state(server: &TestServer) -> GatewayState {
- let bridge = IggyBridge::connect(server.test_config())
+ state_with(server.test_config()).await
+}
+
+async fn state_with(config: IggyBridgeConfig) -> GatewayState {
+ let bridge = IggyBridge::connect(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.
@@ -708,3 +717,93 @@ async fn given_members_kafka_cannot_read_delete_offsets_should_keep_every_offset
(ERROR_NON_EMPTY_GROUP, vec![])
);
}
+
+/// The bridge user may not read an override topic, so `Metadata` leaves it out. A group can hold
+/// offsets there from before the bridge user lost access. The gateway cannot delete them, so it
+/// deletes nothing and does not answer success.
+#[tokio::test]
+#[serial]
+async fn given_an_override_topic_the_bridge_user_may_not_read_delete_should_refuse_and_keep_every_offset()
+ {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let mapping = || {
+ let audit = TopicOverride {
+ stream: "hidden".to_string(),
+ topic: "audit".to_string(),
+ };
+ TopicMapping::new(
+ "kafka".to_string(),
+ HashMap::from([("audit".to_string(), audit)]),
+ )
+ .expect("a valid mapping")
+ };
+ let root = IggyBridgeConfig {
+ topic_mapping: mapping(),
+ ..server.test_config()
+ };
+ IggyBridge::connect(root)
+ .await
+ .expect("root bridge should connect")
+ .ensure_stream_and_topic("audit", 1)
+ .await
+ .expect("create hidden/audit");
+ 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("hidden"), &iggy_id("audit"), "kafka.cg.orders-app")
+ .await
+ .expect("an Iggy group with offsets on hidden/audit");
+ let kafka_stream = raw
+ .get_stream(&iggy_id("kafka"))
+ .await
+ .expect("get the stream")
+ .expect("the stream exists")
+ .id;
+ let read_kafka = StreamPermissions {
+ read_stream: true,
+ ..StreamPermissions::default()
+ };
+ raw.create_user(
+ "kafka-bridge",
+ "kafka-bridge-secret",
+ UserStatus::Active,
+ Some(Permissions {
+ streams: Some(BTreeMap::from([(
+ usize::try_from(kafka_stream).expect("a small stream id"),
+ read_kafka,
+ )])),
+ ..Permissions::default()
+ }),
+ )
+ .await
+ .expect("create a bridge user that may read only the stream kafka");
+ let state = state_with(IggyBridgeConfig {
+ username: "kafka-bridge".to_string(),
+ password: SecretString::from("kafka-bridge-secret"),
+ topic_mapping: mapping(),
+ ..server.test_config()
+ })
+ .await;
+ commit(&state, GROUP, "orders", 4).await;
+
+ assert_eq!(
+ delete_groups(&state, &[GROUP]).await,
+ vec![(GROUP.to_string(), ERROR_GROUP_AUTHORIZATION_FAILED)]
+ );
+
+ assert_eq!(
+ committed(&state, GROUP, "orders").await,
+ 4,
+ "nothing is deleted"
+ );
+ let hidden = raw
+ .get_consumer_group(
+ &iggy_id("hidden"),
+ &iggy_id("audit"),
+ &iggy_id("kafka.cg.orders-app"),
+ )
+ .await
+ .expect("get the Iggy group");
+ assert!(hidden.is_some(), "the offsets on hidden/audit stay");
+}
diff --git a/gateways/kafka/tools/kafka-tool/src/response.rs b/gateways/kafka/tools/kafka-tool/src/response.rs
index b81f6ad43..00ce6b8a0 100644
--- a/gateways/kafka/tools/kafka-tool/src/response.rs
+++ b/gateways/kafka/tools/kafka-tool/src/response.rs
@@ -19,9 +19,9 @@
use bytes::Bytes;
use kafka_protocol::messages::{
- AlterConfigsResponse, ApiKey, ApiVersionsResponse, CreateTopicsResponse,
+ AlterConfigsResponse, ApiKey, ApiVersionsResponse, CreateTopicsResponse, DeleteGroupsResponse,
DescribeConfigsResponse, FetchResponse, InitProducerIdResponse, ListOffsetsResponse,
- MetadataResponse, ProduceResponse,
+ MetadataResponse, OffsetDeleteResponse, ProduceResponse,
};
use kafka_protocol::protocol::Decodable;
@@ -108,7 +108,7 @@ fn is_acceptable_verify_error(api_key: i16, error_code: i16) -> bool {
0..=2 => error_code == 6, // Produce/Fetch/ListOffsets stub: NOT_LEADER_OR_FOLLOWER
3 => error_code == 3, // Metadata stub: UNKNOWN_TOPIC_OR_PARTITION
19 | 32 | 33 => error_code == 41, // CreateTopics / DescribeConfigs / AlterConfigs stub
- 47 => error_code == 69, // OffsetDelete: `test-group` holds no offsets
+ 42 | 47 => error_code == 69, // DeleteGroups / OffsetDelete: `test-group` does not exist
_ => false,
}
}
@@ -388,6 +388,47 @@ fn decode_body(
));
}
}
+ 42 => {
+ let resp = DeleteGroupsResponse::decode(&mut buf, api_version)?;
+ details.push(format!("throttle_time_ms={}", resp.throttle_time_ms));
+ details.push(format!("results={}", resp.results.len()));
+ for (i, result) in resp.results.iter().enumerate().take(4) {
+ codes.push(result.error_code);
+ details.push(format!(
+ "results[{i}]: group={} ec={} ({})",
+ result.group_id.0,
+ result.error_code,
+ format_error_code(result.error_code)
+ ));
+ }
+ }
+ 47 => {
+ let resp = OffsetDeleteResponse::decode(&mut buf, api_version)?;
+ codes.push(resp.error_code);
+ details.push(format!(
+ "top_level.error_code={} ({})",
+ resp.error_code,
+ format_error_code(resp.error_code)
+ ));
+ details.push(format!("throttle_time_ms={}", resp.throttle_time_ms));
+ details.push(format!("topics={}", resp.topics.len()));
+ for (ti, topic) in resp.topics.iter().enumerate().take(4) {
+ details.push(format!(
+ "topics[{ti}]: name={} partitions={}",
+ topic_name(&topic.name),
+ topic.partitions.len()
+ ));
+ for (pi, p) in topic.partitions.iter().enumerate().take(4) {
+ codes.push(p.error_code);
+ details.push(format!(
+ " partitions[{pi}]: index={} ec={} ({})",
+ p.partition_index,
+ p.error_code,
+ format_error_code(p.error_code)
+ ));
+ }
+ }
+ }
22 => {
let resp = InitProducerIdResponse::decode(&mut buf, api_version)?;
codes.push(resp.error_code);
@@ -430,6 +471,8 @@ fn format_error_code(code: i16) -> &'static str {
38 => "INVALID_REPLICATION_FACTOR",
41 => "NOT_CONTROLLER",
42 => "INVALID_REQUEST",
+ 68 => "NON_EMPTY_GROUP",
+ 69 => "GROUP_ID_NOT_FOUND",
-1 => "UNKNOWN",
_ => "OTHER",
}