ryerraguntla commented on code in PR #4415:
URL: https://github.com/apache/iggy/pull/4415#discussion_r4202258240
##########
core/server/src/dispatch/partition.rs:
##########
@@ -729,6 +730,24 @@ pub(in crate::dispatch) async fn
handle_get_consumer_offset<B, MJ, S, SB>(
.await;
return;
}
+ // A missing reply 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.
+ None => {
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ "consumer offset read reply unavailable; replying not
accepted"
+ );
+ send_non_replicated_deny(
+ shard,
+ request,
+ transport_client_id,
+ IggyError::TransientNotAccepted.as_code(),
+ )
+ .await;
+ return;
+ }
_ => Bytes::new(),
Review Comment:
NotFound becomes an empty success body, so OffsetFetch answers -1 and
auto.offset.reset=latest skips. Fix: deny NotFound with TransientNotAccepted.
Empty body only for ConsumerOffset
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// 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 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, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, 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_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, 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 = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
Review Comment:
v8 groups share one deadline in series. Tail groups get retriable 14 and
never call Iggy. Java retries 14, so a head group that eats the 3s budget
livelocks the tail. Fix: start the group reads together under the same deadline.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,449 @@
+// 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 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, 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;
+
+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.
+ ///
+ /// # 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),
+ };
+ let mut started = Vec::with_capacity(commits.len());
+ for &(topic, partition, offset) in commits {
+ 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
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, commit)
+ .await,
Review Comment:
partition offset calls start one at a time, so a busy slot blocks partitions
whose slots are free. Fix: start every partition, then wait in finish. The slot
semaphore already keeps order.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// 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 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, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, 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_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, 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 = Instant::now() + state.groups.offset_deadline();
Review Comment:
the handler holds the connection until the offset deadline, so the heartbeat
waits. A fetch that starts with 3s or less left burns the rest of the session
and the member expires. Fix: cap the hold by the time left on that member
session, or read the heartbeat while the offset call runs.
##########
gateways/kafka/README.md:
##########
Review Comment:
[nit] - the error map records the Fetch fold (7→6) and omits the
OffsetCommit/OffsetFetch fold (6 and 7 → 14 via to_offset_error_code). Fix: add
that fold beside the Fetch sentence.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,449 @@
+// 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 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, 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;
+
+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.
+ ///
+ /// # 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),
+ };
+ let mut started = Vec::with_capacity(commits.len());
+ for &(topic, partition, offset) in commits {
+ 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
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, commit)
+ .await,
+ );
+ }
+ finish_all(started, deadline).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.
+ ///
+ /// # 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) => Arc::new(names),
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let mut started = Vec::with_capacity(reads.len());
+ for &(topic, partition) in reads {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(&names.consumer, &stream_id,
&topic_id, Some(partition))
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, read)
+ .await,
+ );
+ }
+ finish_all(started, deadline).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 (stream_id, topic_id) = (topic.stream_id.clone(),
topic.topic_id.clone());
+ let lookup = move |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&stream_id, &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.start_offset_call(topic, 0, deadline, lookup)
+ .await
+ .finish(deadline)
+ .await
+ }
+
+ /// Starts `call` with the client of the slot of `partition` of `topic`,
and returns without
+ /// waiting for it to end. A call that has not started by `deadline` never
starts.
+ async fn start_offset_call<T, F>(
+ &self,
+ topic: &TopicTarget,
+ partition: u32,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Started<T>
+ where
+ T: Send + 'static,
+ F: Future<Output = Result<T, BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(topic, partition,
deadline).await else {
+ return Started(Err(BridgeError::Timeout));
+ };
+ match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(Ok(client)) => Started(Ok(spawn_holding(turn, call(client)))),
+ Ok(Err(error)) => Started(Err(error)),
+ Err(_elapsed) => Started(Err(BridgeError::Timeout)),
+ }
+ }
+}
+
+/// The offset slots. A partition always takes the same slot, and a call keeps
its slot until the
+/// SDK lets go of it, so the calls of one partition reach Iggy in the order
they started.
+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(),
+ }
+ }
+
+ /// The slot of `partition` of `topic`, with its client. `None` if it is
not free by
+ /// `deadline`.
+ async fn take(
+ &self,
+ topic: &TopicTarget,
+ partition: u32,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot_of(topic, partition)];
+ 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)
+ .await
+ .unwrap_or(Err(BridgeError::Timeout));
+ // The caller may have stopped waiting.
+ let _ = sender.send(ended);
+ drop(turn);
+ });
+ receiver
+}
+
+/// An offset call that keeps its slot, or why it never started.
+struct Started<T>(Result<oneshot::Receiver<Result<T, BridgeError>>,
BridgeError>);
+
+impl<T> Started<T> {
+ /// The call's result. [`BridgeError::Timeout`] if it does not end by
`deadline`.
+ async fn finish(self, deadline: Instant) -> Result<T, BridgeError> {
+ match timeout_at(deadline, self.0?).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
Review Comment:
cross-gateway only. finish returns Timeout while spawn_holding still stores
or deletes. Another gateway can land a newer commit first, and this write
rewinds or deletes it. A deleted key fetches as -1. Fix: store a commit epoch
and reject a stale write. Same-gateway rewind does not happen: the slot permit
is held until the call ends.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// 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 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, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, 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_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, 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 = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
+ answers.push(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);
+ }
+ let read = async {
+ 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 timeout_at(deadline, read).await {
Review Comment:
one timeout_at wraps the whole group read. Topics already read are dropped,
and the slot stays held up to 45s, so a Java retry of 14 cannot enter. Fix:
per-item deadlines, keep finished topics, release the slot when the fetch is
dropped.
##########
gateways/kafka/docs/kafka_api_keys_reference.md:
##########
Review Comment:
[nit] - Group 3 still marks OffsetCommit and OffsetFetch as required stubs.
Line 271 still counts keys 8–14 as stubs. Line 313 says both are supported.
Fix: mark them implemented and drop them from the stub count.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// 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 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, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, 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_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, 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 = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
+ answers.push(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);
+ }
+ let read = async {
+ 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 timeout_at(deadline, read).await {
+ Ok(Ok(topics)) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Ok(Err(error)) => GroupAnswer::error(refusal(&query.group_id, &error)),
+ Err(_elapsed) =>
GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ }
+}
+
+/// 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.
+async fn read_every_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let mut answers = Vec::new();
+ for topic in bridge.list_kafka_topics().await? {
Review Comment:
a null topic list uses the lockstep Produce client, so OffsetFetch blocks
Produce. Docs say offset calls do not. Fix: list topics on an offset-slot
client or probe_client.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -1112,6 +1119,61 @@ 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 {
Review Comment:
generation -1 on a gateway with no local group returns ERROR_NONE and then
stores on the shared kafka.cg.<group>, overwriting another gateway's live
offsets. Fix: refuse generation < 0 unless this process has the group, or fence
the stale write.
##########
gateways/kafka/src/protocol/handlers/offset_commit.rs:
##########
@@ -0,0 +1,314 @@
+// 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.
+
+//! `OffsetCommit` (API key 8).
+//!
+//! Checks the committer against its group, then stores each offset in Iggy
+//! (`docs/OFFSET_STORAGE.md`). The response has no top-level error, so a
group answer goes on
+//! every partition.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_commit_response::{
+ OffsetCommitResponsePartition, OffsetCommitResponseTopic,
+};
+use kafka_protocol::messages::{OffsetCommitRequest, OffsetCommitResponse};
+use tokio::time::Instant;
+
+use crate::bridge::IggyBridge;
+use crate::error::Result;
+use crate::group::CommitRequest;
+use crate::protocol::api::{
+ API_KEY_OFFSET_COMMIT, ApiVersionRange,
ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
GatewayState, HandleOutcome,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_commit_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_COMMIT,
+ min_version: 2,
+ max_version: 9,
+};
+
+/// Partition index, error code and tagged fields of one echoed partition.
+const PER_PARTITION_BYTES: usize = 7;
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) ->
HandleOutcome {
+ // A client reads a partition left out of the answer as committed, so a
request that cannot be
+ // answered partition by partition closes the connection instead.
+ if !is_supported_version(API_KEY_OFFSET_COMMIT, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetCommitRequest>(api_version,
body, |version, body| {
+ validate_offset_commit_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
OffsetCommit request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let error = state
+ .groups
+ .validate_commit(&CommitRequest::from(&request))
+ .await;
+ let topics = match (&state.bridge, error) {
+ (Some(bridge), ERROR_NONE) => {
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ commit_all(bridge, &request, deadline).await
+ }
+ (None, ERROR_NONE) => same_code(&request,
ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ (_, error) => same_code(&request, error),
+ };
+ respond_or_close(encode_response(api_version, topics), "OffsetCommit")
+}
+
+/// Stores every requested offset. The partitions run at once across the
offset slots, and a
+/// partition that the deadline leaves out answers 14 and makes no call.
+async fn commit_all(
+ bridge: &IggyBridge,
+ request: &OffsetCommitRequest,
+ deadline: Instant,
+) -> Vec<OffsetCommitResponseTopic> {
+ let group = request.group_id.0.as_str();
+ let targets: Vec<_> = request
+ .topics
+ .iter()
+ .map(|topic| bridge.topic_target(topic.name.0.as_str()))
+ .collect();
+ let mut topics = same_code(request, ERROR_NONE);
+ let mut commits = Vec::new();
+ // Where each commit's code goes: (topic, partition) in `topics`.
+ let mut positions = Vec::new();
+ for (topic_at, (topic, target)) in
request.topics.iter().zip(&targets).enumerate() {
+ for (partition_at, partition) in topic.partitions.iter().enumerate() {
+ let code = match (target,
u32::try_from(partition.partition_index)) {
+ (Err(error), _) => error.to_offset_error_code(),
+ (Ok(_), Err(_)) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ (Ok(target), Ok(index)) => {
+ commits.push((target, index, partition.committed_offset));
+ positions.push((topic_at, partition_at));
+ continue;
+ }
+ };
+ topics[topic_at].partitions[partition_at].error_code = code;
+ }
+ }
+
+ let results = bridge.commit_group_offsets(group, &commits, deadline).await;
+ let mut refused = 0_usize;
+ let mut first_refusal = None;
+ for ((topic_at, partition_at), result) in
positions.into_iter().zip(results) {
+ let Err(error) = result else {
+ continue;
+ };
+ let code = error.to_offset_error_code();
Review Comment:
a replayed store or delete that already landed (RequestAlreadyApplied)
becomes fatal -1. Classic OffsetCommit treats -1 as non-retriable. Fix: map
IggyError::RequestAlreadyApplied to ERROR_NONE on the commit path.
##########
gateways/kafka/tests/offsets_real_bridge_tests.rs:
##########
@@ -0,0 +1,293 @@
+// 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.
+
+//! `OffsetCommit` and `OffsetFetch` against a real `iggy-server` process,
through the handlers
+//! with a connected [`GatewayState`]. The offsets live in Iggy as external
group offsets, so they
+//! outlive the gateway. Responses are decoded by hand for the reason that
+//! `create_topics_real_bridge_tests.rs` gives.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use iggy::prelude::{Consumer, ConsumerGroupClient, ConsumerOffsetClient,
Identifier};
+use serial_test::serial;
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{IggyBridge, OFFSET_GROUP_PREFIX};
+use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::api::{BrokerAdvertise, ERROR_NONE,
GatewayState};
+use iggy_gateway_kafka::protocol::handlers::{offset_commit, offset_fetch};
+
+use codec::Decoder;
+use iggy_server::TestServer;
+use wire::{build_offset_commit_request, build_offset_fetch_request};
+
+const COMMIT_VERSION: i16 = 8;
+const FETCH_VERSION: i16 = 8;
+const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+const GROUP: &str = "orders-app";
+const UNKNOWN_OFFSET: i64 = -1;
+
+async fn connected_state(server: &TestServer) -> GatewayState {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ TEST_MAX_FRAME_SIZE,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(),
CancellationToken::new()),
+ )
+}
+
+async fn seed_topic(server: &TestServer, topic: &str) {
+ IggyBridge::connect(server.test_config())
+ .await
+ .expect("seed bridge should connect")
+ .ensure_stream_and_topic(topic, 1)
+ .await
+ .expect("seed the topic");
+}
+
+/// Commits `offset` for partition 0 of `topic`, outside the group protocol,
and returns its code.
+async fn commit(state: &GatewayState, topic: &str, offset: i64) -> i16 {
+ let body = build_offset_commit_request(COMMIT_VERSION, GROUP, topic,
offset);
+ let response = offset_commit::handle(state, COMMIT_VERSION, body)
+ .await
+ .expect_response("OffsetCommit answers");
+ let codes = decode_commit_codes(response);
+ assert_eq!(codes.len(), 1, "one partition was committed");
+ codes[0]
+}
+
+/// The group error and partition 0's `(offset, error)` for `group` on `topic`.
+async fn fetch(state: &GatewayState, group: &str, topic: &str) -> (i16, i64,
i16) {
+ let body = build_offset_fetch_request(FETCH_VERSION, group, Some(topic));
+ let response = offset_fetch::handle(state, FETCH_VERSION, body)
+ .await
+ .expect_response("OffsetFetch answers");
+ let (group_error, partitions) = decode_fetch(response);
+ assert_eq!(partitions.len(), 1, "partition 0 was asked for");
+ let (_, index, offset, error) = &partitions[0];
+ assert_eq!(*index, 0);
+ (group_error, *offset, *error)
+}
+
+/// Decodes a v8 `OffsetCommit` response into one code per partition, in wire
order.
+fn decode_commit_codes(body: Bytes) -> Vec<i16> {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let mut codes = Vec::new();
+ for _ in 1..d.read_varint().expect("topics") {
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ for _ in 1..d.read_varint().expect("partitions") {
+ let _index = d.read_i32().expect("partition_index");
+ codes.push(d.read_i16().expect("error_code"));
+ d.read_tagged_fields().expect("partition tagged fields");
+ }
+ d.read_tagged_fields().expect("topic tagged fields");
+ }
+ codes
+}
+
+/// Decodes a one-group v8 `OffsetFetch` response into the group error and
every partition as
+/// `(topic, index, offset, error)`.
+fn decode_fetch(body: Bytes) -> (i16, Vec<(String, i32, i64, i16)>) {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ assert_eq!(d.read_varint().expect("groups"), 2, "one group");
+ let _group_id = d.read_compact_nullable_string().expect("group_id");
+ let mut partitions = Vec::new();
+ for _ in 1..d.read_varint().expect("topics") {
+ let name = d
+ .read_compact_nullable_string()
+ .expect("topic name")
+ .expect("a topic name is never null");
+ for _ in 1..d.read_varint().expect("partitions") {
+ let index = d.read_i32().expect("partition_index");
+ let offset = d.read_i64().expect("committed_offset");
+ let _leader_epoch = d.read_i32().expect("committed_leader_epoch");
Review Comment:
[nit] - decode drops committed_leader_epoch and metadata. Epoch 0 still
passes. Java treats 0 as a real epoch. Fix: assert epoch -1 and metadata empty.
--
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]
