You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "krishvishal (via GitHub)" <gi...@apache.org> on 2026/10/15 16:02:20 UTC
[PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
krishvishal opened a new pull request, #4559:
URL: https://github.com/apache/iggy/pull/4559
Relates to #4511
## Rationale
When the last consumer of a group leaves, the coordinator drops the group, but its committed offsets stay in Iggy. Kafka keeps such a group and reports it as `Empty`. Before this change, `kafka-consumer-groups.sh --list` left the group out, and `--describe` printed `Consumer group 'g-demo' does not exist.`. Also, `--delete` and `--delete-offsets` failed with `UnsupportedVersionException`, because the gateway did not serve DeleteGroups (42) or OffsetDelete (47). So an operator had no way to see or remove the offsets of a stopped consumer group.
Each member's client id and peer address, the last part of #4511, waits for #4535, which adds the peer address to each connection.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on PR #4559:
URL: https://github.com/apache/iggy/pull/4559#issuecomment-6312370033
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4559:
URL: https://github.com/apache/iggy/pull/4559#discussion_r4312596483
##########
gateways/kafka/src/protocol/handlers/list_groups.rs:
##########
@@ -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 {
Review Comment:
warning: until a scan succeeds, ListGroups answers 14 for the whole request and hides live groups too. A failed scan never publishes anything (`group_index.rs:172`), so with Iggy down at boot or a scan over its 30s cap, `--list` never answers again. Before this PR it listed live groups. DescribeGroups for missing ids and OffsetDelete for an absent group fail the same way.
Fix: record the failure, answer 14 only while the first scan is pending, and fall back to live groups after it fails.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -185,6 +207,108 @@ impl IggyBridge {
self.read_in_slot(slot_of(topic, 0), calls, lookup).await
}
+ /// Deletes the Iggy group that holds the offsets of each Kafka group of `groups`, on every
+ /// Kafka topic. Iggy deletes the offsets with it. One result per group, in order: whether any
+ /// topic held the group.
+ ///
+ /// 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 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, 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 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(
+ 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, listed) in listing.topics.iter().zip(listings) {
+ let found = match listed {
+ Ok(found) => found,
+ 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() {
+ if found.iter().any(|name| name == group) {
+ deletions.push((at, topic.kafka_topic.as_str()));
+ }
+ }
+ }
+ 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)| {
Review Comment:
warning: the per-topic deletes share one deadline. If topic A is deleted and topic B times out, the answer is 14, the fence lifts, and a consumer that joins before the retry reads A from `auto.offset.reset` and B from its old offset. A sticky fence would wedge the group if the admin never retries.
Fix: with the owned `RunningDeletion` above, hold the fence until every spawned delete ends, and document the partial state.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -200,6 +324,111 @@ 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, 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
+ /// 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 { 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. 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 = async {
+ let client = self.scan_client.connected(&self.config).await?;
+ let mut groups = BTreeSet::new();
+ for topic in self.list_kafka_topics_on(&client).await? {
Review Comment:
warning: the scan makes one `get_consumer_groups` call per topic, one after another, on one client, under a 30s cap. At a few thousand topics every scan times out and throws away its partial result, and `until` drops the scan client, so each retry reconnects and logs in again.
Fix: bound the concurrency (`buffer_unordered`) and keep the partial or last result.
##########
gateways/kafka/src/protocol/handlers/delete_groups.rs:
##########
@@ -0,0 +1,448 @@
+// 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! 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. 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+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_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;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ // 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()
+ .map(|group_id| {
+ if !is_valid_group_id(group_id) {
+ return Err(ERROR_INVALID_GROUP_ID);
+ }
+ match removals.next() {
+ Some(GroupRemoval::Live) => Err(ERROR_NON_EMPTY_GROUP),
+ removal => Ok(removal == Some(GroupRemoval::Removed)),
+ }
+ })
+ .collect();
+ let deletable: Vec<&str> = group_ids
+ .iter()
+ .zip(&plans)
+ .filter(|(_, plan)| plan.is_ok())
+ .map(|(group_id, _)| group_id.as_str())
+ .collect();
+ let mut deleted = match &state.bridge {
+ Some(bridge) if !deletable.is_empty() => {
+ let calls = OffsetCalls::new(offset_deadline(state, connection).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);
Review Comment:
critical: the mirror case. When `delete_consumer_group` times out, the handler answers 14 and drops the fence here while the delete still runs in its `in_slot` task. A commit for a partition on another slot is not queued behind it, so it lands and gets 0. The late delete then wipes that acked offset.
Fix: move an owned `RunningDeletion` into the spawned delete, so the group stays fenced (14) until the Iggy call ends. `SLOT_LIMIT` bounds that.
##########
gateways/kafka/src/bridge/iggy_bridge/group_index.rs:
##########
@@ -0,0 +1,398 @@
+// 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 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, HashMap};
+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
Review Comment:
warning: freshness counts from when the scan started, so a scan that takes 10s or more publishes a result that is already stale. Every request then waits for a new scan, and OffsetDelete on a memberless group answers 14 every time. Also, the comment on `SCAN_MAX_AGE` (line 32) does not hold: a 30s kafka-ui poll never hits a 10s cache.
Fix: count age from when the scan ended, or serve the stale result at once and refresh in the background.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -1332,10 +1348,85 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// Kafka's `ClassicGroup.validateDeleteGroup`. A group with no joined members is `Empty`, so it
+/// goes, with the ids that wait to join it. A join with one of those ids gets
+/// `UNKNOWN_MEMBER_ID`, and the client joins again from scratch.
+pub fn remove_empty(groups: &mut Groups, group_id: &StrBytes, now: Instant) -> GroupRemoval {
+ if !tick_group(groups, group_id, now) {
+ return GroupRemoval::Absent;
+ }
+ if groups
+ .get(group_id)
+ .is_some_and(|group| !group.members.is_empty())
+ {
+ return GroupRemoval::Live;
+ }
+ groups.remove(group_id);
+ GroupRemoval::Removed
+}
+
+/// Kafka's `ClassicGroup.validateOffsetDelete` and `isSubscribedToTopic`.
+pub fn offset_deletion(groups: &mut Groups, group_id: &StrBytes, now: Instant) -> OffsetDeletion {
+ if !tick_group(groups, group_id, now) {
+ return OffsetDeletion::Absent;
+ }
+ let Some(group) = groups.get(group_id) else {
+ return OffsetDeletion::Absent;
+ };
+ if group.members.is_empty() {
+ return OffsetDeletion::Allowed(Subscribed::Topics(HashSet::new()));
+ }
+ if group.protocol_type.as_str() != CONSUMER_PROTOCOL_TYPE {
+ return OffsetDeletion::Refused;
+ }
+ OffsetDeletion::Allowed(group.subscribed())
+}
+
+/// The topics of a consumer's `ConsumerProtocolSubscription`: a version, then the topics and the
+/// user data of v0, which every later version keeps as its prefix. Kafka reads the same prefix.
+/// `None` when the bytes do not hold it.
+///
+/// Read by hand, not by `kafka_protocol`, which sizes a `Vec` from a declared count before it
+/// checks the bytes (see `protocol::bounds_guard`), and a member chose these bytes.
+fn subscription_topics(metadata: &[u8]) -> Option<Vec<String>> {
+ let mut rest = metadata;
+ let _version = take_i16(&mut rest)?;
Review Comment:
nit: the subscription version is read and ignored. Kafka's `checkSubscriptionVersion` rejects a negative version, so the member counts as subscribed to every topic. Here it decodes as v0 and narrows the set. Also, `Vec::with_capacity(count)` at line 1399 reserves a 24-byte `String` per 2 wire bytes, under the coordinator lock.
Fix: return `None` (all topics) for version < 0, and extend the `HashSet` directly.
##########
gateways/kafka/src/protocol/handlers/delete_groups.rs:
##########
@@ -0,0 +1,448 @@
+// 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! 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. 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+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_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;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ // 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
Review Comment:
nit: two unrelated bools share the type `Result<bool, i16>`: here "the coordinator held it", in `delete_offsets` "an Iggy topic held it". They are matched by position through `unwrap_or(Ok(false))` at line 169. The order is right today, but easy to break.
Fix: use a small enum, e.g. `Plan::Answer(i16)` / `Plan::Delete { held: bool }`.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -200,6 +324,111 @@ 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, 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
+ /// 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 { 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. 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 = async {
+ let client = self.scan_client.connected(&self.config).await?;
+ 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),
Review Comment:
nit: the scan skips topics the bridge user cannot list, so the group shows as `Empty`, but DeleteGroups refuses the same group with 30 on those topics. `kafka-consumer-groups.sh` then reports `GroupAuthorizationException`, which blames the Kafka principal when the bridge user's permission is the cause.
Fix: add a README troubleshooting line naming the log message.
##########
gateways/kafka/src/protocol/api.rs:
##########
@@ -125,6 +127,13 @@ pub const ERROR_GROUP_ID_NOT_FOUND: i16 = ResponseError::GroupIdNotFound.code();
pub const ERROR_INVALID_SESSION_TIMEOUT: i16 = 26;
/// How a follower learns to rejoin: its heartbeat is answered with this while the group prepares.
pub const ERROR_REBALANCE_IN_PROGRESS: i16 = 27;
+/// `DeleteGroups`: the bridge user may not delete the Iggy group that holds the offsets.
Review Comment:
nit: this doc names one use of 30. It is also the answer for a principal without the permission (`handlers/mod.rs:163-176`) and OffsetDelete's top-level refusal. Also, `delete_groups.rs:33`, `offset_delete.rs:39` and the README say `manage_topics`, while CONSUMER_GROUPS.md says `manage_topics` or `manage_streams`. Pick one wording.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -185,6 +207,108 @@ impl IggyBridge {
self.read_in_slot(slot_of(topic, 0), calls, lookup).await
}
+ /// Deletes the Iggy group that holds the offsets of each Kafka group of `groups`, on every
+ /// Kafka topic. Iggy deletes the offsets with it. One result per group, in order: whether any
+ /// topic held the group.
+ ///
+ /// 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 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, 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 listing =
+ |client: Arc<IggyClient>| async move { self.kafka_topic_listing_on(&client).await };
+ let listing = self.read_in_slot(LISTING_SLOT, calls, listing).await?;
Review Comment:
warning: DeleteGroups lists every topic on the shared offset read lanes, in queue with consumer OffsetFetch and inside one 3s deadline. A timeout drops the read client (`fetch.rs:150-152`), so fetches on those lanes reconnect. Reads on this client can also lag a group the commit-lane client just created, and the answer is then 0/69 while offsets survive.
Fix: list on the scan client, and confirm each held topic on the commit-lane client before answering.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -200,6 +324,111 @@ 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, 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
+ /// 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 { 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. 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 = async {
+ let client = self.scan_client.connected(&self.config).await?;
+ 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),
+ }
+ }
+ Ok(groups)
+ };
+ self.scan_client
+ .until(Instant::now() + GROUP_SCAN_LIMIT, scan)
+ .await
+ }
+
+ /// The Kafka groups with offsets on `kafka_topic`, listed on an offset slot.
+ 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 { kafka_groups_on(&client, &topic).await };
+ self.read_in_slot(slot, calls, listing).await
+ }
+
+ /// Deletes the Iggy group of Kafka group `group` on `kafka_topic`, then waits until the topic
+ /// no longer lists it. A group that is not there counts as deleted.
+ async fn delete_offset_group_on(
+ &self,
+ group: &str,
+ kafka_topic: &str,
+ calls: &OffsetCalls,
+ ) -> Result<(), BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let topic = self.topic_target(kafka_topic)?;
+ let (stream_id, topic_id) = (topic.stream_id.clone(), topic.topic_id.clone());
+ let delete = move |client: Arc<IggyClient>| async move {
+ match client
+ .delete_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ {
+ Err(
+ IggyError::ConsumerGroupIdNotFound(..)
+ | IggyError::ConsumerGroupNameNotFound(..),
+ ) => Ok(()),
+ deleted => landed(deleted),
+ }
+ };
+ // Partition 0's slot, so the delete follows the commits of that partition.
+ self.commit_in_slot(slot_of(&topic, 0), calls, delete)
Review Comment:
critical: a commit that times out keeps running in its `in_slot` task for up to `SLOT_LIMIT` (45s), but `RunningCommit` drops when the handler returns. A later DeleteGroups sees `commits_ended` true, deletes on partition 0's lane only and answers 0. The late commit on another slot then hits `ConsumerGroupNameNotFound`, `create_offset_group` recreates `kafka.cg.<group>` and stores a stale offset. That breaks the rule at `offset_work.rs:22-24`.
Fix: make `OffsetWork` an `Arc`, move an owned guard into each `in_slot` task, and add a handler-level test with the bridge blocked past the deadline. Taking every commit lane instead would stall commits of every group.
##########
gateways/kafka/src/protocol/handlers/delete_groups.rs:
##########
@@ -0,0 +1,448 @@
+// 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! 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. 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+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_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;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ // Until this drops, the groups whose offsets go take no join and no commit.
+ let (removals, deletion) = state.groups.remove_empty(&valid).await;
Review Comment:
nit: `remove_empty` drops the pending member ids before the Iggy step runs. If that step answers 14 (listing timeout, say), the Java admin retries, finds the group absent with no offsets, and gets 69 for a group that existed. Kafka changes nothing on a failed delete.
Fix: remove the pending state only after the offset deletion succeeds.
##########
gateways/kafka/src/protocol/handlers/offset_delete.rs:
##########
@@ -0,0 +1,485 @@
+// 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.
+
+//! `OffsetDelete` (API key 47).
+//!
+//! Deletes committed offsets of one group, as `kafka-consumer-groups.sh --delete-offsets` asks.
+//! Kafka's rules (`ClassicGroup.validateOffsetDelete`, `OffsetMetadataManager.deleteOffsets`):
+//!
+//! - A group that has no members here and holds no offsets in Iggy gets `GROUP_ID_NOT_FOUND` (69).
+//! - A group with members of a protocol other than `consumer` gets `NON_EMPTY_GROUP` (68).
+//! - Each partition of a topic that a member subscribes to gets `GROUP_SUBSCRIBED_TO_TOPIC` (86).
+//! - A partition that does not exist gets `UNKNOWN_TOPIC_OR_PARTITION` (3).
+//! - Any other partition gets 0, also one that held no offset.
+//!
+//! A group error leaves the topics out. A deletion that a retry can fix answers
+//! `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.
+
+use std::collections::{HashMap, HashSet};
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_delete_response::{
+ OffsetDeleteResponsePartition, OffsetDeleteResponseTopic,
+};
+use kafka_protocol::messages::{OffsetDeleteRequest, OffsetDeleteResponse};
+
+use crate::auth::AuthenticatedPrincipal;
+use crate::bridge::{BridgeError, IggyBridge, OffsetCalls, TopicTarget};
+use crate::error::Result;
+use crate::group::{OffsetDeletion, Subscribed, is_valid_group_id};
+use crate::protocol::api::{
+ API_KEY_OFFSET_DELETE, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_GROUP_AUTHORIZATION_FAILED, ERROR_GROUP_ID_NOT_FOUND, ERROR_GROUP_SUBSCRIBED_TO_TOPIC,
+ ERROR_INVALID_GROUP_ID, ERROR_INVALID_REQUEST, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_delete_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_DELETE,
+ min_version: 0,
+ max_version: 0,
+};
+
+/// Partition index and error code of one echoed partition.
+const PER_PARTITION_BYTES: usize = 6;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_OFFSET_DELETE, api_version) {
+ return unsupported_version_response(API_KEY_OFFSET_DELETE, api_version, |version| {
+ encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let request = match decode_guarded::<OffsetDeleteRequest>(api_version, body, |version, body| {
+ validate_offset_delete_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetDelete request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "OffsetDelete",
+ );
+ }
+ };
+
+ if !may_delete_groups(principal, "OffsetDelete") {
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_GROUP_AUTHORIZATION_FAILED),
+ "OffsetDelete",
+ );
+ }
+ let encoded = match delete_offsets(state, connection, &request).await {
+ Ok(topics) => encode_response(api_version, topics),
+ Err(code) => encode_error_response(api_version, code),
+ };
+ respond_or_close(encoded, "OffsetDelete")
+}
+
+/// The answer for each requested partition, or the group's error.
+async fn delete_offsets(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ request: &OffsetDeleteRequest,
+) -> core::result::Result<Vec<OffsetDeleteResponseTopic>, i16> {
+ let group = &request.group_id.0;
+ if !is_valid_group_id(group) {
+ return Err(ERROR_INVALID_GROUP_ID);
+ }
+ // 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 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 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.
+ OffsetDeletion::Absent => match bridge.offset_groups(calls.deadline()).await {
Review Comment:
warning: the scan wait uses the same deadline as the whole request. With a stale cache it waits out the deadline, so `requested_topics` then times out and answers 14. The cache-only check also answers non-retriable 69 for a group that committed through another gateway less than 10s ago, while DeleteGroups, which lists live, finds it.
Fix: on a cache miss, probe the requested topics with `holds_offset_group` before falling back to the scan.
##########
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 {
Review Comment:
nit: `committing()` scans every group id of each running deletion, under the std mutex, on every drain wake. `begin_commit` takes the same mutex under the coordinator lock, so a DeleteGroups with many ids slows commits of every group.
Fix: keep a commit counter per group so the check is O(1).
##########
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)) => {
Review Comment:
nit: the guard lives only because the tuple scrutinee is a temporary that lasts to the end of the `let` (`Ok(_)` would keep it too). A refactor like `let ok = validated.is_ok();` would drop it before `commit_all`, and no test drives `offset_commit::handle` against DeleteGroups.
Fix: bind `Ok(commit)` and `drop(commit)` explicitly after `commit_all`, and cover it with the race test from the `group_offsets.rs:414` comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
Posted by "numinnex (via GitHub)" <gi...@apache.org>.
numinnex commented on code in PR #4559:
URL: https://github.com/apache/iggy/pull/4559#discussion_r4335688282
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -221,9 +508,16 @@ impl IggyBridge {
Ok(connected) => connected?,
Err(_elapsed) => return Err(BridgeError::Timeout),
};
+ let (commit, kept) = (call(client), calls.kept.clone());
+ let call = async move {
+ let ended = timeout(SLOT_LIMIT, commit).await;
+ // Held for as long as the call runs.
+ drop(kept);
Review Comment:
critical: the fence drops when this call returns, not when the write can no longer land. At 30 s the SDK returns `OperationUncertain` (`write_with_recovery` only handles `LaneBlocked`), yet the delete still lands later, as late bytes or as the SDK's replay on this client's next metadata write. The `SLOT_LIMIT` cut at :513 has the same gap.
Reproduced on a real server: a failed DeleteGroups wiped an acked commit, and after DeleteGroups answered 0 a late commit brought the deleted offset back.
Fix: free the lane on return and hold `kept` in a detached task until the outcome is proven (`resolve_operation`, or re-delete by numeric group id until Ok/NotFound; `OperationNotRetained` proves nothing). Add a test that stalls the lane past 30 s.
##########
gateways/kafka/src/protocol/handlers/delete_groups.rs:
##########
@@ -0,0 +1,533 @@
+// 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! 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`). The ids that wait to
+//! join a group go only once its offsets did, so a deletion that fails changes nothing, as in
+//! Kafka.
+//!
+//! 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 commit or a deletion runs until its Iggy calls end, also those that its request
+//! stopped waiting for, for at most 45 s. A deletion that does not end by the deadline answers 14
+//! too. Some topics can then hold no offsets of the group while others still do, and a retry
+//! deletes the rest. A member on another gateway is not seen here, so a group whose members all
+//! use another gateway is deleted (`docs/OFFSET_STORAGE.md`).
+//!
+//! When a listener names principals, a principal without a global `manage_topics` or
+//! `manage_streams` gets `GROUP_AUTHORIZATION_FAILED` (30) for 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+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_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;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// What one group of a request gets before the Iggy step.
+#[derive(Clone, Copy)]
+enum Plan {
+ /// This answer, and nothing to delete.
+ Answer(i16),
+ /// Its offsets go. `held`: this coordinator holds the group, so it exists even when no topic
+ /// holds its offsets.
+ Delete { held: bool },
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ // Until this drops, the groups whose offsets go take no join and no commit.
+ let (removals, deletion) = state.groups.group_deletion(&valid).await;
+ let mut removals = removals.into_iter();
+ let plans: Vec<Plan> = group_ids
+ .iter()
+ .map(|group_id| {
+ if !is_valid_group_id(group_id) {
+ return Plan::Answer(ERROR_INVALID_GROUP_ID);
+ }
+ match removals.next() {
+ Some(GroupRemoval::Live) => Plan::Answer(ERROR_NON_EMPTY_GROUP),
+ removal => Plan::Delete {
+ held: removal == Some(GroupRemoval::Empty),
+ },
+ }
+ })
+ .collect();
+ // The index in `group_ids` of each group whose offsets go.
+ let deletable: Vec<usize> = plans
+ .iter()
+ .enumerate()
+ .filter(|(_, plan)| matches!(plan, Plan::Delete { .. }))
+ .map(|(at, _)| at)
+ .collect();
+ let names: Vec<&str> = deletable.iter().map(|&at| group_ids[at].as_str()).collect();
+ let mut codes: Vec<i16> = plans
+ .iter()
+ .map(|plan| match plan {
+ Plan::Answer(code) => *code,
+ Plan::Delete { .. } => ERROR_NONE,
+ })
+ .collect();
+ let found = match &state.bridge {
+ Some(bridge) if !names.is_empty() => {
+ let deadline = offset_deadline(state, connection).await;
+ if deletion.commits_ended(deadline).await {
+ // Each delete keeps the groups fenced until its Iggy calls end, also past the
+ // deadline, so no commit lands while a delete can still land.
+ let calls = OffsetCalls::new(deadline).keeping(deletion);
+ let found = delete_offsets(bridge, &names, &calls).await;
+ // Still fenced by `calls`, so no member joins before the removal.
+ remove_emptied(state, group_ids, &plans, &deletable, &found).await;
+ found
+ } else {
+ vec![Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS); names.len()]
+ }
+ }
+ // Without a bridge no group holds offsets.
+ _ => {
+ let found = vec![Ok(false); names.len()];
+ remove_emptied(state, group_ids, &plans, &deletable, &found).await;
+ found
+ }
+ };
+ for (&at, found) in deletable.iter().zip(found) {
+ let Plan::Delete { held } = plans[at] else {
+ continue;
+ };
+ codes[at] = match found {
+ Ok(found) if found || held => ERROR_NONE,
+ Ok(_) => ERROR_GROUP_ID_NOT_FOUND,
Review Comment:
warning: a delete that landed but answered 14 (removal still finishing in `until_unlisted` at the deadline) is retried by the Java admin within the same call. The retry gets 14 (`LifecycleBusy`) until Iggy finishes, then lists nothing and lands here, so `--delete` reports `GroupIdNotFoundException` for a group it deleted.
Fix: remember groups this gateway deleted for a short TTL and answer 0 for them.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -201,9 +314,183 @@ impl IggyBridge {
self.read_in_slot(LISTING_SLOT, calls, listing).await
}
+ /// The Kafka groups that hold offsets in Iggy, from the last scan, plus the groups that this
+ /// gateway stored offsets for since, less those it deleted. The groups of a scan answer at once
+ /// for 10 s after it ends. After that a caller starts a new scan, and still gets the older
+ /// groups at once for up to 1 min after their scan ended. Otherwise the caller waits for the
+ /// new scan, until `deadline`.
+ ///
+ /// A scan lists the Kafka topics, then the Iggy groups of each topic, so it costs one call per
+ /// topic. It runs on a client of its own, a few topics at once, so no offset call waits behind
+ /// it. A scan keeps running after its caller stops waiting, for up to 30 s, and serves the next
+ /// caller.
+ pub async fn offset_groups(self: &Arc<Self>, deadline: Instant) -> ScannedGroups {
+ let bridge = Arc::clone(self);
+ let scan = move || async move { bridge.scan_offset_groups().await };
+ self.group_index.groups(deadline, scan).await
+ }
+
+ /// The groups of the last scan, if one found them, without a scan.
+ #[must_use]
+ pub fn last_offset_groups(&self) -> Option<Arc<OffsetGroups>> {
+ self.group_index.last()
+ }
+
+ /// Every Kafka group with offsets on a Kafka topic. A topic deleted since the listing holds
+ /// 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. A scan that reaches `GROUP_SCAN_LIMIT` keeps
+ /// what it found.
+ async fn scan_offset_groups(&self) -> Result<Scanned, BridgeError> {
+ let deadline = Instant::now() + GROUP_SCAN_LIMIT;
+ let topics = async {
+ let client = self.scan_client.connected(&self.config).await?;
+ let topics = self.list_kafka_topics_on(&client).await?;
+ Ok::<_, BridgeError>((client, topics))
+ };
+ let (client, topics) = timeout_at(deadline, topics)
+ .await
+ .map_err(|_elapsed| BridgeError::Timeout)??;
+ let mut groups = BTreeSet::new();
+ let mut listings = pin!(self.kafka_groups_by_topic(&client, &topics));
+ loop {
+ let (topic, listed) = match timeout_at(deadline, listings.next()).await {
+ Ok(Some(listing)) => listing,
+ Ok(None) => {
+ return Ok(Scanned {
+ groups,
+ whole: true,
+ });
+ }
+ Err(_elapsed) => {
+ warn!(
+ topics = topics.len(),
+ "the scan for the Kafka groups that hold offsets did not list every topic \
+ in {GROUP_SCAN_LIMIT:?}; keeping the groups it found"
+ );
+ return Ok(Scanned {
+ groups,
+ whole: false,
+ });
+ }
+ };
+ match listed {
+ Ok(found) => groups.extend(found),
+ Err(error)
+ if topic_deleted(&error)
+ || matches!(error, BridgeError::Iggy(IggyError::Unauthorized)) =>
+ {
+ debug!(topic, %error, "skipping a topic in the scan for groups");
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ }
+
+ /// The index in `groups` and the Kafka topic of each Iggy group that holds offsets of a group
+ /// of `groups`. Listed on the scan client.
+ async fn offset_group_topics(
+ &self,
+ groups: &[&str],
+ ) -> Result<Vec<(usize, String)>, BridgeError> {
+ let client = self.scan_client.connected(&self.config).await?;
+ let listing = self.kafka_topic_listing_on(&client).await?;
+ let mut unexamined = listing.unreadable;
+ let mut deletions = Vec::new();
+ let mut listings = pin!(self.kafka_groups_by_topic(&client, &listing.topics));
+ while let Some((topic, listed)) = listings.next().await {
+ let found = match listed {
+ Ok(found) => found,
+ Err(error) if topic_deleted(&error) => continue,
+ Err(BridgeError::Iggy(IggyError::Unauthorized)) => {
+ unexamined.push(topic.to_owned());
+ continue;
+ }
+ Err(error) => return Err(error),
+ };
+ for (at, group) in groups.iter().enumerate() {
+ if found.iter().any(|name| name == group) {
+ deletions.push((at, topic.to_owned()));
+ }
+ }
+ }
+ 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));
+ }
+ Ok(deletions)
+ }
+
+ /// The Kafka groups with offsets on each topic of `topics`, listed on `client`,
+ /// `LISTING_CALLS` topics at once. Each item is a Kafka topic and its listing, in the order the
+ /// listings end.
+ fn kafka_groups_by_topic<'a>(
+ &'a self,
+ client: &'a IggyClient,
+ topics: &'a [KafkaTopicMetadata],
+ ) -> impl Stream<Item = (&'a str, Result<Vec<String>, BridgeError>)> + 'a {
+ stream::iter(topics)
+ .map(move |topic| async move {
+ let listed = match self.topic_target(&topic.kafka_topic) {
+ Ok(target) => kafka_groups_on(client, &target).await,
+ Err(error) => Err(error),
+ };
+ (topic.kafka_topic.as_str(), listed)
+ })
+ .buffer_unordered(LISTING_CALLS)
+ }
+
+ /// Deletes the Iggy group of Kafka group `group` on `kafka_topic`, on the commit lane of
+ /// partition 0 of the topic, so the delete follows the commits of that partition. A group or
+ /// a topic that is not there counts as deleted.
+ ///
+ /// Iggy answers once every partition of the topic dropped the group. Until then a commit could
+ /// still store an offset there, and Iggy would delete it later. In a cluster another replica
+ /// can list the group a little longer, so the delete then checks on its own client, which reads
+ /// what it wrote, until the topic no longer lists the group. The delete keeps the lane, and
+ /// what `calls` keep, until both steps end, for at most `SLOT_LIMIT`, also past the deadline
+ /// of `calls`.
+ async fn delete_offset_group_on(
+ &self,
+ group: &str,
+ kafka_topic: &str,
+ calls: &OffsetCalls,
+ ) -> Result<(), BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let topic = self.topic_target(kafka_topic)?;
+ let slot = slot_of(&topic, 0);
+ let delete = move |client: Arc<IggyClient>| async move {
+ let deadline = (Instant::now() + SLOT_LIMIT).into_std();
+ let deleted = write_with_recovery(&client, deadline, || {
+ client.delete_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ })
+ .await;
+ let deleted = match deleted {
+ Err(
+ IggyError::ConsumerGroupIdNotFound(..)
+ | IggyError::ConsumerGroupNameNotFound(..),
+ ) => return Ok(()),
+ deleted => landed(deleted),
+ };
+ match deleted {
+ Ok(()) => until_unlisted(&client, &topic, &group_id).await,
Review Comment:
warning: `until_unlisted` runs inside the commit-lane task, so the lane stays taken until Iggy has removed the group on every partition. If a commit through another gateway recreates `kafka.cg.<g>` meanwhile, the name lookup never sees it gone and spins until `SLOT_LIMIT`, so commits of every group on this slot get 14 for 45 s.
Fix: compare the numeric group id, and run the wait outside the lane holding only `kept`.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -201,9 +314,183 @@ impl IggyBridge {
self.read_in_slot(LISTING_SLOT, calls, listing).await
}
+ /// The Kafka groups that hold offsets in Iggy, from the last scan, plus the groups that this
+ /// gateway stored offsets for since, less those it deleted. The groups of a scan answer at once
+ /// for 10 s after it ends. After that a caller starts a new scan, and still gets the older
+ /// groups at once for up to 1 min after their scan ended. Otherwise the caller waits for the
+ /// new scan, until `deadline`.
+ ///
+ /// A scan lists the Kafka topics, then the Iggy groups of each topic, so it costs one call per
+ /// topic. It runs on a client of its own, a few topics at once, so no offset call waits behind
+ /// it. A scan keeps running after its caller stops waiting, for up to 30 s, and serves the next
+ /// caller.
+ pub async fn offset_groups(self: &Arc<Self>, deadline: Instant) -> ScannedGroups {
+ let bridge = Arc::clone(self);
+ let scan = move || async move { bridge.scan_offset_groups().await };
+ self.group_index.groups(deadline, scan).await
+ }
+
+ /// The groups of the last scan, if one found them, without a scan.
+ #[must_use]
+ pub fn last_offset_groups(&self) -> Option<Arc<OffsetGroups>> {
+ self.group_index.last()
+ }
+
+ /// Every Kafka group with offsets on a Kafka topic. A topic deleted since the listing holds
+ /// 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. A scan that reaches `GROUP_SCAN_LIMIT` keeps
+ /// what it found.
+ async fn scan_offset_groups(&self) -> Result<Scanned, BridgeError> {
+ let deadline = Instant::now() + GROUP_SCAN_LIMIT;
+ let topics = async {
+ let client = self.scan_client.connected(&self.config).await?;
+ let topics = self.list_kafka_topics_on(&client).await?;
+ Ok::<_, BridgeError>((client, topics))
+ };
+ let (client, topics) = timeout_at(deadline, topics)
+ .await
+ .map_err(|_elapsed| BridgeError::Timeout)??;
+ let mut groups = BTreeSet::new();
+ let mut listings = pin!(self.kafka_groups_by_topic(&client, &topics));
+ loop {
+ let (topic, listed) = match timeout_at(deadline, listings.next()).await {
+ Ok(Some(listing)) => listing,
+ Ok(None) => {
+ return Ok(Scanned {
+ groups,
+ whole: true,
+ });
+ }
+ Err(_elapsed) => {
+ warn!(
+ topics = topics.len(),
+ "the scan for the Kafka groups that hold offsets did not list every topic \
+ in {GROUP_SCAN_LIMIT:?}; keeping the groups it found"
+ );
+ return Ok(Scanned {
+ groups,
+ whole: false,
+ });
+ }
+ };
+ match listed {
+ Ok(found) => groups.extend(found),
+ Err(error)
+ if topic_deleted(&error)
+ || matches!(error, BridgeError::Iggy(IggyError::Unauthorized)) =>
+ {
+ debug!(topic, %error, "skipping a topic in the scan for groups");
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ }
+
+ /// The index in `groups` and the Kafka topic of each Iggy group that holds offsets of a group
+ /// of `groups`. Listed on the scan client.
+ async fn offset_group_topics(
+ &self,
+ groups: &[&str],
+ ) -> Result<Vec<(usize, String)>, BridgeError> {
+ let client = self.scan_client.connected(&self.config).await?;
Review Comment:
warning: the topics to delete on come from the scan client, which can read a replica behind the commit-lane client that just created `kafka.cg.<g>` (`mod.rs:149-150`). A topic missed here is never deleted, so the answer is 0 or 69 while its offsets survive and the group comes back as `Empty`.
Fix: also delete on the topics this gateway's own commits stored to since the last whole scan.
##########
gateways/kafka/src/protocol/handlers/delete_groups.rs:
##########
@@ -0,0 +1,533 @@
+// 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! 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`). The ids that wait to
+//! join a group go only once its offsets did, so a deletion that fails changes nothing, as in
+//! Kafka.
+//!
+//! 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 commit or a deletion runs until its Iggy calls end, also those that its request
+//! stopped waiting for, for at most 45 s. A deletion that does not end by the deadline answers 14
+//! too. Some topics can then hold no offsets of the group while others still do, and a retry
+//! deletes the rest. A member on another gateway is not seen here, so a group whose members all
+//! use another gateway is deleted (`docs/OFFSET_STORAGE.md`).
+//!
+//! When a listener names principals, a principal without a global `manage_topics` or
+//! `manage_streams` gets `GROUP_AUTHORIZATION_FAILED` (30) for 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+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_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;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// What one group of a request gets before the Iggy step.
+#[derive(Clone, Copy)]
+enum Plan {
+ /// This answer, and nothing to delete.
+ Answer(i16),
+ /// Its offsets go. `held`: this coordinator holds the group, so it exists even when no topic
+ /// holds its offsets.
+ Delete { held: bool },
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ // Until this drops, the groups whose offsets go take no join and no commit.
+ let (removals, deletion) = state.groups.group_deletion(&valid).await;
+ let mut removals = removals.into_iter();
+ let plans: Vec<Plan> = group_ids
+ .iter()
+ .map(|group_id| {
+ if !is_valid_group_id(group_id) {
+ return Plan::Answer(ERROR_INVALID_GROUP_ID);
+ }
+ match removals.next() {
+ Some(GroupRemoval::Live) => Plan::Answer(ERROR_NON_EMPTY_GROUP),
+ removal => Plan::Delete {
+ held: removal == Some(GroupRemoval::Empty),
+ },
+ }
+ })
+ .collect();
+ // The index in `group_ids` of each group whose offsets go.
+ let deletable: Vec<usize> = plans
+ .iter()
+ .enumerate()
+ .filter(|(_, plan)| matches!(plan, Plan::Delete { .. }))
+ .map(|(at, _)| at)
+ .collect();
+ let names: Vec<&str> = deletable.iter().map(|&at| group_ids[at].as_str()).collect();
+ let mut codes: Vec<i16> = plans
+ .iter()
+ .map(|plan| match plan {
+ Plan::Answer(code) => *code,
+ Plan::Delete { .. } => ERROR_NONE,
+ })
+ .collect();
+ let found = match &state.bridge {
+ Some(bridge) if !names.is_empty() => {
+ let deadline = offset_deadline(state, connection).await;
Review Comment:
warning: this is the `offset_hold` deadline (half the shortest session, 3 s by default), sized for a heartbeat waiting on the connection. DeleteGroups lists every topic and runs every delete inside it, usually on an admin connection with no member. At ~5k topics and 10 ms round trips the listing alone overruns it, so `--delete` gets 14 on every Java retry, each repeating the full listing.
Fix: give a connection with no group member a deadline sized for admin requests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on PR #4559:
URL: https://github.com/apache/iggy/pull/4559#issuecomment-6197198029
/skill team-review-slim
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4559:
URL: https://github.com/apache/iggy/pull/4559#issuecomment-6199883424
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4559?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 89.49%. Comparing base ([`12cc1f3`](https://app.codecov.io/gh/apache/iggy/commit/12cc1f3cf3ef85debcb97421d5b9933dc18e2d8d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`649c5a5`](https://app.codecov.io/gh/apache/iggy/commit/649c5a5ebcec77b9b5661371984d96d5de796916?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4559 +/- ##
============================================
- Coverage 89.49% 89.49% -0.01%
Complexity 1784 1784
============================================
Files 1330 1330
Lines 284420 284420
Branches 242863 242863
============================================
- Hits 254551 254547 -4
+ Misses 24227 24217 -10
- Partials 5642 5656 +14
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.75% <ø> (-0.01%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.71% <ø> (ø)` | |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `79.37% <ø> (ø)` | |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `90.43% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.84% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `95.12% <ø> (ø)` | |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.88% <ø> (ø)` | |
| [C++ SDK](https://app.codecov.io/gh/apache/iggy/pull/4559/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.44% <ø> (ø)` | |
[see 21 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4559/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal merged PR #4559:
URL: https://github.com/apache/iggy/pull/4559
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): serve DeleteGroups and OffsetDelete, report Empty groups [iggy]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #4559:
URL: https://github.com/apache/iggy/pull/4559#discussion_r4286549596
##########
gateways/kafka/src/protocol/handlers/offset_delete.rs:
##########
@@ -0,0 +1,484 @@
+// 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.
+
+//! `OffsetDelete` (API key 47).
+//!
+//! Deletes committed offsets of one group, as `kafka-consumer-groups.sh --delete-offsets` asks.
+//! Kafka's rules (`ClassicGroup.validateOffsetDelete`, `OffsetMetadataManager.deleteOffsets`):
+//!
+//! - A group that has no members here and holds no offsets in Iggy gets `GROUP_ID_NOT_FOUND` (69).
+//! - A group with members of a protocol other than `consumer` gets `NON_EMPTY_GROUP` (68).
+//! - Each partition of a topic that a member subscribes to gets `GROUP_SUBSCRIBED_TO_TOPIC` (86).
+//! - A partition that does not exist gets `UNKNOWN_TOPIC_OR_PARTITION` (3).
+//! - Any other partition gets 0, also one that held no offset.
+//!
+//! A group error leaves the topics out. A deletion that a retry can fix answers
+//! `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.
+//!
+//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30), and
+//! nothing is deleted.
+
+use std::collections::{HashMap, HashSet};
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_delete_response::{
+ OffsetDeleteResponsePartition, OffsetDeleteResponseTopic,
+};
+use kafka_protocol::messages::{OffsetDeleteRequest, OffsetDeleteResponse};
+
+use crate::auth::AuthenticatedPrincipal;
+use crate::bridge::{BridgeError, IggyBridge, OffsetCalls};
+use crate::error::Result;
+use crate::group::{OffsetDeletion, Subscribed, is_valid_group_id};
+use crate::protocol::api::{
+ API_KEY_OFFSET_DELETE, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_GROUP_AUTHORIZATION_FAILED, ERROR_GROUP_ID_NOT_FOUND, ERROR_GROUP_SUBSCRIBED_TO_TOPIC,
+ ERROR_INVALID_GROUP_ID, ERROR_INVALID_REQUEST, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_delete_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_DELETE,
+ min_version: 0,
+ max_version: 0,
+};
+
+/// Partition index and error code of one echoed partition.
+const PER_PARTITION_BYTES: usize = 6;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_OFFSET_DELETE, api_version) {
+ return unsupported_version_response(API_KEY_OFFSET_DELETE, api_version, |version| {
+ encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let request = match decode_guarded::<OffsetDeleteRequest>(api_version, body, |version, body| {
+ validate_offset_delete_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetDelete request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "OffsetDelete",
+ );
+ }
+ };
+
+ if !may_delete_groups(principal, "OffsetDelete") {
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_GROUP_AUTHORIZATION_FAILED),
+ "OffsetDelete",
+ );
+ }
+ let encoded = match delete_offsets(state, connection, &request).await {
+ Ok(topics) => encode_response(api_version, topics),
+ Err(code) => encode_error_response(api_version, code),
+ };
+ respond_or_close(encoded, "OffsetDelete")
+}
+
+/// The answer for each requested partition, or the group's error.
+async fn delete_offsets(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ request: &OffsetDeleteRequest,
+) -> core::result::Result<Vec<OffsetDeleteResponseTopic>, i16> {
+ let group = &request.group_id.0;
+ if !is_valid_group_id(group) {
+ return Err(ERROR_INVALID_GROUP_ID);
+ }
+ let deletion = state.groups.offset_deletion(group).await;
+ let Some(bridge) = &state.bridge else {
+ // No group holds offsets, and no topic exists.
+ return match deletion {
+ 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 {
+ OffsetDeletion::Refused => return Err(ERROR_NON_EMPTY_GROUP),
+ OffsetDeletion::Allowed(subscribed) => subscribed,
+ // No members here, so the group exists only while it holds offsets.
+ OffsetDeletion::Absent => match bridge.offset_groups(calls.deadline()).await {
+ None => return Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ Some(groups) if groups.contains(group) => Subscribed::Topics(HashSet::new()),
Review Comment:
warning: `OffsetDelete` treats a group with no members here as having no subscribers, so it deletes the offsets of a topic a member subscribes to. The member can be live on another gateway, or join inside the 10-second window the `offset_groups()` answer comes from. Re-read the coordinator before the deletion, and document the other-gateway case.
##########
gateways/kafka/src/protocol/handlers/delete_groups.rs:
##########
@@ -0,0 +1,400 @@
+// 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! 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`).
+//!
+//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30) for
+//! every group, and nothing is deleted.
+//!
+//! 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+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,
+};
+use crate::protocol::bounds_guard::validate_delete_groups_shape;
+use crate::protocol::dedup::dedup_first_seen;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ let mut removals = state.groups.remove_empty(&valid).await.into_iter();
Review Comment:
critical: `delete()` drops the group's state at line 122 and deletes its offsets in Iggy at line 145. Between the two, `commit_step` answers `ERROR_NONE` to an `OffsetCommit` with `generation_id < 0`, so its offsets are acknowledged and then wiped. Refuse a join or a commit for a group while its deletion runs.
##########
gateways/kafka/src/protocol/handlers/delete_groups.rs:
##########
@@ -0,0 +1,400 @@
+// 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.
+
+//! `DeleteGroups` (API key 42).
+//!
+//! Deletes each named group that has no members here, with its offsets, as
+//! `kafka-consumer-groups.sh --delete` asks. Kafka's rules (`ClassicGroup.validateDeleteGroup`):
+//! a group with members gets `NON_EMPTY_GROUP` (68), and a group that does not exist gets
+//! `GROUP_ID_NOT_FOUND` (69). A group exists here while ids wait to join it, or while it holds
+//! 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`).
+//!
+//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30) for
+//! every group, and nothing is deleted.
+//!
+//! 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.
+
+use bytes::Bytes;
+use kafka_protocol::messages::delete_groups_response::DeletableGroupResult;
+use kafka_protocol::messages::{DeleteGroupsRequest, DeleteGroupsResponse, GroupId};
+use kafka_protocol::protocol::StrBytes;
+
+use crate::auth::AuthenticatedPrincipal;
+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,
+};
+use crate::protocol::bounds_guard::validate_delete_groups_shape;
+use crate::protocol::dedup::dedup_first_seen;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+/// v3 adds a per-group `error_message`, which `kafka-protocol` cannot encode yet.
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_DELETE_GROUPS,
+ min_version: 0,
+ max_version: 2,
+};
+
+/// Group id length prefix, error code and tagged fields of one result, past the id itself.
+const PER_RESULT_BYTES: usize = 5;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_DELETE_GROUPS, api_version) {
+ return unsupported_version_response(API_KEY_DELETE_GROUPS, api_version, |version| {
+ encode_response(version, Vec::new())
+ });
+ }
+ let request = match decode_guarded::<DeleteGroupsRequest>(api_version, body, |version, body| {
+ validate_delete_groups_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode DeleteGroups request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let requested: Vec<StrBytes> = request
+ .groups_names
+ .iter()
+ .map(|group_id| group_id.0.clone())
+ .collect();
+ let group_ids = dedup_first_seen(&requested);
+ let codes = if may_delete_groups(principal, "DeleteGroups") {
+ delete(state, connection, &group_ids).await
+ } else {
+ vec![ERROR_GROUP_AUTHORIZATION_FAILED; group_ids.len()]
+ };
+ let results = group_ids
+ .into_iter()
+ .zip(codes)
+ .map(|(group_id, code)| {
+ DeletableGroupResult::default()
+ .with_group_id(GroupId(group_id))
+ .with_error_code(code)
+ })
+ .collect();
+ respond_or_close(encode_response(api_version, results), "DeleteGroups")
+}
+
+/// One code per group of `group_ids`, in order.
+async fn delete(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ group_ids: &[StrBytes],
+) -> Vec<i16> {
+ let valid: Vec<StrBytes> = group_ids
+ .iter()
+ .filter(|group_id| is_valid_group_id(group_id))
+ .cloned()
+ .collect();
+ let mut removals = state.groups.remove_empty(&valid).await.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()
+ .map(|group_id| {
+ if !is_valid_group_id(group_id) {
+ return Err(ERROR_INVALID_GROUP_ID);
+ }
+ match removals.next() {
+ Some(GroupRemoval::Live) => Err(ERROR_NON_EMPTY_GROUP),
+ removal => Ok(removal == Some(GroupRemoval::Removed)),
+ }
+ })
+ .collect();
+ let deletable: Vec<&str> = group_ids
+ .iter()
+ .zip(&plans)
+ .filter(|(_, plan)| plan.is_ok())
+ .map(|(group_id, _)| group_id.as_str())
+ .collect();
+ 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
+ }
+ _ => Vec::new(),
+ }
+ .into_iter();
+ plans
+ .into_iter()
+ .map(|plan| match plan {
+ Err(code) => code,
+ Ok(held) => match deleted.next().unwrap_or(Ok(false)) {
+ Ok(found) if found || held => ERROR_NONE,
Review Comment:
warning: A topic the bridge user cannot read is dropped from the listing, so `delete_offset_groups` never looks for the group there. This line then answers success while the offsets on that topic stay. Answer an error for a group whose offsets could not be examined.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -200,6 +305,120 @@ 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);
Review Comment:
warning: A scan read holds its lane for up to 30 seconds, so a same-slot `OffsetFetch` times out after 3 and answers `COORDINATOR_LOAD_IN_PROGRESS` (14). Cap the scan's reads below the offset deadline, or give the scan its own lane.
##########
gateways/kafka/src/bridge/iggy_bridge/group_index.rs:
##########
@@ -0,0 +1,392 @@
+// 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 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::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>>,
+ /// 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)>>,
+}
+
+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.change(group, true);
+ }
+
+ /// Removes `group` from the groups of the last scan. This gateway deleted its offsets, so it
+ /// goes before the next scan.
+ pub(super) fn forget(&self, group: &str) {
+ self.change(group, false);
+ }
+
+ fn change(&self, group: &str, holds: bool) {
+ self.published.send_if_modified(|current| {
+ if let Some(changes) = current.changes.as_mut() {
Review Comment:
simplification: `changes` holds one entry per commit during a scan, and only a group's last entry matters, so a busy gateway grows the `Vec` for nothing. Keep the last value per group in a map instead. Saves: the duplicate entries and their memory.
##########
gateways/kafka/src/protocol/handlers/describe_groups.rs:
##########
@@ -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 {
Review Comment:
simplification: `loading_group` repeats the empty shape of `too_large_group` at line 170, and only the error code and the v6 message differ. Replace both with one builder that takes the code and an optional message. Saves: about 8 lines.
##########
gateways/kafka/src/protocol/handlers/offset_delete.rs:
##########
@@ -0,0 +1,484 @@
+// 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.
+
+//! `OffsetDelete` (API key 47).
+//!
+//! Deletes committed offsets of one group, as `kafka-consumer-groups.sh --delete-offsets` asks.
+//! Kafka's rules (`ClassicGroup.validateOffsetDelete`, `OffsetMetadataManager.deleteOffsets`):
+//!
+//! - A group that has no members here and holds no offsets in Iggy gets `GROUP_ID_NOT_FOUND` (69).
+//! - A group with members of a protocol other than `consumer` gets `NON_EMPTY_GROUP` (68).
+//! - Each partition of a topic that a member subscribes to gets `GROUP_SUBSCRIBED_TO_TOPIC` (86).
+//! - A partition that does not exist gets `UNKNOWN_TOPIC_OR_PARTITION` (3).
+//! - Any other partition gets 0, also one that held no offset.
+//!
+//! A group error leaves the topics out. A deletion that a retry can fix answers
+//! `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.
+//!
+//! With SASL on, a principal without `manage_topics` gets `GROUP_AUTHORIZATION_FAILED` (30), and
+//! nothing is deleted.
+
+use std::collections::{HashMap, HashSet};
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_delete_response::{
+ OffsetDeleteResponsePartition, OffsetDeleteResponseTopic,
+};
+use kafka_protocol::messages::{OffsetDeleteRequest, OffsetDeleteResponse};
+
+use crate::auth::AuthenticatedPrincipal;
+use crate::bridge::{BridgeError, IggyBridge, OffsetCalls};
+use crate::error::Result;
+use crate::group::{OffsetDeletion, Subscribed, is_valid_group_id};
+use crate::protocol::api::{
+ API_KEY_OFFSET_DELETE, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_GROUP_AUTHORIZATION_FAILED, ERROR_GROUP_ID_NOT_FOUND, ERROR_GROUP_SUBSCRIBED_TO_TOPIC,
+ ERROR_INVALID_GROUP_ID, ERROR_INVALID_REQUEST, ERROR_NON_EMPTY_GROUP, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, ERROR_UNSUPPORTED_VERSION,
+ GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_delete_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, may_delete_groups, offset_deadline, offset_refusal_cause,
+ respond_or_close, unsupported_version_response,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_DELETE,
+ min_version: 0,
+ max_version: 0,
+};
+
+/// Partition index and error code of one echoed partition.
+const PER_PARTITION_BYTES: usize = 6;
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ principal: Option<&AuthenticatedPrincipal>,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ if !is_supported_version(API_KEY_OFFSET_DELETE, api_version) {
+ return unsupported_version_response(API_KEY_OFFSET_DELETE, api_version, |version| {
+ encode_error_response(version, ERROR_UNSUPPORTED_VERSION)
+ });
+ }
+ let request = match decode_guarded::<OffsetDeleteRequest>(api_version, body, |version, body| {
+ validate_offset_delete_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetDelete request");
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_INVALID_REQUEST),
+ "OffsetDelete",
+ );
+ }
+ };
+
+ if !may_delete_groups(principal, "OffsetDelete") {
+ return respond_or_close(
+ encode_error_response(api_version, ERROR_GROUP_AUTHORIZATION_FAILED),
+ "OffsetDelete",
+ );
+ }
+ let encoded = match delete_offsets(state, connection, &request).await {
+ Ok(topics) => encode_response(api_version, topics),
+ Err(code) => encode_error_response(api_version, code),
+ };
+ respond_or_close(encoded, "OffsetDelete")
+}
+
+/// The answer for each requested partition, or the group's error.
+async fn delete_offsets(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ request: &OffsetDeleteRequest,
+) -> core::result::Result<Vec<OffsetDeleteResponseTopic>, i16> {
+ let group = &request.group_id.0;
+ if !is_valid_group_id(group) {
+ return Err(ERROR_INVALID_GROUP_ID);
+ }
+ let deletion = state.groups.offset_deletion(group).await;
+ let Some(bridge) = &state.bridge else {
+ // No group holds offsets, and no topic exists.
+ return match deletion {
+ 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 {
+ OffsetDeletion::Refused => return Err(ERROR_NON_EMPTY_GROUP),
+ OffsetDeletion::Allowed(subscribed) => subscribed,
+ // No members here, so the group exists only while it holds offsets.
+ OffsetDeletion::Absent => match bridge.offset_groups(calls.deadline()).await {
+ None => return Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ Some(groups) if groups.contains(group) => Subscribed::Topics(HashSet::new()),
+ Some(_) => return Err(ERROR_GROUP_ID_NOT_FOUND),
+ },
+ };
+ let partitions = partition_counts(bridge, request, &calls).await?;
+ delete_unsubscribed(bridge, request, &partitions, &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>(
+ bridge: &IggyBridge,
+ request: &'a OffsetDeleteRequest,
+ calls: &OffsetCalls,
+) -> core::result::Result<HashMap<&'a str, core::result::Result<u32, i16>>, i16> {
+ let mut names: Vec<&str> = request
+ .topics
+ .iter()
+ .map(|topic| topic.name.0.as_str())
+ .collect();
+ names.sort_unstable();
+ names.dedup();
+ let counts = join_all(
+ names
+ .iter()
+ .map(|name| bridge.kafka_topic_partitions(name, calls)),
+ )
+ .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),
+ // 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)
+ }
+ Err(error) => match error.to_offset_error_code() {
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS => {
+ return Err(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ code => Err(code),
+ },
+ };
+ found.insert(name, count);
+ }
+ Ok(found)
+}
+
+/// Deletes the offset of each requested partition that exists and whose topic no member
+/// subscribes to.
+async fn delete_unsubscribed(
+ bridge: &IggyBridge,
+ request: &OffsetDeleteRequest,
+ partitions: &HashMap<&str, core::result::Result<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() {
+ let name = topic.name.0.as_str();
+ let count = partitions
+ .get(name)
+ .copied()
+ .unwrap_or(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 => {
+ if subscribed.contains(name) {
+ ERROR_GROUP_SUBSCRIBED_TO_TOPIC
+ } else {
+ deletions.push((target, index, -1));
+ positions.push((topic_at, partition_at));
+ continue;
+ }
+ }
+ (Ok(_), _, Err(error)) => error.to_offset_error_code(),
Review Comment:
simplification: `Ok(count)` proves that `topic_target` accepted that name when `partition_counts` resolved it, so `target` cannot be `Err` here and this arm never fires. Delete the arm. Saves: one unreachable branch.
##########
gateways/kafka/tools/kafka-tool/src/main.rs:
##########
@@ -573,6 +576,24 @@ fn build_payload(api_key: i16, version: i16) -> Result<Bytes> {
.encode(&mut buf, version)
.context("SaslAuthenticate")?;
}
+ 42 => {
Review Comment:
nit: `response.rs` has no decoder for `DeleteGroupsResponse` or `OffsetDeleteResponse`, so `verify` reads the first two bytes of `throttle_time_ms` as the error code. A failed delete then reports as a pass. Add decoder arms for keys 42 and 47.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -200,6 +305,120 @@ 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?;
Review Comment:
simplification: The topic listing and the per-topic group fan-out here repeat lines 230-236 in `delete_offset_groups`, and only the post-processing differs. Extract a helper that yields each topic with its groups. Saves: about 15 lines.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org