hubcio commented on code in PR #4415: URL: https://github.com/apache/iggy/pull/4415#discussion_r4209806933
########## 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: warning: since commits hold the slot for ordering, one stuck commit blocks every group's offset reads on that slot for up to `SLOT_LIMIT`, and they all get 14. give reads their own client per slot so only commits wait on commits. ########## 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: warning: a null-topic fetch queues one call per topic and partition on the offset slots, so one admin `listConsumerGroupOffsets` can stall all other offset calls for up to 3 s. cap each request at one queued call per slot. ########## 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: nit: hashing the partition index puts two partitions of most 4-partition topics on one slot, so their commits run one after another. hash only stream and topic, then add the partition index. ########## gateways/kafka/README.md: ########## @@ -1,11 +1,13 @@ # Kafka gateway (`iggy-gateway-kafka`) -Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata and CreateTopics use Iggy. InitProducerId and consumer group coordination work with or without one. - -> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata and -> CreateTopics use Iggy. With the bridge off (the default), those five are stubs: Produce, Fetch and -> ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6), CreateTopics answers `NOT_CONTROLLER` -> (41), and Metadata reports every requested topic unknown. **CreateTopics runs as the bridge's own +Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, OffsetCommit and OffsetFetch use Iggy. InitProducerId and consumer group coordination work with or without one. + +> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata, +> CreateTopics, OffsetCommit and OffsetFetch use Iggy. With the bridge off (the default), those +> seven are stubs: Produce, Fetch and ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6), +> CreateTopics answers `NOT_CONTROLLER` (41), OffsetCommit and OffsetFetch answer retriable +> `COORDINATOR_LOAD_IN_PROGRESS` (14), and Metadata reports every requested topic unknown. +> **CreateTopics runs as the bridge's own Review Comment: nit: `OffsetCommit` creates iggy resources too - a `kafka.cg.<group>` consumer group per group id and topic, and nothing deletes them. name it next to `CreateTopics` and point at the cleanup commands in `OFFSET_STORAGE.md`. ########## 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: nit: this repeats the permit and deadline guard from `send_in_slot` in `produce.rs`. a small `permit_by(turn, deadline)` helper keeps one copy of the deadline rule. ########## 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 => { Review Comment: nit: the body covers this server contract change, but the squash title will only say `feat(gateways)`. split it into a `fix(server)` PR or name it in the title. ########## gateways/kafka/src/group/state.rs: ########## @@ -1112,6 +1119,83 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) -> } } +/// Kafka's `ClassicGroup.validateOffsetCommit`. Never creates a group. +/// +/// A group with no members is Kafka's `Empty`: a negative generation commits without a member. A +/// gateway restart empties every group, so a member's next commit gets `UNKNOWN_MEMBER_ID` and it +/// rejoins. Kafka's v9 answer, `GROUP_ID_NOT_FOUND`, is fatal in the Java consumer. +pub fn commit_step(groups: &mut Groups, request: &CommitRequest, now: Instant) -> i16 { + if !is_valid_group_id(&request.group_id) { + return ERROR_INVALID_GROUP_ID; + } + let empty = || { + if request.generation_id < 0 { + ERROR_NONE + } else { + ERROR_UNKNOWN_MEMBER_ID + } + }; + if !tick_group(groups, &request.group_id, now) { + return empty(); + } + let Some(group) = groups.get_mut(&request.group_id) else { + return empty(); + }; + if group.members.is_empty() { + return empty(); + } + if request.generation_id < 0 + && request.member_id.is_empty() + && request.group_instance_id.is_none() + { + return ERROR_UNKNOWN_MEMBER_ID; + } + if let Some(instance_id) = &request.group_instance_id { + let holds = |member: &Member| member.group_instance_id.as_ref() == Some(instance_id); + if !group.members.values().any(holds) { + return ERROR_UNKNOWN_MEMBER_ID; + } + if !group.members.get(&request.member_id).is_some_and(holds) { + return ERROR_FENCED_INSTANCE_ID; + } + } + let (generation_id, phase) = (group.generation_id, group.phase); + let Some(member) = group.members.get_mut(&request.member_id) else { + return ERROR_UNKNOWN_MEMBER_ID; + }; + if request.generation_id != generation_id { + return ERROR_ILLEGAL_GENERATION; + } + // A member commits right before it rejoins, so `PreparingRebalance` accepts. + if phase == Phase::CompletingRebalance { + return ERROR_REBALANCE_IN_PROGRESS; + } + member.session_deadline = now + member.session_timeout; + ERROR_NONE +} + +/// How long an offset call may hold the connection that `member` heartbeats on: `hold`, and at +/// most half of what is left of that member's session, so a heartbeat that waits behind the call +/// still lands in time. A member that is not there, or whose session has ended, leaves `hold`. +pub fn offset_hold( Review Comment: simplification: the only production caller already returns on `None` before it calls this, so take `&GroupMember` and drop the `None` branch and its test assertion. ########## 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: nit: below v8 this clones every topic answer just to read it. take the answers by value in `encode_response` and move `topics` out - no clone needed. ########## gateways/kafka/src/group/state.rs: ########## @@ -49,6 +49,13 @@ use crate::protocol::api::{ /// group id this gateway admits must leave room for that prefix (`docs/OFFSET_STORAGE.md`). pub const MAX_GROUP_ID_BYTES: usize = 246; +/// Whether every group API admits `group_id`. Kafka refuses an empty one, and a longer one than Review Comment: nit: kafka refuses an empty group id only for the membership APIs - `commitOffsets` and `fetchOffsets` in `GroupCoordinatorService` accept "" for backwards compatibility. fix the doc, or accept "" for keys 8 and 9. ########## 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: warning: the permit isn't held until the call ends - this frees it at `SLOT_LIMIT` while the SDK attempt still runs, and a timed-out read drops the client, so the next commit opens a new session. reword the order promise in `OFFSET_STORAGE.md`. ########## gateways/kafka/docs/CONSUMER_GROUPS.md: ########## @@ -168,17 +168,15 @@ lock, and stock clients list one or two assignors. ## What a real consumer still cannot do Review Comment: nit: the bridge paragraph below lists things a consumer can do now, so this heading no longer fits. rename it to "what a real consumer can and cannot do". ########## 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> { Review Comment: simplification: `OffsetPool::close` is a line-for-line copy of `FetchPool::close`. one shared helper over the `LazyClient` list covers both. ########## 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: nit: the SDK retries `TransientNotAccepted` every 50 ms, so a read for a partition the owner shard doesn't hold yet logs this `warn!` on every retry. log `NotFound` at `debug!` and keep `warn!` for the timeout. ########## 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>( Review Comment: simplification: this repeats `in_slot` plus the `SLOT_LIMIT` wrapper from `send_in_slot`. pass `timeout(SLOT_LIMIT, call(client))` to `in_slot` like produce does and drop `spawn_holding` - commits keep their client, so the `in_slot` doc needs a line on that. -- 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]
