krishvishal commented on code in PR #4415: URL: https://github.com/apache/iggy/pull/4415#discussion_r4211081142
########## gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs: ########## @@ -0,0 +1,574 @@ +// 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. + +//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`. + +use std::future::Future; +use std::hash::{DefaultHasher, Hash, Hasher}; +use std::sync::Arc; + +use futures::future::join_all; +use iggy::prelude::{ + Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError, +}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; +use tokio::time::{Instant, timeout, timeout_at}; +use tracing::info; + +use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget}; +use crate::bridge::error::BridgeError; + +/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group +/// apart from a native Iggy group of the same name. +pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg."; + +/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait +/// behind a Produce send. +const OFFSET_SLOTS: usize = 4; + +/// The slot of a call that belongs to no partition. Any slot will do. +const LISTING_SLOT: usize = 0; + +impl IggyBridge { + /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group + /// `group`. One result per commit, in order. + /// + /// A negative offset deletes the key instead. Kafka consumers read any negative offset as + /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A + /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already + /// applied counts as done. + /// + /// Each commit starts once its slot is free. Commits on one slot start in the order given. + /// + /// # Errors + /// + /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group, + /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not + /// started by then never starts. + pub async fn commit_group_offsets( + &self, + group: &str, + commits: &[(&TopicTarget, u32, i64)], + deadline: Instant, + ) -> Vec<Result<(), BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => Arc::new(names), + Err(error) => return refused_all(commits.len(), &error), + }; + // `join_all` polls each call once, in order, before any again, so calls on one slot + // queue for it in order. + join_all(commits.iter().map(|&(topic, partition, offset)| { + let (names, stream_id, topic_id) = ( + Arc::clone(&names), + topic.stream_id.clone(), + topic.topic_id.clone(), + ); + let commit = move |client: Arc<IggyClient>| async move { + commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await + }; + self.commit_in_slot(slot_of(topic, partition), deadline, commit) + })) + .await + } + + /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in + /// order. `None` when there is no key or no group. The reads run at once. + /// + /// # Errors + /// + /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too. + pub async fn fetch_group_offsets( + &self, + group: &str, + reads: &[(&TopicTarget, u32)], + deadline: Instant, + ) -> Vec<Result<Option<u64>, BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => names, + Err(error) => return refused_all(reads.len(), &error), + }; + let consumer = &names.consumer; + join_all(reads.iter().map(|&(topic, partition)| { + let read = move |client: Arc<IggyClient>| async move { + let info = client + .get_consumer_offset( + consumer, + &topic.stream_id, + &topic.topic_id, + Some(partition), + ) + .await + .map_err(BridgeError::Iggy)?; + Ok(info.map(|info| info.stored_offset)) + }; + self.read_in_slot(slot_of(topic, partition), deadline, read) + })) + .await + } + + /// Whether Kafka group `group` holds offsets on `topic`. + /// + /// # Errors + /// + /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group` + /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`. + pub async fn holds_offset_group( + &self, + group: &str, + topic: &TopicTarget, + deadline: Instant, + ) -> Result<bool, BridgeError> { + let group_id = OffsetNames::new(group) + .map_err(BridgeError::Iggy)? + .consumer + .id; + let lookup = |client: Arc<IggyClient>| async move { + client + .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id) + .await + .map(|found| found.is_some()) + .map_err(BridgeError::Iggy) + }; + // Any slot will do. Partition 0's keeps the choice fixed. + self.read_in_slot(slot_of(topic, 0), deadline, lookup).await + } + + /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce. + /// + /// # Errors + /// + /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`. + pub async fn list_kafka_topics_until( + &self, + deadline: Instant, + ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> { + let listing = + |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await }; + self.read_in_slot(LISTING_SLOT, deadline, listing).await + } + + /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps + /// the slot until it ends, even after its caller stops waiting, so the next call on the slot + /// cannot overtake it. A call that has not started by `deadline` never starts. + async fn commit_in_slot<F>( + &self, + slot: usize, + deadline: Instant, + call: impl FnOnce(Arc<IggyClient>) -> F, + ) -> Result<(), BridgeError> + where + F: Future<Output = Result<(), BridgeError>> + Send + 'static, + { + let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else { + return Err(BridgeError::Timeout); + }; + let client = match timeout_at(deadline, client.connected(&self.config)).await { + Ok(connected) => connected?, + Err(_elapsed) => return Err(BridgeError::Timeout), + }; + match timeout_at(deadline, spawn_holding(turn, call(client))).await { Review Comment: fixed. each slot has a read client and a commit client now, and a stuck commit only holds up other commits. ########## gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs: ########## @@ -0,0 +1,574 @@ +// 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. + +//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`. + +use std::future::Future; +use std::hash::{DefaultHasher, Hash, Hasher}; +use std::sync::Arc; + +use futures::future::join_all; +use iggy::prelude::{ + Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError, +}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; +use tokio::time::{Instant, timeout, timeout_at}; +use tracing::info; + +use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget}; +use crate::bridge::error::BridgeError; + +/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group +/// apart from a native Iggy group of the same name. +pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg."; + +/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait +/// behind a Produce send. +const OFFSET_SLOTS: usize = 4; + +/// The slot of a call that belongs to no partition. Any slot will do. +const LISTING_SLOT: usize = 0; + +impl IggyBridge { + /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group + /// `group`. One result per commit, in order. + /// + /// A negative offset deletes the key instead. Kafka consumers read any negative offset as + /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A + /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already + /// applied counts as done. + /// + /// Each commit starts once its slot is free. Commits on one slot start in the order given. + /// + /// # Errors + /// + /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group, + /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not + /// started by then never starts. + pub async fn commit_group_offsets( + &self, + group: &str, + commits: &[(&TopicTarget, u32, i64)], + deadline: Instant, + ) -> Vec<Result<(), BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => Arc::new(names), + Err(error) => return refused_all(commits.len(), &error), + }; + // `join_all` polls each call once, in order, before any again, so calls on one slot + // queue for it in order. + join_all(commits.iter().map(|&(topic, partition, offset)| { + let (names, stream_id, topic_id) = ( + Arc::clone(&names), + topic.stream_id.clone(), + topic.topic_id.clone(), + ); + let commit = move |client: Arc<IggyClient>| async move { + commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await + }; + self.commit_in_slot(slot_of(topic, partition), deadline, commit) + })) + .await + } + + /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in + /// order. `None` when there is no key or no group. The reads run at once. + /// + /// # Errors + /// + /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too. + pub async fn fetch_group_offsets( + &self, + group: &str, + reads: &[(&TopicTarget, u32)], + deadline: Instant, + ) -> Vec<Result<Option<u64>, BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => names, + Err(error) => return refused_all(reads.len(), &error), + }; + let consumer = &names.consumer; + join_all(reads.iter().map(|&(topic, partition)| { + let read = move |client: Arc<IggyClient>| async move { + let info = client + .get_consumer_offset( + consumer, + &topic.stream_id, + &topic.topic_id, + Some(partition), + ) + .await + .map_err(BridgeError::Iggy)?; + Ok(info.map(|info| info.stored_offset)) + }; + self.read_in_slot(slot_of(topic, partition), deadline, read) + })) + .await + } + + /// Whether Kafka group `group` holds offsets on `topic`. + /// + /// # Errors + /// + /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group` + /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`. + pub async fn holds_offset_group( + &self, + group: &str, + topic: &TopicTarget, + deadline: Instant, + ) -> Result<bool, BridgeError> { + let group_id = OffsetNames::new(group) + .map_err(BridgeError::Iggy)? + .consumer + .id; + let lookup = |client: Arc<IggyClient>| async move { + client + .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id) + .await + .map(|found| found.is_some()) + .map_err(BridgeError::Iggy) + }; + // Any slot will do. Partition 0's keeps the choice fixed. + self.read_in_slot(slot_of(topic, 0), deadline, lookup).await + } + + /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce. + /// + /// # Errors + /// + /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`. + pub async fn list_kafka_topics_until( + &self, + deadline: Instant, + ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> { + let listing = + |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await }; + self.read_in_slot(LISTING_SLOT, deadline, listing).await + } + + /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps + /// the slot until it ends, even after its caller stops waiting, so the next call on the slot + /// cannot overtake it. A call that has not started by `deadline` never starts. + async fn commit_in_slot<F>( + &self, + slot: usize, + deadline: Instant, + call: impl FnOnce(Arc<IggyClient>) -> F, + ) -> Result<(), BridgeError> + where + F: Future<Output = Result<(), BridgeError>> + Send + 'static, + { + let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else { + return Err(BridgeError::Timeout); + }; + let client = match timeout_at(deadline, client.connected(&self.config)).await { + Ok(connected) => connected?, + Err(_elapsed) => return Err(BridgeError::Timeout), + }; + match timeout_at(deadline, spawn_holding(turn, call(client))).await { + Ok(Ok(ended)) => ended, + // A task that ends with no answer leaves the outcome unknown too. + Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout), + } + } + + /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`. + /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot + /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts. + async fn read_in_slot<T, F>( + &self, + slot: usize, + deadline: Instant, + call: impl FnOnce(Arc<IggyClient>) -> F, + ) -> Result<T, BridgeError> + where + F: Future<Output = Result<T, BridgeError>>, + { + let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else { + return Err(BridgeError::Timeout); + }; + let read = async { call(client.connected(&self.config).await?).await }; + client.until(deadline, read).await + } +} + +/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the +/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a +/// read that comes after them waits for them. A read frees its slot by its deadline. +pub(super) struct OffsetPool { + slots: Vec<OffsetSlot>, +} + +struct OffsetSlot { + client: LazyClient, + /// One call at a time. + turn: Arc<Semaphore>, +} + +impl OffsetPool { + pub(super) fn new() -> Self { + Self { + slots: (0..OFFSET_SLOTS) + .map(|_| OffsetSlot { + client: LazyClient::default(), + turn: Arc::new(Semaphore::new(1)), + }) + .collect(), + } + } + + /// Slot `slot`, with its client. `None` if it is not free by `deadline`. + async fn take( + &self, + slot: usize, + deadline: Instant, + ) -> Option<(OwnedSemaphorePermit, &LazyClient)> { + let slot = &self.slots[slot]; + match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await { + Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)), + _ => None, + } + } + + /// Shuts down each client that connected, and returns the first error. A client that fails to + /// shut down stops none of the others. + pub(super) async fn close(&self) -> Result<(), BridgeError> { + let mut closed = Ok(()); + for slot in &self.slots { + let shutdown = slot.client.close().await; + closed = closed.and(shutdown); + } + closed + } +} + +/// The slot of `partition` of `topic`, the same at every call. +fn slot_of(topic: &TopicTarget, partition: u32) -> usize { + let mut hasher = DefaultHasher::new(); + (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher); + usize::from(hasher.finish().to_le_bytes()[0]) % OFFSET_SLOTS +} + +/// Runs `call` in a task that keeps `turn` until `call` ends, even after its caller stops +/// waiting, so the next call on the slot cannot overtake it. +fn spawn_holding<T: Send + 'static>( + turn: OwnedSemaphorePermit, + call: impl Future<Output = Result<T, BridgeError>> + Send + 'static, +) -> oneshot::Receiver<Result<T, BridgeError>> { + let (sender, receiver) = oneshot::channel(); + tokio::spawn(async move { + let ended = timeout(SLOT_LIMIT, call) Review Comment: yeah, that promise was wrong. reworded it in OFFSET_STORAGE.md. reads have their own client now and can't drop the commit one. a late commit just waits behind the stuck one inside the sdk. ########## gateways/kafka/src/protocol/handlers/offset_fetch.rs: ########## @@ -0,0 +1,553 @@ +// 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. + +//! `OffsetFetch` (API key 9). +//! +//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level +//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones. + +use bytes::Bytes; +use futures::future::join_all; +use kafka_protocol::messages::offset_fetch_response::{ + OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions, + OffsetFetchResponseTopic, OffsetFetchResponseTopics, +}; +use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName}; +use kafka_protocol::protocol::StrBytes; +use tokio::time::Instant; + +use crate::bridge::{BridgeError, IggyBridge, KafkaTopicMetadata, TopicTarget}; +use crate::error::Result; +use crate::group::is_valid_group_id; +use crate::protocol::api::{ + API_KEY_OFFSET_FETCH, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS, + ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR, + ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET, + is_supported_version, +}; +use crate::protocol::bounds_guard::validate_offset_fetch_shape; +use crate::protocol::handlers::{ + decode_guarded, encode_message, offset_deadline, offset_refusal_cause, respond_or_close, +}; + +pub const RANGE: ApiVersionRange = ApiVersionRange { + api_key: API_KEY_OFFSET_FETCH, + min_version: 1, + max_version: 9, +}; + +/// From this version a request names several groups and each answer carries its own error. +const FIRST_BATCHED_VERSION: i16 = 8; +/// Below this version there is no group error field. +const FIRST_GROUP_ERROR_VERSION: i16 = 2; +/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition. +const PER_PARTITION_BYTES: usize = 24; + +/// One group a request asks about, across wire versions. +#[derive(Debug, Clone)] +struct GroupQuery { + group_id: StrBytes, + /// `None` asks for every committed offset of the group. + topics: Option<Vec<(TopicName, Vec<i32>)>>, +} + +impl GroupQuery { + /// One query per group the request names: one below v8, the `groups` array from v8. + fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> { + if version >= FIRST_BATCHED_VERSION { + return request + .groups + .iter() + .map(|group| Self { + group_id: group.group_id.0.clone(), + topics: group.topics.as_ref().map(|topics| { + topics + .iter() + .map(|topic| (topic.name.clone(), topic.partition_indexes.clone())) + .collect() + }), + }) + .collect(); + } + vec![Self { + group_id: request.group_id.0.clone(), + topics: request.topics.as_ref().map(|topics| { + topics + .iter() + .map(|topic| (topic.name.clone(), topic.partition_indexes.clone())) + .collect() + }), + }] + } +} + +/// The answer for one group. A group error carries no topics. +#[derive(Debug, Clone)] +struct GroupAnswer { + error: i16, + topics: Vec<TopicAnswer>, +} + +impl GroupAnswer { + const fn error(error: i16) -> Self { + Self { + error, + topics: Vec::new(), + } + } +} + +#[derive(Debug, Clone)] +struct TopicAnswer { + name: TopicName, + partitions: Vec<PartitionAnswer>, +} + +/// `offset` is -1 when nothing is committed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct PartitionAnswer { + index: i32, + offset: i64, + error: i16, +} + +pub async fn handle( + state: &GatewayState, + connection: &ConnectionState, + api_version: i16, + body: Bytes, +) -> HandleOutcome { + // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing + // committed", so an unreadable request closes the connection at every version. + if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) { + return HandleOutcome::Close; + } + let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| { + validate_offset_fetch_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 OffsetFetch request"); + return HandleOutcome::Close; + } + }; + + let queries = GroupQuery::all(api_version, &request); + let deadline = offset_deadline(state, connection).await; + // The groups read at once, so a slow group cannot spend the time of the others. + let answers = join_all( + queries + .iter() + .map(|query| answer(state.bridge.as_deref(), query, deadline)), + ) + .await; + respond_or_close( + encode_response(api_version, &queries, &answers), + "OffsetFetch", + ) +} + +/// One failed read fails the whole group, so a client never reads a lost answer as "nothing +/// committed" and resets its position. +async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer { + if !is_valid_group_id(&query.group_id) { + return GroupAnswer::error(ERROR_INVALID_GROUP_ID); + } + let Some(bridge) = bridge else { + return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS); + }; + // A group the deadline leaves out costs no Iggy call. + if Instant::now() >= deadline { + return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS); + } + // Each Iggy call stops at `deadline` on its own, so none is cut off holding its slot. + let read = match &query.topics { + Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await, + None => read_every_topic(bridge, &query.group_id, deadline).await, + }; + match read { + Ok(topics) => GroupAnswer { + error: ERROR_NONE, + topics, + }, + Err(error) => GroupAnswer::error(refusal(&query.group_id, &error)), + } +} + +/// The group code for a failed read. A -1 is logged, since the client sees no more. +fn refusal(group: &str, error: &BridgeError) -> i16 { + let code = error.to_offset_error_code(); + if code == ERROR_UNKNOWN_SERVER_ERROR { + tracing::error!( + group, + %error, + cause = offset_refusal_cause(error), + "Iggy refused a Kafka offset read" + ); + } else { + tracing::debug!(group, %error, "Iggy refused a Kafka offset read"); + } + code +} + +/// Every requested partition, in request order. A partition with no offset answers -1. The reads +/// run at once across the offset slots. +async fn read_topics( + bridge: &IggyBridge, + group: &str, + topics: &[(TopicName, Vec<i32>)], + deadline: Instant, +) -> core::result::Result<Vec<TopicAnswer>, BridgeError> { + // A name that cannot map to Iggy names no topic, so nothing is committed under it. + let targets: Vec<Option<TopicTarget>> = topics + .iter() + .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok()) + .collect(); + let mut answers: Vec<TopicAnswer> = topics + .iter() + .map(|(name, indexes)| TopicAnswer { + name: name.clone(), + partitions: indexes + .iter() + .map(|&index| PartitionAnswer { + index, + offset: UNKNOWN_OFFSET, + error: ERROR_NONE, + }) + .collect(), + }) + .collect(); + let mut reads = Vec::new(); + // Where each read's offset goes: (topic, partition) in `answers`. + let mut positions = Vec::new(); + for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() { + let Some(target) = target else { + continue; + }; + for (partition_at, &index) in indexes.iter().enumerate() { + if let Ok(partition) = u32::try_from(index) { + reads.push((target, partition)); + positions.push((topic_at, partition_at)); + } + } + } + let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await; + for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) { + answers[topic_at].partitions[partition_at].offset = committed(read)?; + } + Ok(answers) +} + +/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only +/// the partitions with an offset, on every topic that holds the group. The topics read at once. +async fn read_every_topic( + bridge: &IggyBridge, + group: &str, + deadline: Instant, +) -> core::result::Result<Vec<TopicAnswer>, BridgeError> { + let topics = bridge.list_kafka_topics_until(deadline).await?; + let answers = join_all( Review Comment: added OffsetCalls, one per kafka request. it allows one waiting call per slot, other requests get in after one of its calls. commits use it too. ########## gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs: ########## @@ -0,0 +1,574 @@ +// 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. + +//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`. + +use std::future::Future; +use std::hash::{DefaultHasher, Hash, Hasher}; +use std::sync::Arc; + +use futures::future::join_all; +use iggy::prelude::{ + Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError, +}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; +use tokio::time::{Instant, timeout, timeout_at}; +use tracing::info; + +use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget}; +use crate::bridge::error::BridgeError; + +/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group +/// apart from a native Iggy group of the same name. +pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg."; + +/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait +/// behind a Produce send. +const OFFSET_SLOTS: usize = 4; + +/// The slot of a call that belongs to no partition. Any slot will do. +const LISTING_SLOT: usize = 0; + +impl IggyBridge { + /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group + /// `group`. One result per commit, in order. + /// + /// A negative offset deletes the key instead. Kafka consumers read any negative offset as + /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A + /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already + /// applied counts as done. + /// + /// Each commit starts once its slot is free. Commits on one slot start in the order given. + /// + /// # Errors + /// + /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group, + /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not + /// started by then never starts. + pub async fn commit_group_offsets( + &self, + group: &str, + commits: &[(&TopicTarget, u32, i64)], + deadline: Instant, + ) -> Vec<Result<(), BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => Arc::new(names), + Err(error) => return refused_all(commits.len(), &error), + }; + // `join_all` polls each call once, in order, before any again, so calls on one slot + // queue for it in order. + join_all(commits.iter().map(|&(topic, partition, offset)| { + let (names, stream_id, topic_id) = ( + Arc::clone(&names), + topic.stream_id.clone(), + topic.topic_id.clone(), + ); + let commit = move |client: Arc<IggyClient>| async move { + commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await + }; + self.commit_in_slot(slot_of(topic, partition), deadline, commit) + })) + .await + } + + /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in + /// order. `None` when there is no key or no group. The reads run at once. + /// + /// # Errors + /// + /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too. + pub async fn fetch_group_offsets( + &self, + group: &str, + reads: &[(&TopicTarget, u32)], + deadline: Instant, + ) -> Vec<Result<Option<u64>, BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => names, + Err(error) => return refused_all(reads.len(), &error), + }; + let consumer = &names.consumer; + join_all(reads.iter().map(|&(topic, partition)| { + let read = move |client: Arc<IggyClient>| async move { + let info = client + .get_consumer_offset( + consumer, + &topic.stream_id, + &topic.topic_id, + Some(partition), + ) + .await + .map_err(BridgeError::Iggy)?; + Ok(info.map(|info| info.stored_offset)) + }; + self.read_in_slot(slot_of(topic, partition), deadline, read) + })) + .await + } + + /// Whether Kafka group `group` holds offsets on `topic`. + /// + /// # Errors + /// + /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group` + /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`. + pub async fn holds_offset_group( + &self, + group: &str, + topic: &TopicTarget, + deadline: Instant, + ) -> Result<bool, BridgeError> { + let group_id = OffsetNames::new(group) + .map_err(BridgeError::Iggy)? + .consumer + .id; + let lookup = |client: Arc<IggyClient>| async move { + client + .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id) + .await + .map(|found| found.is_some()) + .map_err(BridgeError::Iggy) + }; + // Any slot will do. Partition 0's keeps the choice fixed. + self.read_in_slot(slot_of(topic, 0), deadline, lookup).await + } + + /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce. + /// + /// # Errors + /// + /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`. + pub async fn list_kafka_topics_until( + &self, + deadline: Instant, + ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> { + let listing = + |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await }; + self.read_in_slot(LISTING_SLOT, deadline, listing).await + } + + /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps + /// the slot until it ends, even after its caller stops waiting, so the next call on the slot + /// cannot overtake it. A call that has not started by `deadline` never starts. + async fn commit_in_slot<F>( + &self, + slot: usize, + deadline: Instant, + call: impl FnOnce(Arc<IggyClient>) -> F, + ) -> Result<(), BridgeError> + where + F: Future<Output = Result<(), BridgeError>> + Send + 'static, + { + let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else { + return Err(BridgeError::Timeout); + }; + let client = match timeout_at(deadline, client.connected(&self.config)).await { + Ok(connected) => connected?, + Err(_elapsed) => return Err(BridgeError::Timeout), + }; + match timeout_at(deadline, spawn_holding(turn, call(client))).await { + Ok(Ok(ended)) => ended, + // A task that ends with no answer leaves the outcome unknown too. + Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout), + } + } + + /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`. + /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot + /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts. + async fn read_in_slot<T, F>( + &self, + slot: usize, + deadline: Instant, + call: impl FnOnce(Arc<IggyClient>) -> F, + ) -> Result<T, BridgeError> + where + F: Future<Output = Result<T, BridgeError>>, + { + let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else { + return Err(BridgeError::Timeout); + }; + let read = async { call(client.connected(&self.config).await?).await }; + client.until(deadline, read).await + } +} + +/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the +/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a +/// read that comes after them waits for them. A read frees its slot by its deadline. +pub(super) struct OffsetPool { + slots: Vec<OffsetSlot>, +} + +struct OffsetSlot { + client: LazyClient, + /// One call at a time. + turn: Arc<Semaphore>, +} + +impl OffsetPool { + pub(super) fn new() -> Self { + Self { + slots: (0..OFFSET_SLOTS) + .map(|_| OffsetSlot { + client: LazyClient::default(), + turn: Arc::new(Semaphore::new(1)), + }) + .collect(), + } + } + + /// Slot `slot`, with its client. `None` if it is not free by `deadline`. + async fn take( + &self, + slot: usize, + deadline: Instant, + ) -> Option<(OwnedSemaphorePermit, &LazyClient)> { + let slot = &self.slots[slot]; + match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await { + Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)), + _ => None, + } + } + + /// Shuts down each client that connected, and returns the first error. A client that fails to + /// shut down stops none of the others. + pub(super) async fn close(&self) -> Result<(), BridgeError> { + let mut closed = Ok(()); + for slot in &self.slots { + let shutdown = slot.client.close().await; + closed = closed.and(shutdown); + } + closed + } +} + +/// The slot of `partition` of `topic`, the same at every call. +fn slot_of(topic: &TopicTarget, partition: u32) -> usize { + let mut hasher = DefaultHasher::new(); + (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher); Review Comment: done, the hash is stream + topic now and the partition index is added on top. 4 partitions get 4 slots. ########## gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs: ########## @@ -0,0 +1,574 @@ +// 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. + +//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`. + +use std::future::Future; +use std::hash::{DefaultHasher, Hash, Hasher}; +use std::sync::Arc; + +use futures::future::join_all; +use iggy::prelude::{ + Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError, +}; +use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; +use tokio::time::{Instant, timeout, timeout_at}; +use tracing::info; + +use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget}; +use crate::bridge::error::BridgeError; + +/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group +/// apart from a native Iggy group of the same name. +pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg."; + +/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait +/// behind a Produce send. +const OFFSET_SLOTS: usize = 4; + +/// The slot of a call that belongs to no partition. Any slot will do. +const LISTING_SLOT: usize = 0; + +impl IggyBridge { + /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group + /// `group`. One result per commit, in order. + /// + /// A negative offset deletes the key instead. Kafka consumers read any negative offset as + /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A + /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already + /// applied counts as done. + /// + /// Each commit starts once its slot is free. Commits on one slot start in the order given. + /// + /// # Errors + /// + /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group, + /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not + /// started by then never starts. + pub async fn commit_group_offsets( + &self, + group: &str, + commits: &[(&TopicTarget, u32, i64)], + deadline: Instant, + ) -> Vec<Result<(), BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => Arc::new(names), + Err(error) => return refused_all(commits.len(), &error), + }; + // `join_all` polls each call once, in order, before any again, so calls on one slot + // queue for it in order. + join_all(commits.iter().map(|&(topic, partition, offset)| { + let (names, stream_id, topic_id) = ( + Arc::clone(&names), + topic.stream_id.clone(), + topic.topic_id.clone(), + ); + let commit = move |client: Arc<IggyClient>| async move { + commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await + }; + self.commit_in_slot(slot_of(topic, partition), deadline, commit) + })) + .await + } + + /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in + /// order. `None` when there is no key or no group. The reads run at once. + /// + /// # Errors + /// + /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too. + pub async fn fetch_group_offsets( + &self, + group: &str, + reads: &[(&TopicTarget, u32)], + deadline: Instant, + ) -> Vec<Result<Option<u64>, BridgeError>> { + let names = match OffsetNames::new(group) { + Ok(names) => names, + Err(error) => return refused_all(reads.len(), &error), + }; + let consumer = &names.consumer; + join_all(reads.iter().map(|&(topic, partition)| { + let read = move |client: Arc<IggyClient>| async move { + let info = client + .get_consumer_offset( + consumer, + &topic.stream_id, + &topic.topic_id, + Some(partition), + ) + .await + .map_err(BridgeError::Iggy)?; + Ok(info.map(|info| info.stored_offset)) + }; + self.read_in_slot(slot_of(topic, partition), deadline, read) + })) + .await + } + + /// Whether Kafka group `group` holds offsets on `topic`. + /// + /// # Errors + /// + /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group` + /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`. + pub async fn holds_offset_group( + &self, + group: &str, + topic: &TopicTarget, + deadline: Instant, + ) -> Result<bool, BridgeError> { + let group_id = OffsetNames::new(group) + .map_err(BridgeError::Iggy)? + .consumer + .id; + let lookup = |client: Arc<IggyClient>| async move { + client + .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id) + .await + .map(|found| found.is_some()) + .map_err(BridgeError::Iggy) + }; + // Any slot will do. Partition 0's keeps the choice fixed. + self.read_in_slot(slot_of(topic, 0), deadline, lookup).await + } + + /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce. + /// + /// # Errors + /// + /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`. + pub async fn list_kafka_topics_until( + &self, + deadline: Instant, + ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> { + let listing = + |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await }; + self.read_in_slot(LISTING_SLOT, deadline, listing).await + } + + /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps + /// the slot until it ends, even after its caller stops waiting, so the next call on the slot + /// cannot overtake it. A call that has not started by `deadline` never starts. + async fn commit_in_slot<F>( + &self, + slot: usize, + deadline: Instant, + call: impl FnOnce(Arc<IggyClient>) -> F, + ) -> Result<(), BridgeError> + where + F: Future<Output = Result<(), BridgeError>> + Send + 'static, + { + let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else { + return Err(BridgeError::Timeout); + }; + let client = match timeout_at(deadline, client.connected(&self.config)).await { + Ok(connected) => connected?, + Err(_elapsed) => return Err(BridgeError::Timeout), + }; + match timeout_at(deadline, spawn_holding(turn, call(client))).await { + Ok(Ok(ended)) => ended, + // A task that ends with no answer leaves the outcome unknown too. + Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout), + } + } + + /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`. + /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot + /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts. + async fn read_in_slot<T, F>( + &self, + slot: usize, + deadline: Instant, + call: impl FnOnce(Arc<IggyClient>) -> F, + ) -> Result<T, BridgeError> + where + F: Future<Output = Result<T, BridgeError>>, + { + let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else { + return Err(BridgeError::Timeout); + }; + let read = async { call(client.connected(&self.config).await?).await }; + client.until(deadline, read).await + } +} + +/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the +/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a +/// read that comes after them waits for them. A read frees its slot by its deadline. +pub(super) struct OffsetPool { + slots: Vec<OffsetSlot>, +} + +struct OffsetSlot { + client: LazyClient, + /// One call at a time. + turn: Arc<Semaphore>, +} + +impl OffsetPool { + pub(super) fn new() -> Self { + Self { + slots: (0..OFFSET_SLOTS) + .map(|_| OffsetSlot { + client: LazyClient::default(), + turn: Arc::new(Semaphore::new(1)), + }) + .collect(), + } + } + + /// Slot `slot`, with its client. `None` if it is not free by `deadline`. + async fn take( + &self, + slot: usize, + deadline: Instant, + ) -> Option<(OwnedSemaphorePermit, &LazyClient)> { + let slot = &self.slots[slot]; + match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await { Review Comment: added permit_by in the bridge mod, produce uses it too. ########## core/server/src/dispatch/partition.rs: ########## @@ -724,12 +725,32 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>( stored: Some(stored_offset), current_offset, }) => build_consumer_offset_body(partition_id, current_offset, stored_offset), + Some(PartitionReadReply::ConsumerOffset { stored: None, .. }) => Bytes::new(), Some(PartitionReadReply::Rejected(error)) => { send_non_replicated_deny(shard, request, transport_client_id, error.as_code()) .await; return; } - _ => Bytes::new(), + // No reply, or an owner that does not hold the partition yet, + // says nothing about the offset, so it must not read as "no + // stored offset". A read moves no progress, so the client may + // retry it. + reply => { + warn!( Review Comment: NotFound logs at debug now, timeout stays at warn. ########## gateways/kafka/src/protocol/handlers/offset_fetch.rs: ########## @@ -0,0 +1,553 @@ +// 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. + +//! `OffsetFetch` (API key 9). +//! +//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level +//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones. + +use bytes::Bytes; +use futures::future::join_all; +use kafka_protocol::messages::offset_fetch_response::{ + OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions, + OffsetFetchResponseTopic, OffsetFetchResponseTopics, +}; +use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName}; +use kafka_protocol::protocol::StrBytes; +use tokio::time::Instant; + +use crate::bridge::{BridgeError, IggyBridge, KafkaTopicMetadata, TopicTarget}; +use crate::error::Result; +use crate::group::is_valid_group_id; +use crate::protocol::api::{ + API_KEY_OFFSET_FETCH, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS, + ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR, + ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET, + is_supported_version, +}; +use crate::protocol::bounds_guard::validate_offset_fetch_shape; +use crate::protocol::handlers::{ + decode_guarded, encode_message, offset_deadline, offset_refusal_cause, respond_or_close, +}; + +pub const RANGE: ApiVersionRange = ApiVersionRange { + api_key: API_KEY_OFFSET_FETCH, + min_version: 1, + max_version: 9, +}; + +/// From this version a request names several groups and each answer carries its own error. +const FIRST_BATCHED_VERSION: i16 = 8; +/// Below this version there is no group error field. +const FIRST_GROUP_ERROR_VERSION: i16 = 2; +/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition. +const PER_PARTITION_BYTES: usize = 24; + +/// One group a request asks about, across wire versions. +#[derive(Debug, Clone)] +struct GroupQuery { + group_id: StrBytes, + /// `None` asks for every committed offset of the group. + topics: Option<Vec<(TopicName, Vec<i32>)>>, +} + +impl GroupQuery { + /// One query per group the request names: one below v8, the `groups` array from v8. + fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> { + if version >= FIRST_BATCHED_VERSION { + return request + .groups + .iter() + .map(|group| Self { + group_id: group.group_id.0.clone(), + topics: group.topics.as_ref().map(|topics| { + topics + .iter() + .map(|topic| (topic.name.clone(), topic.partition_indexes.clone())) + .collect() + }), + }) + .collect(); + } + vec![Self { + group_id: request.group_id.0.clone(), + topics: request.topics.as_ref().map(|topics| { + topics + .iter() + .map(|topic| (topic.name.clone(), topic.partition_indexes.clone())) + .collect() + }), + }] + } +} + +/// The answer for one group. A group error carries no topics. +#[derive(Debug, Clone)] +struct GroupAnswer { + error: i16, + topics: Vec<TopicAnswer>, +} + +impl GroupAnswer { + const fn error(error: i16) -> Self { + Self { + error, + topics: Vec::new(), + } + } +} + +#[derive(Debug, Clone)] +struct TopicAnswer { + name: TopicName, + partitions: Vec<PartitionAnswer>, +} + +/// `offset` is -1 when nothing is committed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct PartitionAnswer { + index: i32, + offset: i64, + error: i16, +} + +pub async fn handle( + state: &GatewayState, + connection: &ConnectionState, + api_version: i16, + body: Bytes, +) -> HandleOutcome { + // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing + // committed", so an unreadable request closes the connection at every version. + if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) { + return HandleOutcome::Close; + } + let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| { + validate_offset_fetch_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 OffsetFetch request"); + return HandleOutcome::Close; + } + }; + + let queries = GroupQuery::all(api_version, &request); + let deadline = offset_deadline(state, connection).await; + // The groups read at once, so a slow group cannot spend the time of the others. + let answers = join_all( + queries + .iter() + .map(|query| answer(state.bridge.as_deref(), query, deadline)), + ) + .await; + respond_or_close( + encode_response(api_version, &queries, &answers), + "OffsetFetch", + ) +} + +/// One failed read fails the whole group, so a client never reads a lost answer as "nothing +/// committed" and resets its position. +async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer { + if !is_valid_group_id(&query.group_id) { + return GroupAnswer::error(ERROR_INVALID_GROUP_ID); + } + let Some(bridge) = bridge else { + return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS); + }; + // A group the deadline leaves out costs no Iggy call. + if Instant::now() >= deadline { + return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS); + } + // Each Iggy call stops at `deadline` on its own, so none is cut off holding its slot. + let read = match &query.topics { + Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await, + None => read_every_topic(bridge, &query.group_id, deadline).await, + }; + match read { + Ok(topics) => GroupAnswer { + error: ERROR_NONE, + topics, + }, + Err(error) => GroupAnswer::error(refusal(&query.group_id, &error)), + } +} + +/// The group code for a failed read. A -1 is logged, since the client sees no more. +fn refusal(group: &str, error: &BridgeError) -> i16 { + let code = error.to_offset_error_code(); + if code == ERROR_UNKNOWN_SERVER_ERROR { + tracing::error!( + group, + %error, + cause = offset_refusal_cause(error), + "Iggy refused a Kafka offset read" + ); + } else { + tracing::debug!(group, %error, "Iggy refused a Kafka offset read"); + } + code +} + +/// Every requested partition, in request order. A partition with no offset answers -1. The reads +/// run at once across the offset slots. +async fn read_topics( + bridge: &IggyBridge, + group: &str, + topics: &[(TopicName, Vec<i32>)], + deadline: Instant, +) -> core::result::Result<Vec<TopicAnswer>, BridgeError> { + // A name that cannot map to Iggy names no topic, so nothing is committed under it. + let targets: Vec<Option<TopicTarget>> = topics + .iter() + .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok()) + .collect(); + let mut answers: Vec<TopicAnswer> = topics + .iter() + .map(|(name, indexes)| TopicAnswer { + name: name.clone(), + partitions: indexes + .iter() + .map(|&index| PartitionAnswer { + index, + offset: UNKNOWN_OFFSET, + error: ERROR_NONE, + }) + .collect(), + }) + .collect(); + let mut reads = Vec::new(); + // Where each read's offset goes: (topic, partition) in `answers`. + let mut positions = Vec::new(); + for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() { + let Some(target) = target else { + continue; + }; + for (partition_at, &index) in indexes.iter().enumerate() { + if let Ok(partition) = u32::try_from(index) { + reads.push((target, partition)); + positions.push((topic_at, partition_at)); + } + } + } + let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await; + for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) { + answers[topic_at].partitions[partition_at].offset = committed(read)?; + } + Ok(answers) +} + +/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only +/// the partitions with an offset, on every topic that holds the group. The topics read at once. +async fn read_every_topic( + bridge: &IggyBridge, + group: &str, + deadline: Instant, +) -> core::result::Result<Vec<TopicAnswer>, BridgeError> { + let topics = bridge.list_kafka_topics_until(deadline).await?; + let answers = join_all( + topics + .into_iter() + .map(|topic| read_listed_topic(bridge, group, topic, deadline)), + ) + .await; + answers + .into_iter() + .filter_map(core::result::Result::transpose) + .collect() +} + +/// The partitions of `topic` with an offset of `group`. `None` when there are none. +async fn read_listed_topic( + bridge: &IggyBridge, + group: &str, + topic: KafkaTopicMetadata, + deadline: Instant, +) -> core::result::Result<Option<TopicAnswer>, BridgeError> { + let target = bridge.topic_target(&topic.kafka_topic)?; + match bridge.holds_offset_group(group, &target, deadline).await { + Ok(true) => {} + Ok(false) => return Ok(None), + // A topic deleted since the listing holds no offsets. + Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => { + return Ok(None); + } + Err(error) => return Err(error), + } + let reads: Vec<_> = (0..topic.partitions_count) + .map(|partition| (&target, partition)) + .collect(); + let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await; + let mut partitions = Vec::new(); + for (partition, read) in (0..topic.partitions_count).zip(offsets) { + let offset = committed(read)?; + if offset != UNKNOWN_OFFSET { + partitions.push(PartitionAnswer { + index: i32::try_from(partition).unwrap_or(i32::MAX), + offset, + error: ERROR_NONE, + }); + } + } + Ok((!partitions.is_empty()).then(|| TopicAnswer { + name: TopicName(StrBytes::from_string(topic.kafka_topic)), + partitions, + })) +} + +/// A read as Kafka answers it: -1 for no offset and for a topic or partition that does not exist. +fn committed( + read: core::result::Result<Option<u64>, BridgeError>, +) -> core::result::Result<i64, BridgeError> { + match read { + Ok(Some(offset)) => Ok(i64::try_from(offset).unwrap_or(i64::MAX)), + Ok(None) => Ok(UNKNOWN_OFFSET), + Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => { + Ok(UNKNOWN_OFFSET) + } + Err(error) => Err(error), + } +} + +/// One entry per query, in request order. Below v8 there is exactly one. +/// +/// # Errors +/// +/// Returns an error when `kafka_protocol` cannot encode the response at `version`. +fn encode_response(version: i16, queries: &[GroupQuery], answers: &[GroupAnswer]) -> Result<Bytes> { + let partitions: usize = answers + .iter() + .flat_map(|answer| &answer.topics) + .map(|topic| topic.name.0.len() + topic.partitions.len() * PER_PARTITION_BYTES) + .sum(); + let groups: usize = queries.iter().map(|query| query.group_id.len() + 8).sum(); + let capacity = 16 + groups + partitions; + + if version >= FIRST_BATCHED_VERSION { + let groups = queries + .iter() + .zip(answers) + .map(|(query, answer)| { + OffsetFetchResponseGroup::default() + .with_group_id(GroupId(query.group_id.clone())) + .with_error_code(answer.error) + .with_topics( + answer + .topics + .iter() + .map(|topic| { + OffsetFetchResponseTopics::default() + .with_name(topic.name.clone()) + .with_partitions( + topic + .partitions + .iter() + .map(|partition| { + OffsetFetchResponsePartitions::default() + .with_partition_index(partition.index) + .with_committed_offset(partition.offset) + .with_error_code(partition.error) + }) + .collect(), + ) + }) + .collect(), + ) + }) + .collect(); + return encode_message( + &OffsetFetchResponse::default().with_groups(groups), + version, + capacity, + ); + } + + let (Some(query), Some(answer)) = (queries.first(), answers.first()) else { + return encode_message(&OffsetFetchResponse::default(), version, capacity); + }; + let topics = if version < FIRST_GROUP_ERROR_VERSION && answer.error != ERROR_NONE { + // No group error field yet: the error goes on every partition asked for. + query + .topics + .iter() + .flatten() + .map(|(name, indexes)| TopicAnswer { + name: name.clone(), + partitions: indexes + .iter() + .map(|&index| PartitionAnswer { + index, + offset: UNKNOWN_OFFSET, + error: answer.error, + }) + .collect(), + }) + .collect() + } else { + answer.topics.clone() Review Comment: encode_response takes the answers by value now, no clone. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
