You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by kr...@apache.org on 2026/10/06 10:59:06 UTC
(iggy) branch kafka-offsets created (now 86ae086d4)
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a change to branch kafka-offsets
in repository https://gitbox.apache.org/repos/asf/iggy.git
at 86ae086d4 fix: some problems.
This branch includes the following new commits:
new be12328b9 feat(gateways): answer OffsetCommit and OffsetFetch
new 138835366 feat(gateways): store Kafka group offsets in Iggy
new 86ae086d4 fix: some problems.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 03/03: fix: some problems.
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-offsets
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 86ae086d43223bba550396a78e5c57ee599e716a
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Oct 6 16:20:32 2026 +0530
fix: some problems.
---
.config/nextest.toml | 1 +
core/server/src/dispatch/partition.rs | 167 ++++---
gateways/kafka/README.md | 10 +-
gateways/kafka/docs/OFFSET_STORAGE.md | 41 +-
gateways/kafka/scripts/ci-wire-fixtures.sh | 2 +-
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 484 ++++++++++++++++-----
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 14 +-
gateways/kafka/src/group/mod.rs | 10 +-
gateways/kafka/src/group/state.rs | 11 +-
gateways/kafka/src/protocol/api.rs | 6 +-
gateways/kafka/src/protocol/handlers/mod.rs | 17 +
.../kafka/src/protocol/handlers/offset_commit.rs | 175 ++++----
.../kafka/src/protocol/handlers/offset_fetch.rs | 161 ++++---
gateways/kafka/tests/common/wire.rs | 56 +--
gateways/kafka/tests/kafka_client_e2e_tests.rs | 84 +++-
gateways/kafka/tests/offsets_real_bridge_tests.rs | 8 +-
gateways/kafka/tests/version_firewall_tests.rs | 2 +-
gateways/kafka/tools/kafka-tool/src/main.rs | 15 +-
18 files changed, 873 insertions(+), 391 deletions(-)
diff --git a/.config/nextest.toml b/.config/nextest.toml
index 1cedeed01..d511b4599 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -65,6 +65,7 @@ filter = '''
+ binary_id(iggy-gateway-kafka::produce_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::create_topics_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::metadata_real_bridge_tests)
+ + binary_id(iggy-gateway-kafka::offsets_real_bridge_tests)
+ binary_id(iggy-gateway-kafka::phase1_e2e_tests)
'''
test-group = "kafka_bridge"
diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs
index 8ef335606..16ca3163d 100644
--- a/core/server/src/dispatch/partition.rs
+++ b/core/server/src/dispatch/partition.rs
@@ -640,7 +640,8 @@ fn empty_poll_fallback(partition_id: u32) -> (Bytes, FrameChannel) {
/// an unresolved consumer group included, since a deleted group has no offset
/// to report. A malformed request or an unresolved stream, topic, or partition
/// denies with a nonzero status instead, so an addressing typo cannot read
-/// back as a fresh consumer.
+/// back as a fresh consumer. A read its owner does not answer in time denies
+/// with `TransientNotAccepted`.
// TODO(hubcio): plain local partition_read with no primary gate, so a
// follower answers from its own (possibly lagging) offset state. Needs the
// same is-caught-up-primary gate the auto-commit path has, or an explicit
@@ -696,6 +697,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(),
}
}
@@ -1757,55 +1776,7 @@ mod tests {
let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
shard.attach_senders(vec![sender]);
let shard = Rc::new(shard);
-
- // Create metadata and a route so authorization and resolution let the
- // poll reach its owner inbox.
- let metadata = shard.plane.metadata();
- metadata.mux_stm.users().ensure_root_user("iggy", "hash");
- metadata
- .mux_stm
- .update(prepare_message(
- Operation::CreateStream,
- VSR_CLIENT_ID,
- 1,
- &CreateStreamRequest {
- name: WireName::new("stream").unwrap(),
- options: WireOptions::empty(),
- }
- .to_bytes(),
- ))
- .unwrap();
- metadata
- .mux_stm
- .update(prepare_message(
- Operation::CreateTopicWithAssignments,
- VSR_CLIENT_ID,
- 2,
- &CreateTopicWithAssignmentsRequest {
- request: CreateTopicRequest {
- stream_id: WireIdentifier::numeric(0),
- partitions_count: 1,
- name: WireName::new("topic").unwrap(),
- options: WireOptions::empty(),
- },
- derived_options: WireOptions::empty(),
- partitions: vec![CreatedPartitionAssignment {
- partition_id: 0,
- consensus_group_id: 1,
- }],
- created_view: 0,
- }
- .to_bytes(),
- ))
- .unwrap();
- let namespace = metadata
- .mux_stm
- .streams()
- .namespace_from_partition(&WireIdentifier::numeric(0), &WireIdentifier::numeric(0), 0)
- .unwrap();
- shard
- .shards_table()
- .insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
+ route_one_partition(&shard);
let poll_body = PollMessagesRequest {
consumer: WireConsumer::consumer(WireIdentifier::numeric(1)),
stream_id: WireIdentifier::numeric(0),
@@ -1859,6 +1830,102 @@ mod tests {
assert_eq!(bus.client_replies.borrow().len(), 1);
}
+ /// An empty body reads as "no stored offset", and a consumer that reads it
+ /// resets its position. A read with no reply must deny instead.
+ #[compio::test]
+ async fn given_pending_offset_read_when_owner_reply_times_out_should_deny_not_accepted() {
+ const TRANSPORT_CLIENT_ID: u128 = 91;
+ let bus = SpyBus::default();
+ bus.instant_timers.set(true);
+ let mut shard = test_shard(&bus, 0, 1, 1);
+ let (sender, owner_inbox, _owner_replies) = shard_channel(0, 1, 1);
+ shard.attach_senders(vec![sender]);
+ let shard = Rc::new(shard);
+ route_one_partition(&shard);
+ let offset_body = GetConsumerOffsetRequest {
+ consumer: WireConsumer::consumer(WireIdentifier::numeric(1)),
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ partition_id: Some(0),
+ }
+ .to_bytes();
+ let request = request_message(Operation::NonReplicated, 1, 1, 1, &offset_body);
+
+ handle_get_consumer_offset(
+ &shard,
+ TRANSPORT_CLIENT_ID,
+ &request,
+ Some(DEFAULT_ROOT_USER_ID),
+ )
+ .await;
+
+ let replies = bus.client_replies.borrow();
+ assert_eq!(replies.len(), 1);
+ let (client_id, frame) = &replies[0];
+ assert_eq!(*client_id, TRANSPORT_CLIENT_ID);
+ assert_eq!(frame.len(), std::mem::size_of::<ReplyHeader>());
+ let status_start = std::mem::offset_of!(ReplyHeader, status);
+ let status = u32::from_le_bytes(frame[status_start..status_start + 4].try_into().unwrap());
+ assert_eq!(status, IggyError::TransientNotAccepted.as_code());
+ assert!(
+ owner_inbox.try_recv().is_ok(),
+ "the read must have reached its owner before the timeout"
+ );
+ }
+
+ /// Creates one stream with one single-partition topic and routes the
+ /// partition to shard 0, so authorization and resolution let a read reach
+ /// the owner inbox.
+ fn route_one_partition(shard: &TestShard) {
+ const VSR_CLIENT_ID: u128 = 1;
+ let metadata = shard.plane.metadata();
+ metadata.mux_stm.users().ensure_root_user("iggy", "hash");
+ metadata
+ .mux_stm
+ .update(prepare_message(
+ Operation::CreateStream,
+ VSR_CLIENT_ID,
+ 1,
+ &CreateStreamRequest {
+ name: WireName::new("stream").unwrap(),
+ options: WireOptions::empty(),
+ }
+ .to_bytes(),
+ ))
+ .unwrap();
+ metadata
+ .mux_stm
+ .update(prepare_message(
+ Operation::CreateTopicWithAssignments,
+ VSR_CLIENT_ID,
+ 2,
+ &CreateTopicWithAssignmentsRequest {
+ request: CreateTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ name: WireName::new("topic").unwrap(),
+ options: WireOptions::empty(),
+ },
+ derived_options: WireOptions::empty(),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ created_view: 0,
+ }
+ .to_bytes(),
+ ))
+ .unwrap();
+ let namespace = metadata
+ .mux_stm
+ .streams()
+ .namespace_from_partition(&WireIdentifier::numeric(0), &WireIdentifier::numeric(0), 0)
+ .unwrap();
+ shard
+ .shards_table()
+ .insert(namespace, PartitionLocation::new(ShardId::new(0), 0));
+ }
+
/// A partition write whose routable wait exhausts (namespace committed,
/// but no reconciler ever seeds this shard's routing row -- the state a
/// teardown/rematerialise churn leaves behind) must answer a nonzero
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index 490d621be..141304de1 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -22,7 +22,10 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34
> resumes from its last commit, also after a gateway restart
> ([docs/OFFSET_STORAGE.md](docs/OFFSET_STORAGE.md)). Every Iggy server must run a build with the
> external group consumer kind ([#4392](https://github.com/apache/iggy/pull/4392)) before the first
-> commit. In a partly upgraded cluster, one commit can make a partition stop committing.
+> commit. In a partly upgraded cluster, one commit can make a partition stop committing. In a
+> cluster, a node that has not applied a group's first commit answers "no offset", so a consumer on
+> `auto.offset.reset=latest` can skip records until
+> [#4409](https://github.com/apache/iggy/issues/4409).
## Run
@@ -475,7 +478,10 @@ cut before that threshold is reached - worth knowing rather than discovering lat
### Concurrency ceiling
- One `IggyClient` serves Produce, Metadata and CreateTopics for every Kafka connection, one Iggy
- request at a time. Fetch polls use 4 more, one per read slot. Topic probes use 1 more.
+ request at a time. Fetch polls use 4 more, one per read slot. Topic probes use 1 more. Offset
+ calls use 4 more, one per offset slot.
+- OffsetCommit and OffsetFetch cost one Iggy call per partition, spread over the offset slots. In a
+ cluster, an offset slot's client can also open one connection per node for stores.
- `IGGY_KAFKA_MAX_CONNECTIONS` does not change that. A Produce client pool is a TODO in
[docs/SCOPE.md](docs/SCOPE.md).
- Order: set `max.in.flight.requests.per.connection=1`, or `retries=0`. Otherwise a retried batch
diff --git a/gateways/kafka/docs/OFFSET_STORAGE.md b/gateways/kafka/docs/OFFSET_STORAGE.md
index be630935d..cc544e4e6 100644
--- a/gateways/kafka/docs/OFFSET_STORAGE.md
+++ b/gateways/kafka/docs/OFFSET_STORAGE.md
@@ -46,13 +46,28 @@ leaves it behind.
- Metadata string and leader epoch: dropped. OffsetFetch returns `""` and -1.
- No key, unknown group or unknown topic: OffsetFetch returns -1 and no error.
- Any other failed read fails the whole group in OffsetFetch. A lost read never goes back as -1,
- because the consumer then resets its position.
+ because the consumer then resets its position. A read that its owning shard does not answer in
+ time comes back from the server as `TransientNotAccepted`, not as "no offset".
- Null topic list (OffsetFetch v2+): read every Kafka topic that has group `kafka.cg.<group>`. Only
- the partitions with an offset come back.
+ the partitions with an offset come back. A topic deleted during the read is skipped.
- A failure that a retry can fix goes back as `COORDINATOR_LOAD_IN_PROGRESS` (14). The Java client
fails an offset call on 6, and fails OffsetFetch on 7.
+- A failure sent as `UNKNOWN_SERVER_ERROR` (-1) is logged at `error!`: once per OffsetCommit
+ request, and once per group in OffsetFetch.
- Committed offset below the oldest kept message: Fetch reads from the first kept message.
+## Calls
+
+- Offset calls run on 4 offset slots, each with its own Iggy client. They never wait behind
+ Produce.
+- A partition always takes the same slot, and a call keeps its slot until the SDK lets go of it.
+ The calls of one partition reach Iggy in order, even after a caller gives up.
+- A request gives up at half of the minimum session timeout: 3 s at the default 6 s. A heartbeat
+ on the same connection waits behind it. A partition that the deadline leaves out answers 14 and
+ makes no call.
+- Cost: one Iggy call per partition. A null topic list costs one topic listing, one call per Kafka
+ topic, and one call per partition of each topic that holds the group.
+
## Limits
| Limit | Value |
@@ -60,7 +75,16 @@ leaves it behind.
| Keys per partition, per kind | 4096, at most 262144: `partition.consumer_offsets_max` |
| Past the limit | `TooManyConsumerOffsets` (3024). Sent as `UNKNOWN_SERVER_ERROR`, logged at `error!` |
-A group nobody uses keeps its keys until its Iggy group is deleted on each topic.
+The gateway never deletes a `kafka.cg.<group>` group, and does not serve DeleteGroups (42). A group
+nobody uses keeps its keys until an operator deletes its Iggy group on each topic:
+
+```bash
+iggy consumer-group list <stream> <topic>
+iggy consumer-group delete <stream> <topic> kafka.cg.<group>
+```
+
+The gateway does not expire groups. No gateway knows whether another gateway or a standalone
+consumer still uses a group, and Iggy keeps no commit time.
## More than one gateway
@@ -93,9 +117,14 @@ The Iggy HTTP API cannot carry kind 3. The gateway uses TCP, so this does not li
An offset read goes to the Iggy node that the gateway is connected to. That is not always the
partition primary (the node that orders the partition writes). So an OffsetFetch right after an
-OffsetCommit can return the older value until that node applies the commit. A consumer that starts
-from the older value reads some records again. Reads from the primary are
-[#4409](https://github.com/apache/iggy/issues/4409).
+OffsetCommit can return the older value until that node applies the commit.
+
+- After a later commit, the consumer reads some records again.
+- After the first commit of a group, the node has no offset and answers -1. The consumer then
+ applies `auto.offset.reset`. With `latest`, it skips records.
+
+Reads from the primary are [#4409](https://github.com/apache/iggy/issues/4409). Until then, use
+`auto.offset.reset=earliest` for a group that a cluster serves.
## References
diff --git a/gateways/kafka/scripts/ci-wire-fixtures.sh b/gateways/kafka/scripts/ci-wire-fixtures.sh
index ec0793a49..93b24fe93 100755
--- a/gateways/kafka/scripts/ci-wire-fixtures.sh
+++ b/gateways/kafka/scripts/ci-wire-fixtures.sh
@@ -24,7 +24,7 @@ set -euo pipefail
FIXTURES_DIR="gateways/kafka/tools/kafka-tool/kafka_messages"
# API keys requested by api_handler_tests, version_firewall_tests, and server_e2e_tests.
-FIXTURE_API_KEYS=(0 1 2 10 11 12 13 14 19 22)
+FIXTURE_API_KEYS=(0 1 2 8 9 10 11 12 13 14 19 22)
usage() {
echo "Usage: $0 {generate|cleanup}" >&2
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
index 9b9a55fb3..b4b02afb5 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
@@ -17,171 +17,371 @@
//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
-use iggy::prelude::{Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyError};
+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, KafkaTopicMetadata, TopicTarget, with_request_timeout};
+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 `offset` as the committed offset of Kafka group `group` on `partition` of `topic`.
+ /// 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.
- ///
- /// The Iggy group that holds the offsets is created when a store finds it missing, so the
- /// common commit costs one call.
+ /// 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
///
- /// [`BridgeError::Iggy`] if Iggy refuses a call. [`BridgeError::Timeout`] past
- /// `REQUEST_TIMEOUT`.
- pub async fn commit_group_offset(
+ /// 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,
- topic: &TopicTarget,
group: &str,
- partition: u32,
- offset: i64,
- ) -> Result<(), BridgeError> {
- let consumer = offset_consumer(group)?;
- let Ok(offset) = u64::try_from(offset) else {
- return match with_request_timeout(self.client.delete_consumer_offset(
- &consumer,
- &topic.stream_id,
- &topic.topic_id,
- Some(partition),
- ))
- .await
- {
- Err(BridgeError::Iggy(
- IggyError::ConsumerOffsetNotFound(_) | IggyError::ConsumerGroupNameNotFound(..),
- )) => Ok(()),
- result => result,
- };
- };
- let store = || {
- with_request_timeout(self.client.store_consumer_offset(
- &consumer,
- &topic.stream_id,
- &topic.topic_id,
- Some(partition),
- offset,
- ))
+ 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),
};
- match store().await {
- Err(BridgeError::Iggy(IggyError::ConsumerGroupNameNotFound(..))) => {
- self.create_offset_group(topic, group).await?;
- store().await
- }
- result => result,
+ 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` on `partition` of `topic`. `None` when there
- /// is no key or no group.
+ /// 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
///
- /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic.
- /// [`BridgeError::Timeout`] past `REQUEST_TIMEOUT`.
- pub async fn fetch_group_offset(
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
&self,
- topic: &TopicTarget,
group: &str,
- partition: u32,
- ) -> Result<Option<u64>, BridgeError> {
- let consumer = offset_consumer(group)?;
- let info = with_request_timeout(self.client.get_consumer_offset(
- &consumer,
- &topic.stream_id,
- &topic.topic_id,
- Some(partition),
- ))
- .await?;
- Ok(info.map(|info| info.stored_offset))
- }
-
- /// Every Kafka topic that holds offsets of Kafka group `group`, for an `OffsetFetch` with a
- /// null topic list. One call per Kafka topic.
+ 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 a call. [`BridgeError::Timeout`] past
- /// `REQUEST_TIMEOUT`.
- pub async fn offset_group_topics(
+ /// [`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,
- ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
- let group_id = offset_group_id(group)?;
- let mut held = Vec::new();
- for topic in self.list_kafka_topics().await? {
- let target = self.topic_target(&topic.kafka_topic)?;
- let found = with_request_timeout(self.client.get_consumer_group(
- &target.stream_id,
- &target.topic_id,
- &group_id,
- ))
- .await?;
- if found.is_some() {
- held.push(topic);
- }
+ 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(),
}
- Ok(held)
}
- /// Creates the Iggy group of Kafka group `group` on `topic`. A group that exists counts as
- /// created, since another commit or another gateway can create it first.
- async fn create_offset_group(
+ /// The slot of `partition` of `topic`, with its client. `None` if it is not free by
+ /// `deadline`.
+ async fn take(
&self,
topic: &TopicTarget,
- group: &str,
- ) -> Result<(), BridgeError> {
- let name = offset_group_name(group);
- match with_request_timeout(self.client.create_consumer_group(
- &topic.stream_id,
- &topic.topic_id,
- &name,
- ))
- .await
- {
- Ok(_) => {
- info!("created Iggy consumer group '{name}' for Kafka group offsets");
- Ok(())
- }
- Err(BridgeError::Iggy(IggyError::ConsumerGroupNameAlreadyExists(..))) => Ok(()),
- Err(error) => Err(error),
+ 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),
}
}
}
-fn offset_group_name(group: &str) -> String {
- format!("{OFFSET_GROUP_PREFIX}{group}")
+/// [`Started::finish`] for each call, in order. The calls run meanwhile.
+async fn finish_all<T>(started: Vec<Started<T>>, deadline: Instant) -> Vec<Result<T, BridgeError>> {
+ let mut results = Vec::with_capacity(started.len());
+ for call in started {
+ results.push(call.finish(deadline).await);
+ }
+ results
+}
+
+/// One refusal per call, for a group id that names no valid Iggy group.
+fn refused_all<T>(count: usize, error: &IggyError) -> Vec<Result<T, BridgeError>> {
+ (0..count)
+ .map(|_| Err(BridgeError::Iggy(error.clone())))
+ .collect()
+}
+
+/// The Iggy names of one Kafka group's offsets.
+struct OffsetNames {
+ /// `kafka.cg.<group>`, the Iggy group that holds the offsets.
+ group: String,
+ /// The key of each offset: the external group kind of that group.
+ consumer: Consumer,
}
-fn offset_group_id(group: &str) -> Result<Identifier, BridgeError> {
- Identifier::named(&offset_group_name(group)).map_err(BridgeError::Iggy)
+impl OffsetNames {
+ fn new(group: &str) -> Result<Self, IggyError> {
+ let name = format!("{OFFSET_GROUP_PREFIX}{group}");
+ let id = Identifier::named(&name)?;
+ Ok(Self {
+ group: name,
+ consumer: Consumer::external_group(id),
+ })
+ }
}
-/// The Iggy key of Kafka group `group`'s offsets.
-fn offset_consumer(group: &str) -> Result<Consumer, BridgeError> {
- offset_group_id(group).map(Consumer::external_group)
+/// One commit on `client`. A store that finds no Iggy group creates it and stores again.
+async fn commit_on(
+ client: &IggyClient,
+ names: &OffsetNames,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partition: u32,
+ offset: i64,
+) -> Result<(), BridgeError> {
+ let consumer = &names.consumer;
+ let Ok(offset) = u64::try_from(offset) else {
+ return match client
+ .delete_consumer_offset(consumer, stream_id, topic_id, Some(partition))
+ .await
+ {
+ Ok(())
+ | Err(
+ IggyError::ConsumerOffsetNotFound(_) | IggyError::ConsumerGroupNameNotFound(..),
+ ) => Ok(()),
+ Err(error) => Err(BridgeError::Iggy(error)),
+ };
+ };
+ let store =
+ || client.store_consumer_offset(consumer, stream_id, topic_id, Some(partition), offset);
+ match store().await {
+ Err(IggyError::ConsumerGroupNameNotFound(..)) => {
+ create_offset_group(client, stream_id, topic_id, &names.group).await?;
+ store().await.map_err(BridgeError::Iggy)
+ }
+ result => result.map_err(BridgeError::Iggy),
+ }
+}
+
+/// Creates the Iggy group `name` on the topic. A group that exists counts as created, since
+/// another commit or another gateway can create it first.
+async fn create_offset_group(
+ client: &IggyClient,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ name: &str,
+) -> Result<(), BridgeError> {
+ match client
+ .create_consumer_group(stream_id, topic_id, name)
+ .await
+ {
+ Ok(_) => {
+ // `{name:?}`: a client picks the group id, and a newline in it would forge a line.
+ info!("created Iggy consumer group {name:?} for Kafka group offsets");
+ Ok(())
+ }
+ Err(IggyError::ConsumerGroupNameAlreadyExists(..)) => Ok(()),
+ Err(error) => Err(BridgeError::Iggy(error)),
+ }
}
#[cfg(test)]
mod tests {
- use super::*;
+ use std::time::Duration;
+
use iggy::prelude::ConsumerKind;
+ use super::*;
+
+ fn target(topic: &str) -> TopicTarget {
+ TopicTarget {
+ stream_id: Identifier::named("kafka").unwrap(),
+ topic_id: Identifier::named(topic).unwrap(),
+ }
+ }
+
+ fn soon() -> Instant {
+ Instant::now() + Duration::from_millis(20)
+ }
+
#[test]
fn given_a_kafka_group_the_key_should_be_its_prefixed_external_group() {
- let consumer = offset_consumer("orders").unwrap();
+ let names = OffsetNames::new("orders").unwrap();
- assert_eq!(consumer.kind, ConsumerKind::ExternalGroup);
- assert_eq!(consumer.id, Identifier::named("kafka.cg.orders").unwrap());
+ assert_eq!(names.group, "kafka.cg.orders");
+ assert_eq!(names.consumer.kind, ConsumerKind::ExternalGroup);
+ assert_eq!(
+ names.consumer.id,
+ Identifier::named("kafka.cg.orders").unwrap()
+ );
}
/// 246 bytes of group id plus the prefix is the 255 byte cap of an Iggy name.
@@ -189,7 +389,61 @@ mod tests {
fn given_the_longest_kafka_group_id_the_key_should_still_be_a_valid_name() {
let longest = "g".repeat(crate::group::MAX_GROUP_ID_BYTES);
- assert!(offset_consumer(&longest).is_ok());
- assert!(offset_consumer(&format!("{longest}g")).is_err());
+ assert!(OffsetNames::new(&longest).is_ok());
+ assert!(OffsetNames::new(&format!("{longest}g")).is_err());
+ }
+
+ #[test]
+ fn given_the_partitions_of_a_topic_should_pin_each_to_one_slot_and_use_several() {
+ let orders = target("orders");
+ let slots: Vec<usize> = (0..64)
+ .map(|partition| slot_of(&orders, partition))
+ .collect();
+
+ for (partition, &slot) in (0..64).zip(&slots) {
+ assert_eq!(slot_of(&target("orders"), partition), slot);
+ }
+ assert!(
+ slots.iter().any(|&slot| slot != slots[0]),
+ "the partitions of one topic must spread over the slots"
+ );
+ }
+
+ #[tokio::test]
+ async fn given_a_taken_slot_when_the_deadline_passes_should_not_start_the_call() {
+ let pool = OffsetPool::new();
+ let orders = target("orders");
+ let held = pool.take(&orders, 0, soon()).await.expect("a free slot");
+
+ assert!(pool.take(&orders, 0, soon()).await.is_none());
+ drop(held);
+ assert!(pool.take(&orders, 0, soon()).await.is_some());
+ }
+
+ /// A commit given up on can still land, so the next call on its slot must wait for it, or a
+ /// newer offset could land first and lose to the older one.
+ #[tokio::test]
+ async fn given_a_call_given_up_on_when_the_next_one_comes_should_wait_for_it_to_end() {
+ let pool = OffsetPool::new();
+ let orders = target("orders");
+ let (release, released) = oneshot::channel::<()>();
+ let (turn, _) = pool.take(&orders, 0, soon()).await.expect("a free slot");
+ let stuck = Started(Ok(spawn_holding(turn, async move {
+ let _ = released.await;
+ Ok(1)
+ })));
+
+ assert!(matches!(
+ stuck.finish(soon()).await,
+ Err(BridgeError::Timeout)
+ ));
+ assert!(
+ pool.take(&orders, 0, soon()).await.is_none(),
+ "the call still holds its slot"
+ );
+
+ release.send(()).expect("the call waits for its release");
+ let later = Instant::now() + Duration::from_secs(5);
+ assert!(pool.take(&orders, 0, later).await.is_some());
}
}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index 82f79d553..414bc8b11 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -39,6 +39,7 @@ mod topics;
use fetch::{FetchPool, LazyClient};
pub(crate) use fetch::{FetchSlot, PartitionProbe, TopicProbe};
pub use group_offsets::OFFSET_GROUP_PREFIX;
+use group_offsets::OffsetPool;
pub use topics::{KafkaTopicMetadata, TopicCreationOutcome};
/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up and returns `Err`.
@@ -105,8 +106,8 @@ async fn in_slot<S: Send + 'static, T: Send + 'static>(
/// Owns the connected `IggyClient`s and resolves Kafka topics against them.
///
/// One lockstep client serves Produce, Metadata and `CreateTopics` for every Kafka connection, so
-/// those Iggy calls run one at a time. Topic probes use a client of their own, and Fetch polls one
-/// per read slot. A Produce pool is a TODO in `docs/SCOPE.md`.
+/// those Iggy calls run one at a time. Topic probes use a client of their own, Fetch polls one per
+/// read slot, and offset calls one per offset slot. A Produce pool is a TODO in `docs/SCOPE.md`.
///
/// Each client signs in at the metadata leader, and a view change or a refused call can move it to
/// another node. So in a cluster, two clients can read replicas that are not at the same offset.
@@ -119,6 +120,8 @@ pub struct IggyBridge {
fetch_pool: Arc<FetchPool>,
/// The client of every topic probe, Fetch's and `ListOffsets`'. See `probe`.
probe_client: LazyClient,
+ /// Group offset slots, each with its own client. See `commit_group_offsets`.
+ offset_pool: OffsetPool,
}
/// The Iggy stream and topic one Kafka topic maps to. Resolve once per topic, use many times.
@@ -156,6 +159,7 @@ impl IggyBridge {
send_slot: Arc::new(Semaphore::new(1)),
fetch_pool: Arc::new(FetchPool::new()),
probe_client: LazyClient::default(),
+ offset_pool: OffsetPool::new(),
})
}
@@ -201,8 +205,12 @@ impl IggyBridge {
pub async fn close(self) -> Result<(), BridgeError> {
let fetch_clients = self.fetch_pool.close().await;
let probe_client = self.probe_client.close().await;
+ let offset_clients = self.offset_pool.close().await;
let shared_client = with_request_timeout(self.client.shutdown()).await;
- fetch_clients.and(probe_client).and(shared_client)
+ fetch_clients
+ .and(probe_client)
+ .and(offset_clients)
+ .and(shared_client)
}
}
diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs
index af42eb09d..d9fe9c8e8 100644
--- a/gateways/kafka/src/group/mod.rs
+++ b/gateways/kafka/src/group/mod.rs
@@ -27,7 +27,7 @@
mod state;
-pub use state::MAX_GROUP_ID_BYTES;
+pub use state::{MAX_GROUP_ID_BYTES, is_valid_group_id};
use std::collections::HashMap;
use std::time::Duration;
@@ -483,6 +483,14 @@ impl GroupCoordinator {
state::commit_step(&mut groups, request, Instant::now())
}
+ /// How long an `OffsetCommit` or `OffsetFetch` may hold its connection. A member's heartbeat
+ /// waits behind it there, so it takes half the shortest session, leaving the other half for
+ /// one heartbeat interval.
+ #[must_use]
+ pub fn offset_deadline(&self) -> Duration {
+ self.config.min_session_timeout / 2
+ }
+
/// Sleeps until the group changes or `wake_at` passes. `false` means the gateway is draining.
async fn wait_until(&self, mut receiver: watch::Receiver<u64>, wake_at: Instant) -> bool {
tokio::select! {
diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs
index e5335bfc8..8d47fd8a4 100644
--- a/gateways/kafka/src/group/state.rs
+++ b/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
+/// [`MAX_GROUP_ID_BYTES`] cannot name its offsets in Iggy.
+#[must_use]
+pub const fn is_valid_group_id(group_id: &str) -> bool {
+ !group_id.is_empty() && group_id.len() <= MAX_GROUP_ID_BYTES
+}
+
/// Protocol selection is polynomial in this count under the coordinator lock. Stock clients list
/// one or two assignors.
pub const MAX_PROTOCOLS_PER_MEMBER: usize = 16;
@@ -677,7 +684,7 @@ fn roster_entry_bytes(
/// Everything a `JoinGroup` can be rejected for before any group is touched.
fn join_request_error(config: &GroupCoordinatorConfig, request: &JoinRequest) -> Option<i16> {
- if request.group_id.is_empty() || request.group_id.len() > MAX_GROUP_ID_BYTES {
+ if !is_valid_group_id(&request.group_id) {
return Some(ERROR_INVALID_GROUP_ID);
}
if request.session_timeout < config.min_session_timeout
@@ -1118,7 +1125,7 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
/// 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 request.group_id.is_empty() || request.group_id.len() > MAX_GROUP_ID_BYTES {
+ if !is_valid_group_id(&request.group_id) {
return ERROR_INVALID_GROUP_ID;
}
let empty = || {
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index 5857ed707..aecd27809 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -296,9 +296,9 @@ pub fn supported_api_ranges() -> &'static [ApiVersionRange] {
/// `bridge` is `None` until `IGGY_KAFKA_BRIDGE_ENABLED` turns it on. A handler that finds `None`
/// answers with its stub, so APIs can be wired one at a time.
///
-/// Every Iggy call but a Fetch poll goes through one lockstep `IggyClient`, so Kafka connections
-/// serialize behind whichever of those calls is in flight. The `Arc` does not change that. See the
-/// README's "Concurrency ceiling".
+/// Every Iggy call but a Fetch poll, a topic probe and an offset call goes through one lockstep
+/// `IggyClient`, so Kafka connections serialize behind whichever of those calls is in flight. The
+/// `Arc` does not change that. See the README's "Concurrency ceiling".
pub struct GatewayState {
pub broker: BrokerAdvertise,
pub bridge: Option<Arc<IggyBridge>>,
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
index 05d52ae26..5645e330d 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -40,10 +40,12 @@ pub mod produce;
pub mod sync_group;
use bytes::{Buf, Bytes, BytesMut};
+use iggy::prelude::IggyError;
use kafka_protocol::messages::TransactionalId;
use kafka_protocol::protocol::{Decodable, Encodable};
use tokio::runtime::{Handle, RuntimeFlavor};
+use crate::bridge::BridgeError;
use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_FETCH, API_KEY_FIND_COORDINATOR,
@@ -110,6 +112,21 @@ pub(crate) fn is_transactional(transactional_id: Option<&TransactionalId>) -> bo
transactional_id.is_some_and(|id| !id.is_empty())
}
+/// What the operator can do about an offset call that the client sees as -1. The client sees no
+/// more, so the log is the one place the cause shows.
+const fn offset_refusal_cause(error: &BridgeError) -> &'static str {
+ match error {
+ BridgeError::Iggy(IggyError::TooManyConsumerOffsets) => {
+ "the partition holds partition.consumer_offsets_max keys"
+ }
+ BridgeError::Iggy(IggyError::InvalidCommand) => {
+ "a server without the external group kind answers this"
+ }
+ error if error.is_bridge_login_rejected() => "Iggy rejected the bridge's credentials",
+ _ => "no Kafka code fits this Iggy error",
+ }
+}
+
/// Encode a `kafka_protocol` message, mapping its `anyhow::Error` (the crate has no stable
/// decode/encode error taxonomy) to a variant callers can log or fold into
/// [`HandleOutcome::Close`].
diff --git a/gateways/kafka/src/protocol/handlers/offset_commit.rs b/gateways/kafka/src/protocol/handlers/offset_commit.rs
index f65515c68..43d428a55 100644
--- a/gateways/kafka/src/protocol/handlers/offset_commit.rs
+++ b/gateways/kafka/src/protocol/handlers/offset_commit.rs
@@ -22,23 +22,24 @@
//! every partition.
use bytes::Bytes;
-use iggy::prelude::IggyError;
use kafka_protocol::messages::offset_commit_response::{
OffsetCommitResponsePartition, OffsetCommitResponseTopic,
};
use kafka_protocol::messages::{OffsetCommitRequest, OffsetCommitResponse};
-use tokio::time::{Instant, timeout_at};
+use tokio::time::Instant;
-use crate::bridge::{BridgeError, IggyBridge};
+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_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, REQUEST_DEADLINE,
+ 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, respond_or_close};
+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,
@@ -49,9 +50,6 @@ pub const RANGE: ApiVersionRange = ApiVersionRange {
/// Partition index, error code and tagged fields of one echoed partition.
const PER_PARTITION_BYTES: usize = 7;
-/// One code per requested partition, in request order.
-pub type PartitionCodes = Vec<Vec<i16>>;
-
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.
@@ -73,119 +71,110 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
.groups
.validate_commit(&CommitRequest::from(&request))
.await;
- let codes = match (&state.bridge, error) {
+ let topics = match (&state.bridge, error) {
(Some(bridge), ERROR_NONE) => {
- commit_all(bridge, &request, Instant::now() + REQUEST_DEADLINE).await
+ 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, &request, &codes),
- "OffsetCommit",
- )
+ respond_or_close(encode_response(api_version, topics), "OffsetCommit")
}
-/// Stores every requested offset, in request order. Each partition costs one call on the lockstep
-/// client, so a partition that the deadline leaves out answers 14 and makes no call.
+/// 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,
-) -> PartitionCodes {
+) -> Vec<OffsetCommitResponseTopic> {
let group = request.group_id.0.as_str();
- let mut codes = Vec::with_capacity(request.topics.len());
- for topic in &request.topics {
- let target = bridge.topic_target(topic.name.0.as_str());
- let mut topic_codes = Vec::with_capacity(topic.partitions.len());
- for partition in &topic.partitions {
- let code = match (&target, u32::try_from(partition.partition_index)) {
+ 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,
- // A call started now still goes out once, and the client retries the 14 anyway.
- (Ok(_), Ok(_)) if Instant::now() >= deadline => ERROR_COORDINATOR_LOAD_IN_PROGRESS,
- (Ok(target), Ok(index)) => match timeout_at(
- deadline,
- bridge.commit_group_offset(target, group, index, partition.committed_offset),
- )
- .await
- {
- Ok(result) => commit_code(result),
- Err(_elapsed) => ERROR_COORDINATOR_LOAD_IN_PROGRESS,
- },
+ (Ok(target), Ok(index)) => {
+ commits.push((target, index, partition.committed_offset));
+ positions.push((topic_at, partition_at));
+ continue;
+ }
};
- topic_codes.push(code);
+ topics[topic_at].partitions[partition_at].error_code = code;
}
- codes.push(topic_codes);
}
- codes
-}
-/// The Kafka code for one partition's store or delete. A refusal that only the operator can fix
-/// is logged, since the client sees no more than -1.
-fn commit_code(result: core::result::Result<(), BridgeError>) -> i16 {
- let Err(error) = result else {
- return ERROR_NONE;
- };
- match &error {
- BridgeError::Iggy(IggyError::TooManyConsumerOffsets) => tracing::error!(
- %error,
- "Iggy refused a Kafka offset commit: the partition holds \
- partition.consumer_offsets_max keys"
- ),
- BridgeError::Iggy(IggyError::InvalidCommand) => tracing::error!(
+ 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();
+ topics[topic_at].partitions[partition_at].error_code = code;
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ refused += 1;
+ if first_refusal.is_none() {
+ first_refusal = Some((topic_at, partition_at, error));
+ }
+ }
+ }
+ // One line per request: a full partition or an old server refuses each commit, every few
+ // seconds.
+ if let Some((topic_at, partition_at, error)) = first_refusal {
+ let topic = &topics[topic_at];
+ tracing::error!(
+ group,
+ kafka_topic = topic.name.0.as_str(),
+ partition = topic.partitions[partition_at].partition_index,
+ refused,
%error,
- "Iggy refused a Kafka offset commit as an invalid command, which is what a server \
- without the external group kind answers"
- ),
- _ => {}
+ cause = offset_refusal_cause(&error),
+ "Iggy refused Kafka offset commits"
+ );
}
- error.to_offset_error_code()
+ topics
}
-/// `code` for every requested partition.
-#[must_use]
-pub fn same_code(request: &OffsetCommitRequest, code: i16) -> PartitionCodes {
+/// Every requested topic and partition, in request order, with `code`.
+fn same_code(request: &OffsetCommitRequest, code: i16) -> Vec<OffsetCommitResponseTopic> {
request
.topics
.iter()
- .map(|topic| vec![code; topic.partitions.len()])
- .collect()
-}
-
-/// Echoes every requested topic and partition, in request order, with its code.
-///
-/// # Errors
-///
-/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_response(
- version: i16,
- request: &OffsetCommitRequest,
- codes: &PartitionCodes,
-) -> Result<Bytes> {
- let topics = request
- .topics
- .iter()
- .zip(codes)
- .map(|(topic, codes)| {
+ .map(|topic| {
OffsetCommitResponseTopic::default()
.with_name(topic.name.clone())
.with_partitions(
topic
.partitions
.iter()
- .zip(codes)
- .map(|(partition, code)| {
+ .map(|partition| {
OffsetCommitResponsePartition::default()
.with_partition_index(partition.partition_index)
- .with_error_code(*code)
+ .with_error_code(code)
})
.collect(),
)
})
- .collect();
- let echoed: usize = request
- .topics
+ .collect()
+}
+
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(version: i16, topics: Vec<OffsetCommitResponseTopic>) -> Result<Bytes> {
+ let echoed: usize = topics
.iter()
.map(|topic| topic.name.0.len() + topic.partitions.len() * PER_PARTITION_BYTES)
.sum();
@@ -199,6 +188,7 @@ pub fn encode_response(
#[cfg(test)]
mod tests {
use bytes::BytesMut;
+ use iggy::prelude::IggyError;
use kafka_protocol::messages::offset_commit_request::{
OffsetCommitRequestPartition, OffsetCommitRequestTopic,
};
@@ -206,9 +196,8 @@ mod tests {
use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
use super::*;
- use crate::protocol::api::{
- BrokerAdvertise, ERROR_UNKNOWN_MEMBER_ID, ERROR_UNKNOWN_SERVER_ERROR,
- };
+ use crate::bridge::BridgeError;
+ use crate::protocol::api::{BrokerAdvertise, ERROR_UNKNOWN_MEMBER_ID};
fn request(generation_id: i32, member_id: &'static str) -> OffsetCommitRequest {
OffsetCommitRequest::default()
@@ -283,27 +272,27 @@ mod tests {
}
#[test]
- fn given_bridge_results_the_commit_codes_should_keep_retriable_failures_retriable() {
- assert_eq!(commit_code(Ok(())), ERROR_NONE);
+ fn given_bridge_errors_when_mapped_for_offsets_should_keep_retriable_failures_retriable() {
for retriable in [
BridgeError::Timeout,
BridgeError::Iggy(IggyError::Disconnected),
BridgeError::Iggy(IggyError::TransientNotCommitted),
] {
assert_eq!(
- commit_code(Err(retriable)),
+ retriable.to_offset_error_code(),
ERROR_COORDINATOR_LOAD_IN_PROGRESS
);
}
assert_eq!(
- commit_code(Err(BridgeError::Iggy(IggyError::TooManyConsumerOffsets))),
+ BridgeError::Iggy(IggyError::TooManyConsumerOffsets).to_offset_error_code(),
ERROR_UNKNOWN_SERVER_ERROR
);
assert_eq!(
- commit_code(Err(BridgeError::Iggy(IggyError::TopicNameNotFound(
+ BridgeError::Iggy(IggyError::TopicNameNotFound(
"orders".to_string(),
"kafka".to_string(),
- )))),
+ ))
+ .to_offset_error_code(),
ERROR_UNKNOWN_TOPIC_OR_PARTITION
);
}
diff --git a/gateways/kafka/src/protocol/handlers/offset_fetch.rs b/gateways/kafka/src/protocol/handlers/offset_fetch.rs
index 507c6d84d..eb5f92bf3 100644
--- a/gateways/kafka/src/protocol/handlers/offset_fetch.rs
+++ b/gateways/kafka/src/protocol/handlers/offset_fetch.rs
@@ -29,16 +29,19 @@ use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse,
use kafka_protocol::protocol::StrBytes;
use tokio::time::{Instant, timeout_at};
-use crate::bridge::{BridgeError, IggyBridge};
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
use crate::error::Result;
-use crate::group::MAX_GROUP_ID_BYTES;
+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_TOPIC_OR_PARTITION, GatewayState,
- HandleOutcome, REQUEST_DEADLINE, UNKNOWN_OFFSET, is_supported_version,
+ 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, respond_or_close};
+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,
@@ -55,16 +58,15 @@ const PER_PARTITION_BYTES: usize = 24;
/// One group a request asks about, across wire versions.
#[derive(Debug, Clone)]
-pub struct GroupQuery {
- pub group_id: StrBytes,
+struct GroupQuery {
+ group_id: StrBytes,
/// `None` asks for every committed offset of the group.
- pub topics: Option<Vec<(TopicName, Vec<i32>)>>,
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
}
impl GroupQuery {
/// One query per group the request names: one below v8, the `groups` array from v8.
- #[must_use]
- pub fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
if version >= FIRST_BATCHED_VERSION {
return request
.groups
@@ -94,14 +96,13 @@ impl GroupQuery {
/// The answer for one group. A group error carries no topics.
#[derive(Debug, Clone)]
-pub struct GroupAnswer {
- pub error: i16,
- pub topics: Vec<TopicAnswer>,
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
}
impl GroupAnswer {
- #[must_use]
- pub const fn error(error: i16) -> Self {
+ const fn error(error: i16) -> Self {
Self {
error,
topics: Vec::new(),
@@ -110,17 +111,17 @@ impl GroupAnswer {
}
#[derive(Debug, Clone)]
-pub struct TopicAnswer {
- pub name: TopicName,
- pub partitions: Vec<PartitionAnswer>,
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
}
/// `offset` is -1 when nothing is committed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub struct PartitionAnswer {
- pub index: i32,
- pub offset: i64,
- pub error: i16,
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
}
pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
@@ -141,7 +142,7 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
};
let queries = GroupQuery::all(api_version, &request);
- let deadline = Instant::now() + REQUEST_DEADLINE;
+ 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);
@@ -155,16 +156,20 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
/// 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 query.group_id.is_empty() || query.group_id.len() > MAX_GROUP_ID_BYTES {
+ 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).await,
- None => read_every_topic(bridge, &query.group_id).await,
+ 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 {
@@ -172,39 +177,71 @@ async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Insta
error: ERROR_NONE,
topics,
},
- Ok(Err(error)) => GroupAnswer::error(error.to_offset_error_code()),
+ Ok(Err(error)) => GroupAnswer::error(refusal(&query.group_id, &error)),
Err(_elapsed) => GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
}
}
-/// Every requested partition, in request order. A partition with no offset answers -1.
+/// 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> {
- let mut answers = Vec::with_capacity(topics.len());
- for (name, indexes) in topics {
- // A name that cannot map to Iggy names no topic, so nothing is committed under it.
- let target = bridge.topic_target(name.0.as_str()).ok();
- let mut partitions = Vec::with_capacity(indexes.len());
- for &index in indexes {
- let offset = match (&target, u32::try_from(index)) {
- (Some(target), Ok(partition)) => {
- committed(bridge.fetch_group_offset(target, group, partition).await)?
- }
- _ => UNKNOWN_OFFSET,
- };
- partitions.push(PartitionAnswer {
- index,
- offset,
- error: ERROR_NONE,
- });
- }
- answers.push(TopicAnswer {
+ // 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,
- });
+ 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)
}
@@ -214,13 +251,27 @@ async fn read_topics(
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.offset_group_topics(group).await? {
+ for topic in bridge.list_kafka_topics().await? {
let target = bridge.topic_target(&topic.kafka_topic)?;
+ match bridge.holds_offset_group(group, &target, deadline).await {
+ Ok(true) => {}
+ Ok(false) => continue,
+ // A topic deleted since the listing holds no offsets.
+ Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => {
+ continue;
+ }
+ 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 in 0..topic.partitions_count {
- let offset = committed(bridge.fetch_group_offset(&target, group, partition).await)?;
+ 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),
@@ -258,11 +309,7 @@ fn committed(
/// # Errors
///
/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
-pub fn encode_response(
- version: i16,
- queries: &[GroupQuery],
- answers: &[GroupAnswer],
-) -> Result<Bytes> {
+fn encode_response(version: i16, queries: &[GroupQuery], answers: &[GroupAnswer]) -> Result<Bytes> {
let partitions: usize = answers
.iter()
.flat_map(|answer| &answer.topics)
diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs
index 8238f9adc..50be77d1c 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -633,9 +633,10 @@ pub fn build_offset_commit_request(
enc.freeze()
}
-/// `OffsetFetch` request for partition 0 of `topic`. One group: top level below v8, the
-/// `groups` array from v8.
-pub fn build_offset_fetch_request(version: i16, group_id: &str, topic: &str) -> Bytes {
+/// `OffsetFetch` request for one group: top level below v8, the `groups` array from v8.
+/// `Some(topic)` asks for partition 0 of `topic`. `None` sends a null topic list, which asks for
+/// every offset of the group, from version 2.
+pub fn build_offset_fetch_request(version: i16, group_id: &str, topic: Option<&str>) -> Bytes {
let flexible = version >= 6;
let mut enc = Encoder::with_capacity(64);
@@ -647,43 +648,18 @@ pub fn build_offset_fetch_request(version: i16, group_id: &str, topic: &str) ->
write_string(&mut enc, true, None); // member_id
enc.write_i32(-1); // member_epoch
}
- write_array_count(&mut enc, flexible, 1); // topics
- write_string(&mut enc, flexible, Some(topic));
- write_array_count(&mut enc, flexible, 1); // partition_indexes
- enc.write_i32(0);
- if flexible {
- enc.write_empty_tagged_fields(); // topic
- }
- if version >= 8 {
- enc.write_empty_tagged_fields(); // group
- }
- if version >= 7 {
- enc.write_bool(false); // require_stable
- }
- if flexible {
- enc.write_empty_tagged_fields(); // request
- }
- enc.freeze()
-}
-
-/// `OffsetFetch` request with a null topic list, which asks for every offset of the group.
-/// Version 2 or later.
-pub fn build_offset_fetch_all_request(version: i16, group_id: &str) -> Bytes {
- let flexible = version >= 6;
- let mut enc = Encoder::with_capacity(32);
-
- if version >= 8 {
- write_array_count(&mut enc, true, 1); // groups
- }
- write_string(&mut enc, flexible, Some(group_id));
- if version >= 9 {
- write_string(&mut enc, true, None); // member_id
- enc.write_i32(-1); // member_epoch
- }
- if flexible {
- enc.write_varint(0); // topics: null
- } else {
- enc.write_i32(-1); // topics: null
+ match topic {
+ Some(topic) => {
+ write_array_count(&mut enc, flexible, 1); // topics
+ write_string(&mut enc, flexible, Some(topic));
+ write_array_count(&mut enc, flexible, 1); // partition_indexes
+ enc.write_i32(0);
+ if flexible {
+ enc.write_empty_tagged_fields(); // topic
+ }
+ }
+ None if flexible => enc.write_varint(0), // topics: null
+ None => enc.write_i32(-1), // topics: null
}
if version >= 8 {
enc.write_empty_tagged_fields(); // group
diff --git a/gateways/kafka/tests/kafka_client_e2e_tests.rs b/gateways/kafka/tests/kafka_client_e2e_tests.rs
index e283f050f..aab4e0868 100644
--- a/gateways/kafka/tests/kafka_client_e2e_tests.rs
+++ b/gateways/kafka/tests/kafka_client_e2e_tests.rs
@@ -25,7 +25,7 @@
//!
//! The stack is real on all three sides: a spawned `iggy-server` process, the gateway in-process
//! with a real `IggyAuthenticator`, and a client from a container. They automate categories S and T
-//! of `docs/MANUAL_TESTING.md`.
+//! of `docs/MANUAL_TESTING.md`, and test G9.
//!
//! The Fetch tests swap the authenticator for a bridge: kcat writes records through Produce, and
//! kcat and the Java consumer read them back through Fetch.
@@ -512,14 +512,15 @@ fn ready_server() -> Option<TestServer> {
}
}
-/// Writes `RECORDS` to partition 0 of `TOPIC` through the gateway, with kcat's producer and one
+/// Writes `records` to partition 0 of `TOPIC` through the gateway, with kcat's producer and one
/// header on each record.
-fn produce_with_kcat(gateway: SocketAddr) {
+fn produce_with_kcat(gateway: SocketAddr, records: &[(&str, &str)]) {
let dir = tempfile::tempdir().expect("create records dir");
let path = dir.path().join("records.txt");
- let lines = RECORDS
- .map(|(key, value)| format!("{key}:{value}\n"))
- .join("");
+ let lines: String = records
+ .iter()
+ .flat_map(|&(key, value)| [key, ":", value, "\n"])
+ .collect();
std::fs::write(&path, lines).expect("write records");
run_client(
KCAT_IMAGE,
@@ -824,7 +825,7 @@ async fn given_records_produced_by_kcat_when_kcat_consumes_should_read_them_back
let Some((_server, gateway)) = bridged_stack().await else {
return;
};
- produce_with_kcat(gateway);
+ produce_with_kcat(gateway, &RECORDS);
// `-e` stops at the end of the partition, which kcat takes from the high watermark that Fetch
// sends. A wrong watermark stops it early or holds it until the container timeout.
let output = run_client(
@@ -868,6 +869,73 @@ async fn given_records_produced_by_kcat_when_kcat_consumes_should_read_them_back
);
}
+/// G9 of `docs/MANUAL_TESTING.md`, with a real client on both offset APIs. The first run reads
+/// every record and commits as it closes. `--from-beginning` applies only to a group with no
+/// committed offset, so the second run must print only the record written in between.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn given_a_group_that_committed_when_the_java_consumer_runs_again_should_resume_after_it() {
+ let Some((_server, gateway)) = bridged_stack().await else {
+ return;
+ };
+ produce_with_kcat(gateway, &RECORDS);
+ let expected: Vec<String> = RECORDS
+ .iter()
+ .enumerate()
+ .map(|(offset, (key, value))| format!("Offset:{offset}\t{key}\t{value}"))
+ .collect();
+ let first = consume_as_group(gateway, RECORDS.len());
+ assert_eq!(
+ records_read(&first),
+ expected,
+ "a group with no offset must start from the beginning, got: {first}"
+ );
+
+ produce_with_kcat(gateway, &[("k4", "delta")]);
+ let second = consume_as_group(gateway, 1);
+ assert_eq!(
+ records_read(&second),
+ vec![format!("Offset:{}\tk4\tdelta", RECORDS.len())],
+ "the group must resume after its commit, got: {second}"
+ );
+}
+
+/// Reads `count` records of `TOPIC` with the Java console consumer in group `e2e-resume`, from
+/// the group's committed offset, or from the beginning when it has none. It commits as it closes.
+fn consume_as_group(gateway: SocketAddr, count: usize) -> String {
+ let count = count.to_string();
+ run_client(
+ KAFKA_IMAGE,
+ &[
+ "/opt/kafka/bin/kafka-console-consumer.sh",
+ "--bootstrap-server",
+ &gateway.to_string(),
+ "--topic",
+ TOPIC,
+ "--group",
+ "e2e-resume",
+ "--from-beginning",
+ "--max-messages",
+ &count,
+ "--timeout-ms",
+ "20000",
+ "--property",
+ "print.offset=true",
+ "--property",
+ "print.key=true",
+ ],
+ &[],
+ )
+ .expect_ran("java console consumer in a group")
+}
+
+/// The records a console consumer printed, as `Offset:<offset>\t<key>\t<value>`.
+fn records_read(output: &str) -> Vec<&str> {
+ output
+ .lines()
+ .filter(|line| line.starts_with("Offset:"))
+ .collect()
+}
+
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn given_records_when_the_java_consumer_reads_the_partition_should_get_them_in_order() {
// A second Fetch client with its own code: the Java consumer resolves `earliest` through
@@ -876,7 +944,7 @@ async fn given_records_when_the_java_consumer_reads_the_partition_should_get_the
let Some((_server, gateway)) = bridged_stack().await else {
return;
};
- produce_with_kcat(gateway);
+ produce_with_kcat(gateway, &RECORDS);
let count = RECORDS.len().to_string();
let output = run_client(
KAFKA_IMAGE,
diff --git a/gateways/kafka/tests/offsets_real_bridge_tests.rs b/gateways/kafka/tests/offsets_real_bridge_tests.rs
index e4ee39905..3ddd44203 100644
--- a/gateways/kafka/tests/offsets_real_bridge_tests.rs
+++ b/gateways/kafka/tests/offsets_real_bridge_tests.rs
@@ -41,9 +41,7 @@ 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_all_request, build_offset_fetch_request,
-};
+use wire::{build_offset_commit_request, build_offset_fetch_request};
const COMMIT_VERSION: i16 = 8;
const FETCH_VERSION: i16 = 8;
@@ -87,7 +85,7 @@ async fn commit(state: &GatewayState, topic: &str, offset: i64) -> i16 {
/// 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, topic);
+ 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");
@@ -277,7 +275,7 @@ async fn given_a_null_topic_list_fetch_should_answer_every_committed_partition()
assert_eq!(commit(&state, "orders", 3).await, ERROR_NONE);
assert_eq!(commit(&state, "payments", 5).await, ERROR_NONE);
- let body = build_offset_fetch_all_request(FETCH_VERSION, GROUP);
+ let body = build_offset_fetch_request(FETCH_VERSION, GROUP, None);
let response = offset_fetch::handle(&state, FETCH_VERSION, body)
.await
.expect_response("OffsetFetch answers");
diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs
index d54ccc928..4ee8cee68 100644
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@ -521,7 +521,7 @@ fn request_body_for_scoped_api(api_key: i16, name: &str, version: i16) -> Bytes
),
API_KEY_INIT_PRODUCER_ID => build_init_producer_id_request(version, None),
API_KEY_OFFSET_COMMIT => build_offset_commit_request(version, "scope-group", "orders", 0),
- API_KEY_OFFSET_FETCH => build_offset_fetch_request(version, "scope-group", "orders"),
+ API_KEY_OFFSET_FETCH => build_offset_fetch_request(version, "scope-group", Some("orders")),
_ => Bytes::new(),
}
}
diff --git a/gateways/kafka/tools/kafka-tool/src/main.rs b/gateways/kafka/tools/kafka-tool/src/main.rs
index 0d70a23fe..09f14b9a5 100644
--- a/gateways/kafka/tools/kafka-tool/src/main.rs
+++ b/gateways/kafka/tools/kafka-tool/src/main.rs
@@ -29,6 +29,7 @@ use kafka_protocol::messages::join_group_request::*;
use kafka_protocol::messages::leave_group_request::MemberIdentity;
use kafka_protocol::messages::list_offsets_request::*;
use kafka_protocol::messages::offset_commit_request::*;
+use kafka_protocol::messages::offset_fetch_request::OffsetFetchRequestGroup;
use kafka_protocol::messages::produce_request::*;
use kafka_protocol::messages::txn_offset_commit_request::*;
use kafka_protocol::messages::*;
@@ -360,10 +361,16 @@ fn build_payload(api_key: i16, version: i16) -> Result<Bytes> {
.context("OffsetCommit")?;
}
9 => {
- OffsetFetchRequest::default()
- .with_group_id(GroupId::from(StrBytes::from_static_str("test-group")))
- .encode(&mut buf, version)
- .context("OffsetFetch")?;
+ let group = GroupId::from(StrBytes::from_static_str("test-group"));
+ // v8 moved the group into `groups`, and the encoder refuses the top-level id there.
+ let request = if version >= 8 {
+ OffsetFetchRequest::default().with_groups(vec![
+ OffsetFetchRequestGroup::default().with_group_id(group),
+ ])
+ } else {
+ OffsetFetchRequest::default().with_group_id(group)
+ };
+ request.encode(&mut buf, version).context("OffsetFetch")?;
}
10 => {
let key = StrBytes::from_static_str("test-group");
(iggy) 01/03: feat(gateways): answer OffsetCommit and OffsetFetch
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-offsets
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit be12328b90006d13d8f723c11ac6eb802951160b
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Oct 6 09:09:18 2026 +0530
feat(gateways): answer OffsetCommit and OffsetFetch
Protocol code only. Offsets go to Iggy once the ExternalGroup kind lands.
---
gateways/kafka/docs/OFFSET_STORAGE.md | 243 ++++----------
gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 4 +-
gateways/kafka/src/group/mod.rs | 40 ++-
gateways/kafka/src/group/state.rs | 236 +++++++++++++-
gateways/kafka/src/protocol/api.rs | 10 +-
gateways/kafka/src/protocol/bounds_guard.rs | 264 ++++++++++++++-
gateways/kafka/src/protocol/handlers/mod.rs | 11 +-
.../kafka/src/protocol/handlers/offset_commit.rs | 234 +++++++++++++
.../kafka/src/protocol/handlers/offset_fetch.rs | 362 +++++++++++++++++++++
gateways/kafka/tests/common/scope.rs | 2 +
gateways/kafka/tests/common/wire.rs | 71 +++-
gateways/kafka/tests/golden_wire_fixtures_tests.rs | 16 +-
gateways/kafka/tests/server_e2e_tests.rs | 6 +-
gateways/kafka/tests/version_firewall_tests.rs | 16 +-
14 files changed, 1299 insertions(+), 216 deletions(-)
diff --git a/gateways/kafka/docs/OFFSET_STORAGE.md b/gateways/kafka/docs/OFFSET_STORAGE.md
index 9389224e5..fdc9e5fbb 100644
--- a/gateways/kafka/docs/OFFSET_STORAGE.md
+++ b/gateways/kafka/docs/OFFSET_STORAGE.md
@@ -1,202 +1,75 @@
# Consumer offset storage
-Status: accepted. Answers [#3540](https://github.com/apache/iggy/issues/3540) and unblocks
-[#3542](https://github.com/apache/iggy/issues/3542), OffsetCommit and OffsetFetch.
+Status: accepted. Replaces the first design ([#4205](https://github.com/apache/iggy/pull/4205),
+[#4289](https://github.com/apache/iggy/pull/4289)). Answers
+[#3540](https://github.com/apache/iggy/issues/3540). Used by OffsetCommit and OffsetFetch
+([#3542](https://github.com/apache/iggy/issues/3542)).
## Decision
-Store Kafka group offsets as Iggy consumer offsets, one key per partition, under a consumer
-group whose name is derived from the Kafka group id.
+Kafka group offsets go in a third Iggy consumer kind, `ConsumerKind::ExternalGroup` (wire kind 3),
+next to `Consumer` and `ConsumerGroup`.
+
+| Property | Value |
+| --- | --- |
+| Key | The partition, plus Iggy consumer group `kafka.cg.<group>` on the mapped topic, resolved by name |
+| Value | The Kafka offset as sent. Iggy does not read it |
+| Membership | Not needed. The gateway never joins the group |
+| Range check | None. 0 on an empty partition and the high watermark are both stored |
+| Polling | Refused for this kind |
+| Retention | Not held back by these offsets, as in Kafka |
+| Group deleted | Its offsets go with it |
+
+## Why not the first design
+
+It stored offsets as `ConsumerGroup` offsets. Two server rules block that:
+
+- Only a member that owns the partition can store or delete a group offset (`fence_group_offset`,
+ `core/server/src/namespace.rs:239`). The fence stays. It holds for an empty group too
+ (`partition_primary_routing.rs:888`).
+- A stored offset must be at most the last message, and an empty partition takes none
+ (`store_offset_range_error`, `core/partitions/src/iggy_partition.rs:3625`). A caught-up Kafka
+ consumer commits last + 1. A Java consumer on defaults commits 0 on an empty partition.
+
+A plain `Consumer` key passes both checks, but its name is hashed to 32 bits and deleting a group
+leaves it behind.
+
+## Gateway rules
+
+- Create group `kafka.cg.<group>` on the topic before the first commit there. Already exists is fine.
+- The prefix keeps a Kafka group apart from a native Iggy group with the same name.
+- Group id: 1 to 246 bytes, else `INVALID_GROUP_ID` (24). Iggy names stop at 255.
+- Negative offset: delete the key. Not found counts as success. Kafka consumers read any negative
+ offset as none.
+- Metadata string and leader epoch: dropped. OffsetFetch returns `""` and -1.
+- No key, unknown group or unknown topic: OffsetFetch returns -1 and no error.
+- Null topic list (OffsetFetch v2+): read every Kafka topic that has group `kafka.cg.<group>`.
+- Committed offset below the oldest kept message: Fetch reads from the first kept message.
-The issue lists three options. None of them is this one.
-
-| Option | Why not |
-| -------- | --------- |
-| A, an Iggy-backed `__consumer_offsets` topic | Rebuilds what Iggy already has. A compacted offset topic needs compaction, which Iggy does not have, so the gateway replays the whole topic at every startup |
-| B, a SQLite file on the gateway host | A second durability story, a second backup story, and offsets that do not survive moving the gateway |
-| C, in memory only | Fails the acceptance criterion in #3542, which is that offsets survive a restart |
-
-Iggy already stores a durable offset per consumer and per partition, replicated with the
-partition itself. Using it costs one call per partition on commit and one on fetch.
-
-## The key
-
-One Iggy consumer offset per Kafka `(group, topic, partition)`.
-
-- consumer kind: `ConsumerKind::ConsumerGroup`
-- consumer id: `Identifier::named("kafka.cg.<group>")`
-- stream and topic: whatever `TopicMapping` resolves the Kafka topic to
-- partition: the Kafka partition index, unchanged, because both sides number from 0
-
-The gateway calls `create_consumer_group(stream, topic, "kafka.cg.<group>")` before the first
-commit for a group on a topic. If the group does not resolve in metadata, the server rejects the
-offset write. The group has to exist first. The gateway never joins the group. Offsets are
-readable by any client, member or not.
-
-### Why the group kind and not a named consumer
-
-`ConsumerKind::Consumer` with a name looks simpler, because it needs no registration call. It is
-not. The server hashes a named consumer id to a `u32` with `XxHash32`
-(`core/server/src/dispatch/partition.rs:916`), and that hash is the offset key. Two different
-group names can collide and silently share one offset.
-
-A consumer group name resolves through metadata to a monotonic id instead. No hash, no
-collision, and `get_consumer_groups(stream, topic)` lists what exists.
-
-### Why the prefix
-
-`kafka.cg.` keeps a Kafka group called `orders` off the key that a native Iggy consumer group
-called `orders` uses. Without it the two share an offset and each one moves the other.
-
-The prefix does not make the offsets safe to poll with. That is the next section.
-
-## What is stored
-
-The Kafka committed offset, verbatim, with no conversion.
-
-The two systems mean different things by the number. Kafka commits the next offset to read.
-Iggy stores the last offset processed, and `PollingKind::Next` resumes at the stored value plus
-one (`core/partitions/src/iggy_partition.rs:3835`). A Kafka offset stored in an Iggy key is
-therefore one greater than Iggy's own convention for that key.
-
-This is inert because the gateway never polls that way. Fetch always polls with an explicit
-offset, `PollingKind::Offset`, taken from the Kafka request. Nothing in the gateway reads the
-stored value to decide where to resume. It is returned to the client on OffsetFetch and
-otherwise untouched.
-
-The rule this creates: no code path polls a `kafka.cg.*` key with `PollingKind::Next`. Doing so
-skips one record per partition. The prefix is what keeps a native Iggy consumer from doing it by
-accident.
-
-Converting on write instead, and storing the Kafka offset minus one, breaks at offset 0. It also
-makes an empty commit look the same as a commit of the first record. Storing verbatim is the
-smaller problem.
-
-### A commit of -1
-
-Kafka does not validate the sign of a committed offset. `OffsetMetadataManager` checks the
-metadata length and nothing else, so a real broker stores `-1` and hands it back on the next
-OffsetFetch, where a consumer reads it as no committed offset.
-
-Iggy cannot store that value, because `store_consumer_offset` takes a `u64`. A commit of `-1`
-therefore calls `delete_consumer_offset` on the key. What a client can observe is the same: the
-next OffsetFetch finds nothing and the gateway answers `-1`.
-
-Deleting a key that is not there returns `ConsumerOffsetNotFound` (3021). On this path that
-counts as success, because the client asked for the offset to be absent and it is absent. Any
-other negative offset is rejected with `OFFSET_OUT_OF_RANGE` (1).
-
-## OffsetFetch with no topics named
-
-OffsetFetch v2 and later let a client pass a null topic list, which asks for every offset the
-group holds. `kafka-consumer-groups.sh --describe` does this.
-
-Iggy has no lookup by consumer. Offsets are read one partition at a time
-(`core/common/src/traits/consumer_offset_client.rs:41`). The gateway answers by enumerating the
-topics in the mapped stream and querying each partition of each one.
-
-Answering also means naming each topic the way the client named it, and `TopicMapping::resolve`
-runs the wrong way. Its own doc says it is not injective, so it cannot be inverted in general.
-Two cases divide it. A topic with no override resolves to `(default_stream, kafka_topic)`, so the
-Iggy topic name is the Kafka name and reverses for free. A topic with an override needs a reverse
-index, built once at config load, which `TopicMapping::new` already makes safe by rejecting two
-overrides that share a target.
-
-What neither case covers is an unlisted Kafka topic whose name collides with an override's target
-inside the default stream. `new` rejects the shapes it can check, but the space of unlisted names
-is unbounded, so the collision is disclosed rather than prevented. An offset under a colliding
-name is reported against whichever Kafka name the reverse index holds.
+## Limits
-That is one round trip per partition on an admin call. The cost is bounded by the topic and
-partition count of one stream. This is an admin path and not a data path, so the cost is
-acceptable. It is written here so nobody discovers it in a test.
+| Limit | Value |
+| --- | --- |
+| Keys per partition, per kind | 4096, at most 262144: `partition.consumer_offsets_max` |
+| Past the limit | `TooManyConsumerOffsets` (3024). Sent as `UNKNOWN_SERVER_ERROR`, logged at `error!` |
-## What is dropped
+A group nobody uses keeps its keys until its Iggy group is deleted on each topic.
-Kafka lets a client attach a metadata string to a commit. Iggy stores a number and nothing else.
-The string is dropped on commit, and OffsetFetch returns an empty string.
+## More than one gateway
-`committed_leader_epoch` is dropped the same way. The gateway reports `-1`.
+Offsets are shared, since the key depends only on the Kafka group id. Membership is not
+([`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)).
-## Limits
+## Server work
-A partition admits a bounded number of distinct offset keys per consumer kind. The default is
-4096, set by `partition.consumer_offsets_max`
-(`core/configs/src/server_config/partition.rs:151`). The configurable ceiling is 262144. Passing
-the limit returns `TooManyConsumerOffsets` (3024). Kafka has no error code for this condition, so
-it maps to `UNKNOWN_SERVER_ERROR`, which is what `bridge/error.rs` already does where no honest
-code exists.
-
-Consumer groups and plain consumers count against separate limits, so Kafka groups do not
-compete with native Iggy consumers for the same 4096.
-
-A client cannot act on that error, so the operator has to. `partition.consumer_offsets_max` is
-named in the gateway README for that reason, and a handler that hits the limit logs the Iggy
-error at `error!` level, which is what `bridge/error.rs` already asks handlers to do wherever the
-Kafka code it sends is less specific than the Iggy error it received.
-
-An Iggy name is capped at 255 bytes (`core/common/src/lib.rs:168`), which leaves 246 for a Kafka
-group id after the prefix. A longer group id is rejected with `INVALID_GROUP_ID` (24).
-
-### Committed offsets hold back retention
-
-Iggy deletes a partition's segments, by retention or by `DeleteSegments`, only up to the lowest
-offset any consumer or consumer group has committed there
-(`core/partitions/src/iggy_partition.rs`, `min_committed_offset`). Every `kafka.cg.*` offset is
-one of those, so a Kafka group that stops consuming stops retention on every partition it
-committed on, for as long as the offset exists. Kafka itself never lets committed offsets block
-retention, and it expires them after `offsets.retention.minutes`. Iggy has no offset expiry, so an
-abandoned group's offsets stay, keep that barrier in place, and keep counting against the
-4096-key limit above until an operator deletes the Iggy consumer group `kafka.cg.<group>` on every
-topic the group committed on, since [the key](#the-key) creates one per topic. The gateway never
-does, since it does not implement `DeleteGroups` (42).
-
-The barrier reads the stored value as Iggy's last consumed offset and keeps a segment only while
-its end offset is above it, but a `kafka.cg.*` key holds the Kafka commit, the next offset to
-read (see [What is stored](#what-is-stored)). A sealed segment that ends exactly at that offset
-is therefore deletable while its last record is still unread. OffsetCommit and OffsetFetch
-([#3542](https://github.com/apache/iggy/issues/3542)) have to close this before any `kafka.cg.*`
-offset is written, for example by having the barrier subtract one for those keys.
-
-A blocked trim stays pending rather than failing. The reconciler restages a `DeleteSegments`
-trim the barrier blocks on every pass (`reconcile_segment_truncations`), and each blocked
-attempt, by that trim or by retention, logs `segment retained: blocked by committed consumer
-offset` at `warn!`, for as long as an abandoned group's offsets exist.
-
-`PurgeTopic` resets each partition to offset 0 and clears its consumer offsets, Kafka groups'
-included, so a purge leaves every group with nothing committed.
-
-## More than one gateway instance
-
-Two gateway instances that share an Iggy cluster read and write the same offset keys. The key
-comes from the Kafka group id and nothing else. Two instances that serve one group therefore
-agree on committed offsets without talking to each other.
-
-They do not agree on group membership. That belongs to the coordinator
-([#3541](https://github.com/apache/iggy/issues/3541)) and is not settled here.
-
-## Partition assignment
-
-The gateway runs no assignor: the group leader assigns on the client, whatever its configured
-strategy ([`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md#assignment-is-the-clients-job)). Offset
-storage does not depend on the choice, since a commit names the partition it is for, whichever
-member owns it.
-Iggy's balanced consumer-group assignment is not used for Kafka groups, and KIP-848's server-side
-assignment (`ConsumerGroupHeartbeat`, key 68) is out of scope.
-
-## Decision record
-
-The open question was whether to store offsets as Iggy consumer offsets keyed by group, as above,
-or take one of A, B and C from the issue. This document proposed a default: the design above
-unless an answer landed by 2026-09-22. It was merged in
-[#4205](https://github.com/apache/iggy/pull/4205) as a proposal, approved by @hubcio and
-@numinnex. No alternative was raised by that date, so it was decided by default and the design
-above is the one taken.
+- Wire: `KIND_EXTERNAL_GROUP = 3`, next to kinds 1 and 2 in `core/binary_protocol/src/primitives/consumer.rs`.
+- SDK: `ConsumerKind::ExternalGroup`. The enum is not `#[non_exhaustive]`, so this breaks the API.
+- `core/partitions`: a third offset table, with recovery, state transfer and its own key count.
+- Server: no fence and no range check for the kind, refuse polls, leave it out of
+ `min_committed_offset`, delete its keys with the group (`partition_reconciler.rs`).
## References
- Record mapping: [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md)
-- Scope and phases: [`SCOPE.md`](SCOPE.md)
+- Scope: [`SCOPE.md`](SCOPE.md)
- Offset API: `core/common/src/traits/consumer_offset_client.rs`
-- Group API: `core/common/src/traits/consumer_group_client.rs`
-- Offset key resolution: `core/server/src/dispatch/partition.rs`
diff --git a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
index 6a5c3d1cc..6b527a94b 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
@@ -283,8 +283,8 @@ impl IggyBridge {
/// Reads up to `count` messages of `partition`, from `offset` on, with the client of `slot`.
/// Hands the slot back with the result. `None` if no result comes by `deadline`.
///
- /// A plain consumer without auto commit, so Iggy stores no offset. Always an explicit
- /// offset, never `Next` (see `docs/OFFSET_STORAGE.md`).
+ /// A plain consumer without auto commit, so Iggy stores no offset. Always the request's
+ /// offset, never `Next`.
///
/// The errors are those of [`Self::probe`]. A partition the topic lacks is
/// [`BridgeError::Iggy`] here. The poll stops at `deadline` too, so a result that comes then
diff --git a/gateways/kafka/src/group/mod.rs b/gateways/kafka/src/group/mod.rs
index 07cd1a891..af42eb09d 100644
--- a/gateways/kafka/src/group/mod.rs
+++ b/gateways/kafka/src/group/mod.rs
@@ -18,20 +18,24 @@
//! In-memory coordinator for Kafka's classic consumer group protocol.
//!
//! [`GroupCoordinator`] owns every group this gateway instance coordinates and is the only
-//! module that awaits: `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup` handlers
-//! translate wire messages into the request types here, and `state` holds the synchronous state
-//! machine those requests drive.
+//! module that awaits: `FindCoordinator`/`JoinGroup`/`Heartbeat`/`LeaveGroup`/`SyncGroup`/
+//! `OffsetCommit` handlers translate wire messages into the request types here, and `state` holds
+//! the synchronous state machine those requests drive.
//!
//! Membership is process memory, not Iggy state. Two gateway instances fronting one Iggy cluster
//! therefore coordinate two independent groups under one name; see `docs/CONSUMER_GROUPS.md`.
mod state;
+pub use state::MAX_GROUP_ID_BYTES;
+
use std::collections::HashMap;
use std::time::Duration;
use bytes::Bytes;
-use kafka_protocol::messages::{JoinGroupRequest, LeaveGroupRequest, SyncGroupRequest};
+use kafka_protocol::messages::{
+ JoinGroupRequest, LeaveGroupRequest, OffsetCommitRequest, SyncGroupRequest,
+};
use kafka_protocol::protocol::StrBytes;
use tokio::sync::{Mutex, watch};
use tokio::time::Instant;
@@ -338,6 +342,27 @@ impl From<(&LeavingMember, i16)> for LeftMember {
}
}
+/// Who an `OffsetCommit` says it is.
+#[derive(Debug, Clone)]
+pub struct CommitRequest {
+ pub group_id: StrBytes,
+ /// Negative for a commit made outside the group protocol, as admin tools send.
+ pub generation_id: i32,
+ pub member_id: StrBytes,
+ pub group_instance_id: Option<StrBytes>,
+}
+
+impl From<&OffsetCommitRequest> for CommitRequest {
+ fn from(request: &OffsetCommitRequest) -> Self {
+ Self {
+ group_id: request.group_id.0.clone(),
+ generation_id: request.generation_id_or_member_epoch,
+ member_id: request.member_id.clone(),
+ group_instance_id: request.group_instance_id.clone(),
+ }
+ }
+}
+
/// Every consumer group this gateway instance coordinates.
///
/// There is no timer task. A request that touches a group first expires whatever is overdue in
@@ -451,6 +476,13 @@ impl GroupCoordinator {
state::leave_step(&mut groups, request, Instant::now())
}
+ /// Whether `request` may commit offsets for its group. A member's commit also refreshes its
+ /// session. Never parks.
+ pub async fn validate_commit(&self, request: &CommitRequest) -> i16 {
+ let mut groups = self.groups.lock().await;
+ state::commit_step(&mut groups, request, Instant::now())
+ }
+
/// Sleeps until the group changes or `wake_at` passes. `false` means the gateway is draining.
async fn wait_until(&self, mut receiver: watch::Receiver<u64>, wake_at: Instant) -> bool {
tokio::select! {
diff --git a/gateways/kafka/src/group/state.rs b/gateways/kafka/src/group/state.rs
index 1597fb922..e5335bfc8 100644
--- a/gateways/kafka/src/group/state.rs
+++ b/gateways/kafka/src/group/state.rs
@@ -35,8 +35,8 @@ use tokio::time::Instant;
use uuid::Uuid;
use crate::group::{
- GroupCoordinatorConfig, JoinRequest, JoinResult, JoinedMember, LeaveRequest, LeaveResult,
- LeavingMember, LeftMember, SyncRequest, SyncResult, owned_str,
+ CommitRequest, GroupCoordinatorConfig, JoinRequest, JoinResult, JoinedMember, LeaveRequest,
+ LeaveResult, LeavingMember, LeftMember, SyncRequest, SyncResult, owned_str,
};
use crate::protocol::api::{
ERROR_COORDINATOR_NOT_AVAILABLE, ERROR_FENCED_INSTANCE_ID, ERROR_GROUP_MAX_SIZE_REACHED,
@@ -1112,6 +1112,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 request.group_id.is_empty() || request.group_id.len() > MAX_GROUP_ID_BYTES {
+ 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
+}
+
fn admit(
group: &mut GroupState,
config: &GroupCoordinatorConfig,
@@ -2953,4 +3008,181 @@ mod tests {
assert_eq!(result.error, ERROR_INVALID_GROUP_ID);
assert!(result.members.is_empty());
}
+
+ fn commit(member_id: &str, generation_id: i32) -> CommitRequest {
+ CommitRequest {
+ group_id: group_id(),
+ generation_id,
+ member_id: StrBytes::from_string(member_id.to_owned()),
+ group_instance_id: None,
+ }
+ }
+
+ /// Admin tools and `assign()` consumers commit this way.
+ #[test]
+ fn given_no_group_when_committing_without_a_generation_should_accept_and_not_create_it() {
+ let mut groups = Groups::new();
+
+ let error = commit_step(&mut groups, &commit("", -1), Instant::now());
+
+ assert_eq!(error, ERROR_NONE);
+ assert!(groups.is_empty());
+ }
+
+ /// What every member meets after a gateway restart. `UNKNOWN_MEMBER_ID` makes it rejoin.
+ #[test]
+ fn given_no_group_when_committing_with_a_generation_should_answer_unknown_member_id() {
+ let mut groups = Groups::new();
+
+ let error = commit_step(&mut groups, &commit("m-1", 3), Instant::now());
+
+ assert_eq!(error, ERROR_UNKNOWN_MEMBER_ID);
+ }
+
+ #[test]
+ fn given_only_pending_ids_when_committing_without_a_generation_should_accept() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let _ = join_step(&mut groups, &config, &pending_request(), now);
+
+ assert_eq!(commit_step(&mut groups, &commit("", -1), now), ERROR_NONE);
+ }
+
+ /// Without the refresh a member that only commits, and so heartbeats less, is evicted.
+ #[test]
+ fn given_a_stable_member_when_committing_should_refresh_its_session() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let (_leader, follower) = stable_two_members(&mut groups, &config, now);
+ let generation = groups[&group_id()].generation_id;
+
+ let error = commit_step(
+ &mut groups,
+ &commit(follower.as_str(), generation),
+ now + Duration::from_secs(9),
+ );
+ let heartbeat = heartbeat_step(
+ &mut groups,
+ &group_id(),
+ generation,
+ &follower,
+ now + Duration::from_secs(15),
+ );
+
+ assert_eq!(error, ERROR_NONE);
+ assert_ne!(heartbeat, ERROR_UNKNOWN_MEMBER_ID);
+ assert!(groups[&group_id()].members.contains_key(&follower));
+ }
+
+ /// The consumer commits before it rejoins, while the old generation is still current.
+ #[test]
+ fn given_a_preparing_rebalance_when_a_member_commits_its_generation_should_accept() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let (leader, follower) = stable_two_members(&mut groups, &config, now);
+ let generation = groups[&group_id()].generation_id;
+ let _ = leave(&mut groups, &[(follower.as_str(), None)], now);
+ assert_eq!(groups[&group_id()].phase, Phase::PreparingRebalance);
+
+ let error = commit_step(&mut groups, &commit(leader.as_str(), generation), now);
+
+ assert_eq!(error, ERROR_NONE);
+ }
+
+ #[test]
+ fn given_a_completing_rebalance_when_a_member_commits_should_answer_rebalance_in_progress() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let (_leader, follower) = two_members(&mut groups, &config, &["x"], &["x"], now);
+ let group = &groups[&group_id()];
+ assert_eq!(group.phase, Phase::CompletingRebalance);
+ let generation = group.generation_id;
+
+ let error = commit_step(&mut groups, &commit(follower.as_str(), generation), now);
+
+ assert_eq!(error, ERROR_REBALANCE_IN_PROGRESS);
+ }
+
+ #[test]
+ fn given_a_group_with_members_when_committing_should_check_member_and_generation() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let (leader, _follower) = stable_two_members(&mut groups, &config, now);
+ let generation = groups[&group_id()].generation_id;
+ let mut check = |request: &CommitRequest| commit_step(&mut groups, request, now);
+
+ assert_eq!(check(&commit("", -1)), ERROR_UNKNOWN_MEMBER_ID);
+ assert_eq!(
+ check(&commit("stranger", generation)),
+ ERROR_UNKNOWN_MEMBER_ID
+ );
+ assert_eq!(
+ check(&commit(leader.as_str(), generation - 1)),
+ ERROR_ILLEGAL_GENERATION
+ );
+ assert_eq!(
+ check(&commit(leader.as_str(), -1)),
+ ERROR_ILLEGAL_GENERATION
+ );
+ assert_eq!(check(&commit(leader.as_str(), generation)), ERROR_NONE);
+ }
+
+ #[test]
+ fn given_a_static_member_when_committing_should_check_its_instance_id() {
+ let config = config();
+ let mut groups = Groups::new();
+ let now = Instant::now();
+ let member = member_id_of(&join_step(
+ &mut groups,
+ &config,
+ &static_request("i-1"),
+ now,
+ ));
+ let generation = groups[&group_id()].generation_id;
+ let _ = sync_step(
+ &mut groups,
+ &config,
+ &sync_request(&member, generation),
+ now,
+ );
+ let with_instance = |member_id: &str, instance_id: &str| CommitRequest {
+ group_instance_id: Some(StrBytes::from_string(instance_id.to_owned())),
+ ..commit(member_id, generation)
+ };
+
+ assert_eq!(
+ commit_step(&mut groups, &with_instance("other", "i-1"), now),
+ ERROR_FENCED_INSTANCE_ID
+ );
+ assert_eq!(
+ commit_step(&mut groups, &with_instance(member.as_str(), "i-2"), now),
+ ERROR_UNKNOWN_MEMBER_ID
+ );
+ assert_eq!(
+ commit_step(&mut groups, &with_instance(member.as_str(), "i-1"), now),
+ ERROR_NONE
+ );
+ }
+
+ #[test]
+ fn given_an_invalid_group_id_when_committing_should_answer_invalid_group_id() {
+ let mut groups = Groups::new();
+ let mut request = commit("", -1);
+ request.group_id = StrBytes::new();
+ assert_eq!(
+ commit_step(&mut groups, &request, Instant::now()),
+ ERROR_INVALID_GROUP_ID
+ );
+
+ request.group_id = StrBytes::from_string("g".repeat(MAX_GROUP_ID_BYTES + 1));
+ assert_eq!(
+ commit_step(&mut groups, &request, Instant::now()),
+ ERROR_INVALID_GROUP_ID
+ );
+ }
}
diff --git a/gateways/kafka/src/protocol/api.rs b/gateways/kafka/src/protocol/api.rs
index c2478ecbf..5857ed707 100644
--- a/gateways/kafka/src/protocol/api.rs
+++ b/gateways/kafka/src/protocol/api.rs
@@ -39,8 +39,8 @@ use crate::protocol::bounds_guard::{
use crate::protocol::handlers::init_producer_id::ProducerIdAllocator;
use crate::protocol::handlers::{
api_versions, create_topics, decode_guarded, dispatch, fetch, find_coordinator, heartbeat,
- init_producer_id, join_group, leave_group, list_offsets, metadata, produce, respond_or_close,
- sync_group,
+ init_producer_id, join_group, leave_group, list_offsets, metadata, offset_commit, offset_fetch,
+ produce, respond_or_close, sync_group,
};
use crate::protocol::probe_board::ProbeBoard;
use crate::protocol::sasl::{
@@ -51,6 +51,8 @@ pub const API_KEY_PRODUCE: i16 = 0;
pub const API_KEY_FETCH: i16 = 1;
pub const API_KEY_LIST_OFFSETS: i16 = 2;
pub const API_KEY_METADATA: i16 = 3;
+pub const API_KEY_OFFSET_COMMIT: i16 = 8;
+pub const API_KEY_OFFSET_FETCH: i16 = 9;
pub const API_KEY_FIND_COORDINATOR: i16 = 10;
pub const API_KEY_JOIN_GROUP: i16 = 11;
pub const API_KEY_HEARTBEAT: i16 = 12;
@@ -90,6 +92,8 @@ pub const ERROR_INVALID_TOPIC_EXCEPTION: i16 = ResponseError::InvalidTopicExcept
/// Produce: `acks` is not 0, 1 or -1. A conformant client never sends one, since `acks` comes
/// from validated configuration rather than from application input.
pub const ERROR_INVALID_REQUIRED_ACKS: i16 = ResponseError::InvalidRequiredAcks.code();
+/// Retriable, and the client keeps its coordinator. An offset call to Iggy that may work on retry.
+pub const ERROR_COORDINATOR_LOAD_IN_PROGRESS: i16 = ResponseError::CoordinatorLoadInProgress.code();
/// Retriable. Sent when this coordinator is at one of its `GroupCoordinatorConfig` capacity
/// caps: the client should back off and retry rather than treat the group as unusable.
pub const ERROR_COORDINATOR_NOT_AVAILABLE: i16 = 15;
@@ -270,6 +274,8 @@ static SUPPORTED_RANGES: &[ApiVersionRange] = &[
fetch::RANGE,
list_offsets::RANGE,
metadata::RANGE,
+ offset_commit::RANGE,
+ offset_fetch::RANGE,
api_versions::RANGE,
create_topics::RANGE,
init_producer_id::RANGE,
diff --git a/gateways/kafka/src/protocol/bounds_guard.rs b/gateways/kafka/src/protocol/bounds_guard.rs
index ae265b8ed..9f1faa3bd 100644
--- a/gateways/kafka/src/protocol/bounds_guard.rs
+++ b/gateways/kafka/src/protocol/bounds_guard.rs
@@ -1039,6 +1039,132 @@ pub fn validate_sync_group_shape(version: i16, body: &Bytes, max_frame_size: usi
Ok(())
}
+/// Mirrors the field order `OffsetCommitRequest::decode` walks.
+///
+/// # Errors
+///
+/// Returns an error when a declared array/string length cannot fit in the bytes remaining in the
+/// frame, or the body is truncated or malformed in a way that cannot be walked.
+pub fn validate_offset_commit_shape(
+ version: i16,
+ body: &Bytes,
+ max_frame_size: usize,
+) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), max_frame_size);
+ let flexible = version >= 8;
+
+ if flexible {
+ c.compact_string(false)?;
+ } else {
+ c.legacy_string(false)?;
+ }
+ let _generation_id_or_member_epoch = c.read_i32()?;
+ if flexible {
+ c.compact_string(false)?;
+ } else {
+ c.legacy_string(false)?;
+ }
+ if version >= 7 {
+ if flexible {
+ c.compact_string(true)?;
+ } else {
+ c.legacy_string(true)?;
+ }
+ }
+ if version <= 4 {
+ let _retention_time_ms = c.read_i64()?;
+ }
+
+ let topics_count = if flexible {
+ c.compact_array_count()?
+ } else {
+ c.legacy_array_count()?
+ };
+ for _ in 0..topics_count {
+ if flexible {
+ c.compact_string(false)?;
+ } else {
+ c.legacy_string(false)?;
+ }
+ let partitions_count = if flexible {
+ c.compact_array_count()?
+ } else {
+ c.legacy_array_count()?
+ };
+ for _ in 0..partitions_count {
+ let _partition_index = c.read_i32()?;
+ let _committed_offset = c.read_i64()?;
+ if version >= 6 {
+ let _committed_leader_epoch = c.read_i32()?;
+ }
+ if flexible {
+ c.compact_string(true)?;
+ c.tagged_fields()?;
+ } else {
+ c.legacy_string(true)?;
+ }
+ }
+ if flexible {
+ c.tagged_fields()?;
+ }
+ }
+ if flexible {
+ c.tagged_fields()?;
+ }
+ Ok(())
+}
+
+/// Mirrors the field order `OffsetFetchRequest::decode` walks: one group up to v7, a `groups`
+/// array from v8.
+///
+/// # Errors
+///
+/// Returns an error when a declared array/string length cannot fit in the bytes remaining in the
+/// frame, or the body is truncated or malformed in a way that cannot be walked.
+pub fn validate_offset_fetch_shape(
+ version: i16,
+ body: &Bytes,
+ max_frame_size: usize,
+) -> Result<()> {
+ let mut c = ShapeCursor::new(body.clone(), max_frame_size);
+ let flexible = version >= 6;
+
+ if version <= 7 {
+ if flexible {
+ c.compact_string(false)?;
+ } else {
+ c.legacy_string(false)?;
+ }
+ let topics_count = if flexible {
+ c.compact_array_count_nullable()?
+ } else if version >= 2 {
+ c.legacy_array_count_nullable()?
+ } else {
+ c.legacy_array_count()?
+ };
+ walk_offset_fetch_topics(&mut c, topics_count, flexible)?;
+ } else {
+ let groups_count = c.compact_array_count()?;
+ for _ in 0..groups_count {
+ c.compact_string(false)?;
+ if version >= 9 {
+ c.compact_string(true)?;
+ let _member_epoch = c.read_i32()?;
+ }
+ let topics_count = c.compact_array_count_nullable()?;
+ walk_offset_fetch_topics(&mut c, topics_count, flexible)?;
+ c.tagged_fields()?;
+ }
+ }
+ if version >= 7 {
+ let _require_stable = c.read_bool()?;
+ }
+ if flexible {
+ c.tagged_fields()?;
+ }
+ Ok(())
+}
+
/// Cap on a whole `SaslAuthenticate` body.
///
/// The body is one length-prefixed blob, so capping it is the same as capping `auth_bytes`.
@@ -1125,10 +1251,34 @@ pub fn validate_sasl_authenticate_shape(version: i16, body: &Bytes) -> Result<()
Ok(())
}
+/// One `OffsetFetch` topic list: each topic's name and its partition indexes.
+fn walk_offset_fetch_topics(
+ c: &mut ShapeCursor,
+ topics_count: usize,
+ flexible: bool,
+) -> Result<()> {
+ for _ in 0..topics_count {
+ let partitions_count = if flexible {
+ c.compact_string(false)?;
+ c.compact_array_count()?
+ } else {
+ c.legacy_string(false)?;
+ c.legacy_array_count()?
+ };
+ for _ in 0..partitions_count {
+ let _partition_index = c.read_i32()?;
+ }
+ if flexible {
+ c.tagged_fields()?;
+ }
+ }
+ Ok(())
+}
+
#[cfg(test)]
mod tests {
use bytes::BytesMut;
- use kafka_protocol::messages::LeaveGroupRequest;
+ use kafka_protocol::messages::{LeaveGroupRequest, OffsetCommitRequest, OffsetFetchRequest};
use super::*;
use crate::protocol::handlers::decode_exhaustive;
@@ -1623,4 +1773,116 @@ mod tests {
assert!(validate_leave_group_shape(4, &body, 8 * 1024 * 1024).is_ok());
assert!(validate_leave_group_shape(4, &body, 1_024).is_err());
}
+
+ #[test]
+ fn offset_commit_v2_null_metadata_accepted() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x01, b'g', // group_id
+ 0xFF, 0xFF, 0xFF, 0xFF, // generation_id
+ 0x00, 0x00, // member_id
+ 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, // retention_time_ms
+ 0x00, 0x00, 0x00, 0x01, // topics: 1
+ 0x00, 0x01, b't', // name
+ 0x00, 0x00, 0x00, 0x01, // partitions: 1
+ 0x00, 0x00, 0x00, 0x00, // partition_index
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, // committed_offset
+ 0xFF, 0xFF, // committed_metadata null
+ ]);
+ assert!(validate_offset_commit_shape(2, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetCommitRequest>(2, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_commit_v8_flexible_body_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, b'g', // group_id
+ 0x00, 0x00, 0x00, 0x01, // generation_id
+ 0x02, b'm', // member_id
+ 0x00, // group_instance_id null
+ 0x02, // topics: 1
+ 0x02, b't', // name
+ 0x02, // partitions: 1
+ 0x00, 0x00, 0x00, 0x00, // partition_index
+ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, // committed_offset
+ 0xFF, 0xFF, 0xFF, 0xFF, // committed_leader_epoch
+ 0x01, // committed_metadata ""
+ 0x00, // partition tagged fields
+ 0x00, // topic tagged fields
+ 0x00, // request tagged fields
+ ]);
+ assert!(validate_offset_commit_shape(8, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetCommitRequest>(8, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_commit_v9_huge_partitions_count_rejected() {
+ let body = Bytes::from_static(&[
+ 0x02, b'g', // group_id
+ 0x00, 0x00, 0x00, 0x01, // generation_id
+ 0x02, b'm', // member_id
+ 0x00, // group_instance_id null
+ 0x02, // topics: 1
+ 0x02, b't', // name
+ 0xFF, 0xFF, 0xFF, 0xFF, 0x0F, // partitions count
+ ]);
+ assert!(validate_offset_commit_shape(9, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn offset_fetch_v1_partition_indexes_accepted() {
+ let body = Bytes::from_static(&[
+ 0x00, 0x01, b'g', // group_id
+ 0x00, 0x00, 0x00, 0x01, // topics: 1
+ 0x00, 0x01, b't', // name
+ 0x00, 0x00, 0x00, 0x02, // partition_indexes: 2
+ 0x00, 0x00, 0x00, 0x00, // 0
+ 0x00, 0x00, 0x00, 0x01, // 1
+ ]);
+ assert!(validate_offset_fetch_shape(1, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetFetchRequest>(1, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_fetch_v7_null_topics_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, b'g', // group_id
+ 0x00, // topics null: every committed offset
+ 0x01, // require_stable
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_offset_fetch_shape(7, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetFetchRequest>(7, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_fetch_v9_group_with_member_accepted() {
+ let body = Bytes::from_static(&[
+ 0x02, // groups: 1
+ 0x02, b'g', // group_id
+ 0x00, // member_id null
+ 0xFF, 0xFF, 0xFF, 0xFF, // member_epoch
+ 0x02, // topics: 1
+ 0x02, b't', // name
+ 0x02, // partition_indexes: 1
+ 0x00, 0x00, 0x00, 0x00, // 0
+ 0x00, // topic tagged fields
+ 0x00, // group tagged fields
+ 0x00, // require_stable
+ 0x00, // tagged fields
+ ]);
+ assert!(validate_offset_fetch_shape(9, &body, TEST_MAX_FRAME_SIZE).is_ok());
+ decode_exhaustive::<OffsetFetchRequest>(9, body).expect("kafka_protocol agrees");
+ }
+
+ #[test]
+ fn offset_fetch_v6_huge_topics_count_rejected() {
+ let body = Bytes::from_static(&[0x02, b'g', 0xFF, 0xFF, 0xFF, 0xFF, 0x0F]);
+ assert!(validate_offset_fetch_shape(6, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
+
+ #[test]
+ fn offset_fetch_v1_null_topics_rejected() {
+ let body = Bytes::from_static(&[0x00, 0x01, b'g', 0xFF, 0xFF, 0xFF, 0xFF]);
+ assert!(validate_offset_fetch_shape(1, &body, TEST_MAX_FRAME_SIZE).is_err());
+ }
}
diff --git a/gateways/kafka/src/protocol/handlers/mod.rs b/gateways/kafka/src/protocol/handlers/mod.rs
index 4ecfbe9fe..05d52ae26 100644
--- a/gateways/kafka/src/protocol/handlers/mod.rs
+++ b/gateways/kafka/src/protocol/handlers/mod.rs
@@ -34,6 +34,8 @@ pub mod join_group;
pub mod leave_group;
pub mod list_offsets;
pub mod metadata;
+pub mod offset_commit;
+pub mod offset_fetch;
pub mod produce;
pub mod sync_group;
@@ -46,9 +48,10 @@ use crate::error::{KafkaProtocolError, Result};
use crate::protocol::api::{
API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_FETCH, API_KEY_FIND_COORDINATOR,
API_KEY_HEARTBEAT, API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
- API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_PRODUCE, API_KEY_SYNC_GROUP, ConnectionState,
- ERROR_INVALID_REQUEST, ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome,
- is_supported_version, supported_max_version,
+ API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH,
+ API_KEY_PRODUCE, API_KEY_SYNC_GROUP, ConnectionState, ERROR_INVALID_REQUEST,
+ ERROR_UNSUPPORTED_VERSION, GatewayState, HandleOutcome, is_supported_version,
+ supported_max_version,
};
/// Record encodes and decodes of this many bytes or more run off the async worker.
@@ -84,6 +87,8 @@ pub async fn dispatch(
API_KEY_FETCH => fetch::handle(state, connection, api_version, body).await,
API_KEY_LIST_OFFSETS => list_offsets::handle(state, api_version, body).await,
API_KEY_METADATA => metadata::handle(state, api_version, body).await,
+ API_KEY_OFFSET_COMMIT => offset_commit::handle(state, api_version, body).await,
+ API_KEY_OFFSET_FETCH => offset_fetch::handle(state, api_version, body).await,
API_KEY_API_VERSIONS => api_versions::handle(state, api_version, body).await,
API_KEY_CREATE_TOPICS => create_topics::handle(state, api_version, body).await,
API_KEY_FIND_COORDINATOR => find_coordinator::handle(state, api_version, body).await,
diff --git a/gateways/kafka/src/protocol/handlers/offset_commit.rs b/gateways/kafka/src/protocol/handlers/offset_commit.rs
new file mode 100644
index 000000000..454eda271
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/offset_commit.rs
@@ -0,0 +1,234 @@
+// 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 crate::error::Result;
+use crate::group::CommitRequest;
+use crate::protocol::api::{
+ API_KEY_OFFSET_COMMIT, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_NONE,
+ GatewayState, HandleOutcome, is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_commit_shape;
+use crate::protocol::handlers::{decode_guarded, encode_message, 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;
+
+/// One code per requested partition, in request order.
+pub type PartitionCodes = Vec<Vec<i16>>;
+
+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 code = if error == ERROR_NONE {
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS
+ } else {
+ error
+ };
+ respond_or_close(
+ encode_response(api_version, &request, &same_code(&request, code)),
+ "OffsetCommit",
+ )
+}
+
+/// `code` for every requested partition.
+#[must_use]
+pub fn same_code(request: &OffsetCommitRequest, code: i16) -> PartitionCodes {
+ request
+ .topics
+ .iter()
+ .map(|topic| vec![code; topic.partitions.len()])
+ .collect()
+}
+
+/// Echoes every requested topic and partition, in request order, with its code.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+pub fn encode_response(
+ version: i16,
+ request: &OffsetCommitRequest,
+ codes: &PartitionCodes,
+) -> Result<Bytes> {
+ let topics = request
+ .topics
+ .iter()
+ .zip(codes)
+ .map(|(topic, codes)| {
+ OffsetCommitResponseTopic::default()
+ .with_name(topic.name.clone())
+ .with_partitions(
+ topic
+ .partitions
+ .iter()
+ .zip(codes)
+ .map(|(partition, code)| {
+ OffsetCommitResponsePartition::default()
+ .with_partition_index(partition.partition_index)
+ .with_error_code(*code)
+ })
+ .collect(),
+ )
+ })
+ .collect();
+ let echoed: usize = request
+ .topics
+ .iter()
+ .map(|topic| topic.name.0.len() + topic.partitions.len() * PER_PARTITION_BYTES)
+ .sum();
+ encode_message(
+ &OffsetCommitResponse::default().with_topics(topics),
+ version,
+ 16 + echoed,
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use bytes::BytesMut;
+ use kafka_protocol::messages::offset_commit_request::{
+ OffsetCommitRequestPartition, OffsetCommitRequestTopic,
+ };
+ use kafka_protocol::messages::{GroupId, TopicName};
+ use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
+
+ use super::*;
+ use crate::protocol::api::{BrokerAdvertise, ERROR_UNKNOWN_MEMBER_ID};
+
+ fn request(generation_id: i32, member_id: &'static str) -> OffsetCommitRequest {
+ OffsetCommitRequest::default()
+ .with_group_id(GroupId(StrBytes::from_static_str("g")))
+ .with_generation_id_or_member_epoch(generation_id)
+ .with_member_id(StrBytes::from_static_str(member_id))
+ .with_topics(vec![
+ OffsetCommitRequestTopic::default()
+ .with_name(TopicName(StrBytes::from_static_str("orders")))
+ .with_partitions(vec![
+ OffsetCommitRequestPartition::default()
+ .with_partition_index(0)
+ .with_committed_offset(5),
+ OffsetCommitRequestPartition::default()
+ .with_partition_index(1)
+ .with_committed_offset(7),
+ ]),
+ ])
+ }
+
+ fn body(version: i16, request: &OffsetCommitRequest) -> Bytes {
+ let mut buf = BytesMut::new();
+ request.encode(&mut buf, version).unwrap();
+ buf.freeze()
+ }
+
+ async fn codes(version: i16, request: &OffsetCommitRequest) -> Vec<(String, i32, i16)> {
+ let state = GatewayState::stub(BrokerAdvertise::default(), 8 * 1024 * 1024);
+ let mut response = handle(&state, version, body(version, request))
+ .await
+ .expect_response("OffsetCommit answers");
+ let response = OffsetCommitResponse::decode(&mut response, version).unwrap();
+ response
+ .topics
+ .iter()
+ .flat_map(|topic| {
+ topic.partitions.iter().map(|partition| {
+ (
+ topic.name.0.to_string(),
+ partition.partition_index,
+ partition.error_code,
+ )
+ })
+ })
+ .collect()
+ }
+
+ /// After a gateway restart every group is gone. A member's commit must make it rejoin.
+ #[tokio::test]
+ async fn given_an_unknown_group_when_a_member_commits_should_answer_unknown_member_id() {
+ for version in [2, 8, 9] {
+ assert_eq!(
+ codes(version, &request(3, "m-1")).await,
+ vec![
+ ("orders".to_string(), 0, ERROR_UNKNOWN_MEMBER_ID),
+ ("orders".to_string(), 1, ERROR_UNKNOWN_MEMBER_ID),
+ ],
+ "v{version}"
+ );
+ }
+ }
+
+ #[tokio::test]
+ async fn given_no_bridge_when_committing_should_answer_retriable_on_every_partition() {
+ assert_eq!(
+ codes(9, &request(-1, "")).await,
+ vec![
+ ("orders".to_string(), 0, ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ ("orders".to_string(), 1, ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ ]
+ );
+ }
+
+ /// A partition left out of the answer reads as committed, so nothing short of a full answer
+ /// may go back.
+ #[tokio::test]
+ async fn given_an_unreadable_request_should_close_rather_than_answer() {
+ let state = GatewayState::stub(BrokerAdvertise::default(), 8 * 1024 * 1024);
+
+ let truncated = body(9, &request(-1, "")).slice(..5);
+ assert!(handle(&state, 9, truncated).await.is_close());
+ assert!(
+ handle(&state, 10, body(9, &request(-1, "")))
+ .await
+ .is_close()
+ );
+ }
+}
diff --git a/gateways/kafka/src/protocol/handlers/offset_fetch.rs b/gateways/kafka/src/protocol/handlers/offset_fetch.rs
new file mode 100644
index 000000000..9c13c14fc
--- /dev/null
+++ b/gateways/kafka/src/protocol/handlers/offset_fetch.rs
@@ -0,0 +1,362 @@
+// 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 crate::error::Result;
+use crate::group::MAX_GROUP_ID_BYTES;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{decode_guarded, encode_message, 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)]
+pub struct GroupQuery {
+ pub group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ pub topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ #[must_use]
+ pub 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)]
+pub struct GroupAnswer {
+ pub error: i16,
+ pub topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ #[must_use]
+ pub const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+pub struct TopicAnswer {
+ pub name: TopicName,
+ pub partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PartitionAnswer {
+ pub index: i32,
+ pub offset: i64,
+ pub error: i16,
+}
+
+#[expect(
+ clippy::unused_async,
+ reason = "the shared handler signature, kept until this handler awaits the bridge"
+)]
+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 answers: Vec<GroupAnswer> = queries.iter().map(answer).collect();
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+fn answer(query: &GroupQuery) -> GroupAnswer {
+ if query.group_id.is_empty() || query.group_id.len() > MAX_GROUP_ID_BYTES {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS)
+}
+
+/// 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`.
+pub 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()
+ };
+ let topics = topics
+ .into_iter()
+ .map(|topic| {
+ OffsetFetchResponseTopic::default()
+ .with_name(topic.name)
+ .with_partitions(
+ topic
+ .partitions
+ .iter()
+ .map(|partition| {
+ OffsetFetchResponsePartition::default()
+ .with_partition_index(partition.index)
+ .with_committed_offset(partition.offset)
+ .with_error_code(partition.error)
+ })
+ .collect(),
+ )
+ })
+ .collect();
+ let mut response = OffsetFetchResponse::default().with_topics(topics);
+ if version >= FIRST_GROUP_ERROR_VERSION {
+ response = response.with_error_code(answer.error);
+ }
+ encode_message(&response, version, capacity)
+}
+
+#[cfg(test)]
+mod tests {
+ use bytes::BytesMut;
+ use kafka_protocol::messages::offset_fetch_request::{
+ OffsetFetchRequestGroup, OffsetFetchRequestTopic, OffsetFetchRequestTopics,
+ };
+ use kafka_protocol::protocol::{Decodable, Encodable};
+
+ use super::*;
+ use crate::protocol::api::BrokerAdvertise;
+
+ fn topic(name: &'static str) -> TopicName {
+ TopicName(StrBytes::from_static_str(name))
+ }
+
+ fn single_group(group_id: &'static str) -> OffsetFetchRequest {
+ OffsetFetchRequest::default()
+ .with_group_id(GroupId(StrBytes::from_static_str(group_id)))
+ .with_topics(Some(vec![
+ OffsetFetchRequestTopic::default()
+ .with_name(topic("orders"))
+ .with_partition_indexes(vec![0, 1]),
+ ]))
+ }
+
+ async fn fetch(version: i16, request: &OffsetFetchRequest) -> OffsetFetchResponse {
+ let state = GatewayState::stub(BrokerAdvertise::default(), 8 * 1024 * 1024);
+ let mut body = BytesMut::new();
+ request.encode(&mut body, version).unwrap();
+ let mut response = handle(&state, version, body.freeze())
+ .await
+ .expect_response("OffsetFetch answers");
+ OffsetFetchResponse::decode(&mut response, version).unwrap()
+ }
+
+ #[tokio::test]
+ async fn given_no_bridge_at_v1_should_put_the_error_on_every_partition() {
+ let response = fetch(1, &single_group("g")).await;
+
+ let partitions = &response.topics[0].partitions;
+ assert_eq!(partitions.len(), 2);
+ for partition in partitions {
+ assert_eq!(partition.error_code, ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ assert_eq!(partition.committed_offset, UNKNOWN_OFFSET);
+ }
+ }
+
+ #[tokio::test]
+ async fn given_no_bridge_at_v7_should_answer_a_group_error_and_no_topics() {
+ let response = fetch(7, &single_group("g")).await;
+
+ assert_eq!(response.error_code, ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ assert!(response.topics.is_empty());
+ }
+
+ /// The Java client looks each group up by id in a v8 answer and throws if one is missing.
+ #[tokio::test]
+ async fn given_several_groups_at_v9_should_echo_each_group_id() {
+ let request = OffsetFetchRequest::default().with_groups(vec![
+ OffsetFetchRequestGroup::default()
+ .with_group_id(GroupId(StrBytes::from_static_str("a")))
+ .with_topics(Some(vec![
+ OffsetFetchRequestTopics::default()
+ .with_name(topic("orders"))
+ .with_partition_indexes(vec![0]),
+ ])),
+ OffsetFetchRequestGroup::default().with_group_id(GroupId(StrBytes::new())),
+ ]);
+
+ let response = fetch(9, &request).await;
+
+ let groups: Vec<(&str, i16)> = response
+ .groups
+ .iter()
+ .map(|group| (group.group_id.0.as_str(), group.error_code))
+ .collect();
+ assert_eq!(
+ groups,
+ vec![
+ ("a", ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ ("", ERROR_INVALID_GROUP_ID)
+ ]
+ );
+ }
+
+ #[tokio::test]
+ async fn given_an_unreadable_request_should_close_rather_than_answer() {
+ let state = GatewayState::stub(BrokerAdvertise::default(), 8 * 1024 * 1024);
+ let mut body = BytesMut::new();
+ single_group("g").encode(&mut body, 6).unwrap();
+ let body = body.freeze();
+
+ assert!(handle(&state, 6, body.slice(..3)).await.is_close());
+ assert!(handle(&state, 10, body).await.is_close());
+ }
+}
diff --git a/gateways/kafka/tests/common/scope.rs b/gateways/kafka/tests/common/scope.rs
index 2926b0751..750c8c5cc 100644
--- a/gateways/kafka/tests/common/scope.rs
+++ b/gateways/kafka/tests/common/scope.rs
@@ -30,6 +30,8 @@ pub const SCOPED_API_KEYS: &[(i16, &str, i16, i16)] = &[
(1, "Fetch", 4, 12),
(2, "ListOffsets", 1, 6),
(3, "Metadata", 0, 9),
+ (8, "OffsetCommit", 2, 9),
+ (9, "OffsetFetch", 1, 9),
(10, "FindCoordinator", 0, 4),
(11, "JoinGroup", 0, 9),
(12, "Heartbeat", 0, 4),
diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs
index 89db5f219..a9a525292 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -32,8 +32,6 @@ use super::codec::Encoder;
/// Consumer-group and admin keys explicitly out of scope in SCOPE.md.
pub const OUT_OF_SCOPE_API_KEYS: &[(i16, &str)] = &[
- (8, "OffsetCommit"),
- (9, "OffsetFetch"),
(15, "DescribeGroups"),
(16, "ListGroups"),
(20, "DeleteTopics"),
@@ -599,6 +597,75 @@ pub fn build_heartbeat_request(
enc.freeze()
}
+/// `OffsetCommit` request for partition 0 of `topic`, made outside the group protocol.
+pub fn build_offset_commit_request(
+ version: i16,
+ group_id: &str,
+ topic: &str,
+ offset: i64,
+) -> Bytes {
+ let flexible = version >= 8;
+ let mut enc = Encoder::with_capacity(64);
+
+ write_string(&mut enc, flexible, Some(group_id));
+ enc.write_i32(-1); // generation_id_or_member_epoch
+ write_string(&mut enc, flexible, Some("")); // member_id
+ if version >= 7 {
+ write_string(&mut enc, flexible, None); // group_instance_id
+ }
+ if version <= 4 {
+ enc.write_i64(-1); // retention_time_ms
+ }
+ write_array_count(&mut enc, flexible, 1);
+ write_string(&mut enc, flexible, Some(topic));
+ write_array_count(&mut enc, flexible, 1);
+ enc.write_i32(0); // partition_index
+ enc.write_i64(offset);
+ if version >= 6 {
+ enc.write_i32(-1); // committed_leader_epoch
+ }
+ write_string(&mut enc, flexible, None); // committed_metadata
+ if flexible {
+ enc.write_empty_tagged_fields(); // partition
+ enc.write_empty_tagged_fields(); // topic
+ enc.write_empty_tagged_fields(); // request
+ }
+ enc.freeze()
+}
+
+/// `OffsetFetch` request for partition 0 of `topic`. One group: top level below v8, the
+/// `groups` array from v8.
+pub fn build_offset_fetch_request(version: i16, group_id: &str, topic: &str) -> Bytes {
+ let flexible = version >= 6;
+ let mut enc = Encoder::with_capacity(64);
+
+ if version >= 8 {
+ write_array_count(&mut enc, true, 1); // groups
+ }
+ write_string(&mut enc, flexible, Some(group_id));
+ if version >= 9 {
+ write_string(&mut enc, true, None); // member_id
+ enc.write_i32(-1); // member_epoch
+ }
+ write_array_count(&mut enc, flexible, 1); // topics
+ write_string(&mut enc, flexible, Some(topic));
+ write_array_count(&mut enc, flexible, 1); // partition_indexes
+ enc.write_i32(0);
+ if flexible {
+ enc.write_empty_tagged_fields(); // topic
+ }
+ if version >= 8 {
+ enc.write_empty_tagged_fields(); // group
+ }
+ if version >= 7 {
+ enc.write_bool(false); // require_stable
+ }
+ if flexible {
+ enc.write_empty_tagged_fields(); // request
+ }
+ enc.freeze()
+}
+
/// `LeaveGroup` request. Each member is `(member_id, group_instance_id, reason)`. Below v3 the
/// body carries one member id and no identities array, so only `members[0].0` is written.
pub fn build_leave_group_request(
diff --git a/gateways/kafka/tests/golden_wire_fixtures_tests.rs b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
index e99b77584..be6b54c2e 100644
--- a/gateways/kafka/tests/golden_wire_fixtures_tests.rs
+++ b/gateways/kafka/tests/golden_wire_fixtures_tests.rs
@@ -41,15 +41,17 @@ async fn golden_apiversions_v3_flexible_response_fixture() {
.await
.expect_response("test request has acks != 0 and expects a response");
- // error_code=0, api_count=12 (compact array: N+1=13)
+ // error_code=0, api_count=14 (compact array: N+1=15)
// each entry followed by an empty tagged-fields byte; throttle_ms=0; top-level tagged fields
- let expected: [u8; 92] = [
+ let expected: [u8; 106] = [
0x00, 0x00, // error_code
- 0x0D, // compact array count (12+1)
+ 0x0F, // compact array count (14+1)
0x00, 0x00, 0x00, 0x00, 0x00, 0x09, 0x00, // key 0: Produce 0-9 (advertised)
0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, 0x00, // key 1: Fetch 4-12
0x00, 0x02, 0x00, 0x01, 0x00, 0x06, 0x00, // key 2: ListOffsets 1-6
0x00, 0x03, 0x00, 0x00, 0x00, 0x09, 0x00, // key 3: Metadata 0-9
+ 0x00, 0x08, 0x00, 0x02, 0x00, 0x09, 0x00, // key 8: OffsetCommit 2-9
+ 0x00, 0x09, 0x00, 0x01, 0x00, 0x09, 0x00, // key 9: OffsetFetch 1-9
0x00, 0x0A, 0x00, 0x00, 0x00, 0x04, 0x00, // key 10: FindCoordinator 0-4
0x00, 0x0B, 0x00, 0x00, 0x00, 0x09, 0x00, // key 11: JoinGroup 0-9
0x00, 0x0C, 0x00, 0x00, 0x00, 0x04, 0x00, // key 12: Heartbeat 0-4
@@ -71,14 +73,16 @@ async fn golden_apiversions_v1_response_fixture() {
.await
.expect_response("test request has acks != 0 and expects a response");
- // error_code=0, api_count=12; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
- let expected: [u8; 82] = [
+ // error_code=0, api_count=14; Produce advertises min=0 per KAFKA-18659; throttle_ms=0
+ let expected: [u8; 94] = [
0x00, 0x00, // error_code
- 0x00, 0x00, 0x00, 0x0C, // api count = 12
+ 0x00, 0x00, 0x00, 0x0E, // api count = 14
0x00, 0x00, 0x00, 0x00, 0x00, 0x09, // key 0: Produce 0-9 (advertised)
0x00, 0x01, 0x00, 0x04, 0x00, 0x0C, // key 1: Fetch 4-12
0x00, 0x02, 0x00, 0x01, 0x00, 0x06, // key 2: ListOffsets 1-6
0x00, 0x03, 0x00, 0x00, 0x00, 0x09, // key 3: Metadata 0-9
+ 0x00, 0x08, 0x00, 0x02, 0x00, 0x09, // key 8: OffsetCommit 2-9
+ 0x00, 0x09, 0x00, 0x01, 0x00, 0x09, // key 9: OffsetFetch 1-9
0x00, 0x0A, 0x00, 0x00, 0x00, 0x04, // key 10: FindCoordinator 0-4
0x00, 0x0B, 0x00, 0x00, 0x00, 0x09, // key 11: JoinGroup 0-9
0x00, 0x0C, 0x00, 0x00, 0x00, 0x04, // key 12: Heartbeat 0-4
diff --git a/gateways/kafka/tests/server_e2e_tests.rs b/gateways/kafka/tests/server_e2e_tests.rs
index d12764040..fdedee428 100644
--- a/gateways/kafka/tests/server_e2e_tests.rs
+++ b/gateways/kafka/tests/server_e2e_tests.rs
@@ -105,9 +105,9 @@ async fn e2e_unsupported_api_key_closes_connection() {
let (addr, _shutdown) = spawn_test_server().await;
let mut stream = TcpStream::connect(addr).await.unwrap();
- // Unknown api key (8, OffsetCommit) has no response schema this gateway can encode, so the
+ // Unknown api key (16, ListGroups) has no response schema this gateway can encode, so the
// server closes the connection without a (misparseable) response body.
- let frame1 = build_request_frame(8, 2, 99, Some("e2e-test"), &[]);
+ let frame1 = build_request_frame(16, 2, 99, Some("e2e-test"), &[]);
stream.write_all(&frame1).await.unwrap();
assert_eq!(
@@ -422,7 +422,7 @@ async fn metadata_empty_body_e2e_closes_connection() {
async fn out_of_scope_api_keys_e2e_close() {
let (addr, _shutdown) = spawn_test_server().await;
- for &(api_key, name) in &OUT_OF_SCOPE_API_KEYS[..4] {
+ for &(api_key, name) in OUT_OF_SCOPE_API_KEYS {
let mut stream = TcpStream::connect(addr).await.expect("connect");
let frame = build_request_frame(api_key, 0, i32::from(api_key), Some("scope-test"), &[]);
stream.write_all(&frame).await.expect("write oos key");
diff --git a/gateways/kafka/tests/version_firewall_tests.rs b/gateways/kafka/tests/version_firewall_tests.rs
index 63199e200..d54ccc928 100644
--- a/gateways/kafka/tests/version_firewall_tests.rs
+++ b/gateways/kafka/tests/version_firewall_tests.rs
@@ -39,9 +39,10 @@ use tokio::net::TcpStream;
use iggy_gateway_kafka::protocol::api::{
API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_FETCH, API_KEY_FIND_COORDINATOR,
API_KEY_HEARTBEAT, API_KEY_INIT_PRODUCER_ID, API_KEY_JOIN_GROUP, API_KEY_LEAVE_GROUP,
- API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_PRODUCE, API_KEY_SYNC_GROUP,
- ERROR_INVALID_REQUEST, ERROR_NONE, ERROR_UNSUPPORTED_VERSION, advertised_min_version,
- handle_request, is_supported_version, supported_api_ranges,
+ API_KEY_LIST_OFFSETS, API_KEY_METADATA, API_KEY_OFFSET_COMMIT, API_KEY_OFFSET_FETCH,
+ API_KEY_PRODUCE, API_KEY_SYNC_GROUP, ERROR_INVALID_REQUEST, ERROR_NONE,
+ ERROR_UNSUPPORTED_VERSION, advertised_min_version, handle_request, is_supported_version,
+ supported_api_ranges,
};
use codec::Decoder;
@@ -59,12 +60,13 @@ use wire::{
build_find_coordinator_request, build_heartbeat_request, build_init_producer_id_request,
build_join_group_request, build_leave_group_request, build_list_offsets_request,
build_metadata_all_topics_flexible, build_metadata_all_topics_legacy,
- build_metadata_flexible_request_v10, build_sync_group_request,
+ build_metadata_flexible_request_v10, build_offset_commit_request, build_offset_fetch_request,
+ build_sync_group_request,
};
#[test]
-fn supported_ranges_table_has_twelve_entries() {
- assert_eq!(supported_api_ranges().len(), 12);
+fn supported_ranges_table_has_fourteen_entries() {
+ assert_eq!(supported_api_ranges().len(), 14);
}
#[test]
@@ -518,6 +520,8 @@ fn request_body_for_scoped_api(api_key: i16, name: &str, version: i16) -> Bytes
},
),
API_KEY_INIT_PRODUCER_ID => build_init_producer_id_request(version, None),
+ API_KEY_OFFSET_COMMIT => build_offset_commit_request(version, "scope-group", "orders", 0),
+ API_KEY_OFFSET_FETCH => build_offset_fetch_request(version, "scope-group", "orders"),
_ => Bytes::new(),
}
}
(iggy) 02/03: feat(gateways): store Kafka group offsets in Iggy
Posted by kr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
krishvishal pushed a commit to branch kafka-offsets
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 138835366be4906d5fd94250ff4a94c3b98d70ff
Author: Krishna Vishal <kr...@gmail.com>
AuthorDate: Tue Oct 6 09:32:52 2026 +0530
feat(gateways): store Kafka group offsets in Iggy
OffsetCommit and OffsetFetch keep offsets as Iggy external group offsets, so a group resumes after a gateway restart.
---
gateways/kafka/README.md | 39 +--
gateways/kafka/docs/ACL_MAPPING.md | 5 +-
gateways/kafka/docs/CONSUMER_GROUPS.md | 20 +-
gateways/kafka/docs/MANUAL_TESTING.md | 19 +-
gateways/kafka/docs/OFFSET_STORAGE.md | 39 ++-
gateways/kafka/docs/SCOPE.md | 23 +-
gateways/kafka/docs/TEST_SUITE.md | 1 +
gateways/kafka/docs/kafka_api_keys_reference.md | 6 +-
gateways/kafka/src/bridge/error.rs | 22 +-
.../kafka/src/bridge/iggy_bridge/group_offsets.rs | 195 ++++++++++++++
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 2 +
gateways/kafka/src/bridge/mod.rs | 11 +-
.../kafka/src/protocol/handlers/offset_commit.rs | 105 +++++++-
.../kafka/src/protocol/handlers/offset_fetch.rs | 135 +++++++++-
gateways/kafka/tests/common/wire.rs | 31 +++
gateways/kafka/tests/offsets_real_bridge_tests.rs | 295 +++++++++++++++++++++
16 files changed, 871 insertions(+), 77 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index e8841b8d6..490d621be 100644
--- a/gateways/kafka/README.md
+++ b/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
> Iggy user**: with the bridge on and `IGGY_KAFKA_SASL_ENABLED` off (the default), any client that
> can reach this port can create topics (up to 1000 partitions each). Every client reads and writes
> as the bridge's Iggy user. See [docs/SCOPE.md](docs/SCOPE.md).
@@ -16,9 +18,11 @@ Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/34
> `LeaveGroup` and `SyncGroup` are real, with real membership, rebalances, graceful leave and
> session expiry ([docs/CONSUMER_GROUPS.md](docs/CONSUMER_GROUPS.md)). With the bridge off,
> Metadata reports every topic unknown, so a consumer joins a group and is assigned 0 partitions.
-> Offset commit and fetch are not implemented yet, so a consumer must use `assign()` with explicit
-> start offsets and `enable.auto.commit=false`. The Java client needs no `group.id`. librdkafka
-> refuses `assign()` without one, so set any value there.
+> With the bridge on, OffsetCommit and OffsetFetch keep each group's offsets in Iggy, so a group
+> resumes from its last commit, also after a gateway restart
+> ([docs/OFFSET_STORAGE.md](docs/OFFSET_STORAGE.md)). Every Iggy server must run a build with the
+> external group consumer kind ([#4392](https://github.com/apache/iggy/pull/4392)) before the first
+> commit. In a partly upgraded cluster, one commit can make a partition stop committing.
## Run
@@ -118,11 +122,10 @@ gateway's `IGGY_KAFKA_IGGY_PASSWORD` is known ahead of time - never reuse them o
stack. See `gateways/kafka/docker-compose.yml` and the environment variable tables below for
every other knob.
-**Limitations** (see the stub warning above for the full list): OffsetCommit and OffsetFetch are
-not implemented yet, so a consumer group can form and get assigned partitions but can't commit or
-resume from a saved offset - use `assign()` with an explicit start offset, as the Fetch example
-above does with `-o beginning`. Single gateway, single Iggy node, SASL off by default. This is a
-development quick start, not a production deployment shape.
+**Limitations** (see the stub warning above for the full list): single gateway, single Iggy node,
+SASL off by default. This is a development quick start, not a production deployment shape. If your
+local `apache/iggy:edge` predates [#4392](https://github.com/apache/iggy/pull/4392), run
+`docker compose pull`, because that server refuses every offset commit.
This quick start's topics use Iggy's default `Durability::Replicated` - a Produce ack is a
quorum commit, not an fsync. A `docker kill` or host power loss can lose acked records despite
@@ -275,10 +278,12 @@ Full reasoning, including what was rejected and why, is in
`src/bridge/` is the SDK integration layer: connects to Iggy, maps Kafka topics to Iggy
streams/topics, provisions them on demand, looks up high watermarks (one or many partitions of
-a topic per call) for `ListOffsets`, and probes and polls partitions for Fetch.
+a topic per call) for `ListOffsets`, probes and polls partitions for Fetch, and stores and reads
+group offsets for OffsetCommit and OffsetFetch.
Produce ([#3535](https://github.com/apache/iggy/issues/3535)), Fetch ([#3536](https://github.com/apache/iggy/issues/3536)), ListOffsets
-([#3537](https://github.com/apache/iggy/issues/3537)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)) and CreateTopics
-([#3538](https://github.com/apache/iggy/issues/3538)) call it.
+([#3537](https://github.com/apache/iggy/issues/3537)), Metadata ([#3534](https://github.com/apache/iggy/issues/3534)), CreateTopics
+([#3538](https://github.com/apache/iggy/issues/3538)), OffsetCommit and OffsetFetch
+([#3542](https://github.com/apache/iggy/issues/3542)) call it.
Tested by `bridge`'s own unit tests, `tests/bridge_iggy_integration_tests.rs` and the
`tests/*_real_bridge_tests.rs` suites. All but the unit tests start a real `iggy-server`.
diff --git a/gateways/kafka/docs/ACL_MAPPING.md b/gateways/kafka/docs/ACL_MAPPING.md
index c552306f7..2fe7c455c 100644
--- a/gateways/kafka/docs/ACL_MAPPING.md
+++ b/gateways/kafka/docs/ACL_MAPPING.md
@@ -120,9 +120,8 @@ An empty result is `error_code` 0 with no resources, not an error. Kafka draws a
rendering a Kafka topic name means resolving every id through a lookup, one call per resource, on
what should be a cheap read. Worse, the resolution can be ambiguous in the direction we need it:
the topic mapping is one-way, from Kafka name to Iggy stream and topic, and an override means an
-Iggy topic name does not identify the Kafka name it came from. The same limitation is already
-recorded against `OffsetFetch` with a null topic list. Closing it needs a reverse index, which
-belongs with whichever change first needs one.
+Iggy topic name does not identify the Kafka name it came from. Closing it needs a reverse index,
+which belongs with whichever change first needs one.
The consequence is honest but worth stating: a principal whose access is granted per topic rather
than globally is described as having no topic bindings. That under-reports rather than
diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md
index f649147e9..efbc09be6 100644
--- a/gateways/kafka/docs/CONSUMER_GROUPS.md
+++ b/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
-A consumer completes JoinGroup and SyncGroup and then holds no partitions. Metadata is a stub that
-answers `UNKNOWN_TOPIC_OR_PARTITION` (3) for every topic, so the leader's assignor sees no
-partitions and hands every member an empty assignment. The consumer stays a member and keeps
-heartbeating, but has nothing to fetch.
-
-Once Metadata reports partitions, the next wall is `OffsetFetch` (9), which a consumer sends after
-`SyncGroup` for a non-empty assignment and which is not in scope
-([#3542](https://github.com/apache/iggy/issues/3542)). An unlisted key closes the connection, so
-that consumer would loop: coordinator connection closes, client marks the coordinator unknown,
-re-runs FindCoordinator, retries OffsetFetch, closes again. Fetch reads records with a bridge, so a
-consumer that uses `assign()` outside a group can consume. A group member cannot until #3542 lands.
+Without a bridge, a consumer completes JoinGroup and SyncGroup and then holds no partitions.
+Metadata is a stub that answers `UNKNOWN_TOPIC_OR_PARTITION` (3) for every topic, so the leader's
+assignor sees no partitions and hands every member an empty assignment. The consumer stays a
+member and keeps heartbeating, but has nothing to fetch.
+
+With a bridge, Metadata reports the partitions and a group member consumes. After `SyncGroup`, it
+reads its start position with `OffsetFetch` (9) and later commits with `OffsetCommit` (8). Both
+keep the offsets in Iggy ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). Without a bridge, both answer
+`COORDINATOR_LOAD_IN_PROGRESS` (14), and the client retries them.
A dynamic consumer releases its partitions on close through `LeaveGroup`. A static one does not
send it (see [Static membership](#static-membership-is-accepted-not-honoured)), so a static
diff --git a/gateways/kafka/docs/MANUAL_TESTING.md b/gateways/kafka/docs/MANUAL_TESTING.md
index 3df52b832..4569f50fa 100644
--- a/gateways/kafka/docs/MANUAL_TESTING.md
+++ b/gateways/kafka/docs/MANUAL_TESTING.md
@@ -216,6 +216,8 @@ For each API key, test **min−1**, **min**, **max**, **max+1** using `kafka-mes
| 1 | Fetch | 4 | 12 | 3, 4, 12, 13 |
| 2 | ListOffsets | 1 | 6 | 0, 1, 6, 7 |
| 19 | CreateTopics | 2 | 5 | 1, 2, 5, 6 |
+| 8 | OffsetCommit | 2 | 9 | 1, 2, 9, 10 |
+| 9 | OffsetFetch | 1 | 9 | 0, 1, 9, 10 |
| 10 | FindCoordinator | 0 | 4 | −1, 0, 4, 5 |
| 11 | JoinGroup | 0 | 9 | −1, 0, 9, 10 |
| 12 | Heartbeat | 0 | 4 | −1, 0, 4, 5 |
@@ -225,10 +227,10 @@ For each API key, test **min−1**, **min**, **max**, **max+1** using `kafka-mes
| ID | Test | Expected for in-range | Expected for out-of-range |
| ---- | ------ | ---------------------- | --------------------------- |
-| B1 | ApiVersions negotiation | `error_code=0`; body lists 12 API keys with correct min/max | KIP-511 exception: still answers, `error_code=35` (UNSUPPORTED_VERSION), v0 response header regardless of the request's own encoding |
+| B1 | ApiVersions negotiation | `error_code=0`; body lists 14 API keys with correct min/max | KIP-511 exception: still answers, `error_code=35` (UNSUPPORTED_VERSION), v0 response header regardless of the request's own encoding |
| B2 | Metadata out-of-range | N/A | **Connection closes**, no response sent - Metadata has no top-level error field to carry a version-correct error in |
-| B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these five messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
-| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,10,11,12,13,14,18,19,22 only — no OffsetCommit/OffsetFetch, and no transaction keys (24, 25, 26, 28) |
+| B3 | Produce/Fetch/ListOffsets/CreateTopics/InitProducerId/OffsetCommit/OffsetFetch out-of-range | N/A | **Connection closes** for both above-max and below-min - `kafka_protocol`'s schema floor for each of these seven messages equals `SUPPORTED_RANGES`' own min, so there is no encodable error response below min either (see `SCOPE.md`'s Governance model) |
+| B4 | ApiVersions lists only scoped keys | Decode response | Contains keys 0,1,2,3,8,9,10,11,12,13,14,18,19,22 only, and no transaction keys (24, 25, 26, 28) |
An out-of-range version only ever produces `error_code=35` on ApiVersions (B1); every other API
key's out-of-range case closes the connection - see B2/B3. InitProducerId and Produce also send
@@ -249,7 +251,7 @@ decoded `ec`.
| ID | API key | Name | Steps | Expected |
| ---- | --------- | ------ | ------- | ---------- |
-| C1 | 8 | OffsetCommit | `send --host 127.0.0.1:9093 --api-key 8 --version 2` | Connection closes, no response bytes |
+| C1 | 15 | DescribeGroups | `send --host 127.0.0.1:9093 --api-key 15` | Connection closes, no response bytes |
| C2 | 10 | FindCoordinator | `send --host 127.0.0.1:9093 --api-key 10` | Connection closes |
| C3 | 17 | SaslHandshake | `send --host 127.0.0.1:9093 --api-key 17` | Connection closes |
| C4 | 20 | DeleteTopics | `send --host 127.0.0.1:9093 --api-key 20` | Connection closes |
@@ -313,6 +315,8 @@ success. This category predates the bridge landing: G2/G3's "fails at metadata"
| G6 | Graceful kcat exit | G3's two kcats, then Ctrl-C one (kcat closes its consumer, which sends LeaveGroup v0/v1) | The survivor rebalances within one heartbeat interval, not `session.timeout.ms`. Join, leave and sync are real, but the assignment is empty: the Metadata stub reports the topic unknown, so the assignor has no partitions to hand out |
| G7 | Graceful Java exit | G5 in two terminals, then Ctrl-C one | Same as G6 |
| G8 | Static member exit | G7 with `--consumer-property group.instance.id=x` on the one stopped | No LeaveGroup is sent; the survivor waits out the session timeout before rebalancing |
+| G9 | Group resumes from its commit | Bridge on and topic `orders` made as in Category I. Run `kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9093 --group g9 --topic orders --from-beginning`, stop it, produce one record, run it again | The second run prints only the new record. `--from-beginning` applies only to a group with no committed offset |
+| G10 | Resume after a gateway restart | G9 with a gateway restart before the second run | Same as G9, because the offsets live in Iggy ([#3542](https://github.com/apache/iggy/issues/3542)) |
Record kcat version and exact error strings in your test log. G1 passing is the minimum bar for client compatibility smoke.
@@ -437,12 +441,12 @@ Gateway commit: ___________
kcat version (if used): ___________
[ ] A1–A10 Smoke tests
-[ ] B1–B4 Version firewall (all 7 keys × 4 boundary versions)
+[ ] B1–B4 Version firewall (every listed key × 4 boundary versions)
[ ] C1–C4 Unsupported API keys
[ ] D1–D10 Flexible vs legacy encoding
[ ] E1–E4 Metadata stub semantics
[ ] F1–F6 TCP / connection behavior
-[ ] G1–G8 kcat / Java client (record errors for G2/G3)
+[ ] G1–G10 kcat / Java client (record errors for G2/G3)
[ ] H1–H6 Adversarial input
[ ] I1–I6 Docker Compose quick start (record wall-clock time to I3; I5 is restart durability)
@@ -478,5 +482,4 @@ These are documented as TODO in [SCOPE.md](SCOPE.md) — do not fail #3421 valid
- SASL authentication
- Accurate partition leadership / ISR
- Transactional produce
-- Real offset commit semantics
-- Consumer group offset commit/fetch and admin group views (join/sync/heartbeat/leave themselves are covered by G3-G8)
+- Admin group views (join/sync/heartbeat/leave are covered by G3-G8, offset commit and fetch by G9-G10)
diff --git a/gateways/kafka/docs/OFFSET_STORAGE.md b/gateways/kafka/docs/OFFSET_STORAGE.md
index fdc9e5fbb..be630935d 100644
--- a/gateways/kafka/docs/OFFSET_STORAGE.md
+++ b/gateways/kafka/docs/OFFSET_STORAGE.md
@@ -36,14 +36,21 @@ leaves it behind.
## Gateway rules
-- Create group `kafka.cg.<group>` on the topic before the first commit there. Already exists is fine.
+- A commit stores the offset. If group `kafka.cg.<group>` is not on the topic, the gateway creates it
+ and stores again. A group that already exists counts as created. An operator can delete the group,
+ and the next commit creates it again.
- The prefix keeps a Kafka group apart from a native Iggy group with the same name.
- Group id: 1 to 246 bytes, else `INVALID_GROUP_ID` (24). Iggy names stop at 255.
-- Negative offset: delete the key. Not found counts as success. Kafka consumers read any negative
- offset as none.
+- Negative offset: delete the key. A missing key or group counts as success. Kafka consumers read any
+ negative offset as none.
- Metadata string and leader epoch: dropped. OffsetFetch returns `""` and -1.
- No key, unknown group or unknown topic: OffsetFetch returns -1 and no error.
-- Null topic list (OffsetFetch v2+): read every Kafka topic that has group `kafka.cg.<group>`.
+- Any other failed read fails the whole group in OffsetFetch. A lost read never goes back as -1,
+ because the consumer then resets its position.
+- Null topic list (OffsetFetch v2+): read every Kafka topic that has group `kafka.cg.<group>`. Only
+ the partitions with an offset come back.
+- A failure that a retry can fix goes back as `COORDINATOR_LOAD_IN_PROGRESS` (14). The Java client
+ fails an offset call on 6, and fails OffsetFetch on 7.
- Committed offset below the oldest kept message: Fetch reads from the first kept message.
## Limits
@@ -60,7 +67,9 @@ A group nobody uses keeps its keys until its Iggy group is deleted on each topic
Offsets are shared, since the key depends only on the Kafka group id. Membership is not
([`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)).
-## Server work
+## Server side
+
+[#4392](https://github.com/apache/iggy/pull/4392) adds the kind:
- Wire: `KIND_EXTERNAL_GROUP = 3`, next to kinds 1 and 2 in `core/binary_protocol/src/primitives/consumer.rs`.
- SDK: `ConsumerKind::ExternalGroup`. The enum is not `#[non_exhaustive]`, so this breaks the API.
@@ -68,6 +77,26 @@ Offsets are shared, since the key depends only on the Kafka group id. Membership
- Server: no fence and no range check for the kind, refuse polls, leave it out of
`min_committed_offset`, delete its keys with the group (`partition_reconciler.rs`).
+## Upgrade
+
+Every Iggy server must run a build with kind 3 before the first external group store. Nothing
+enforces this yet.
+
+- A server with an older build refuses kind 3 with `InvalidCommand`. The gateway sends that as
+ `UNKNOWN_SERVER_ERROR` and logs it at `error!`.
+- A primary with the new build accepts the store. If most replicas of that partition run an older
+ build, they cannot decode it, and the partition stops committing (`core/server/README.md`).
+
+The Iggy HTTP API cannot carry kind 3. The gateway uses TCP, so this does not limit it.
+
+## Reads in a cluster
+
+An offset read goes to the Iggy node that the gateway is connected to. That is not always the
+partition primary (the node that orders the partition writes). So an OffsetFetch right after an
+OffsetCommit can return the older value until that node applies the commit. A consumer that starts
+from the older value reads some records again. Reads from the primary are
+[#4409](https://github.com/apache/iggy/issues/4409).
+
## References
- Record mapping: [`BRIDGE_MAPPING.md`](BRIDGE_MAPPING.md)
diff --git a/gateways/kafka/docs/SCOPE.md b/gateways/kafka/docs/SCOPE.md
index b2c736040..60b3a6b67 100644
--- a/gateways/kafka/docs/SCOPE.md
+++ b/gateways/kafka/docs/SCOPE.md
@@ -2,9 +2,14 @@
## Issue #3421 — in scope (this iteration)
-A TCP listener on the Kafka wire port. It decodes requests, validates scoped API keys, versions and wire formats, and answers them. With a bridge, Produce, Fetch, ListOffsets, Metadata and CreateTopics use Iggy.
+A TCP listener on the Kafka wire port. It decodes requests, validates scoped API keys, versions and wire formats, and answers them. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, OffsetCommit and OffsetFetch use Iggy.
-**Stub semantics (important):** without a bridge, every API answers with a stub. Produce discards the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). Fetch and ListOffsets answer 6 too. CreateTopics validates the request but answers `NOT_CONTROLLER` (41), so clients do not believe topics were created. Do not read `ec=0` from a stub as durable storage. Produce is the one API that stores records once you configure a bridge ([#3535](https://github.com/apache/iggy/issues/3535)).
+**Stub semantics (important):** without a bridge, every API answers with a stub. Produce discards
+the payload and answers with retriable `NOT_LEADER_OR_FOLLOWER` (6). Fetch and ListOffsets answer 6
+too. CreateTopics validates the request but answers `NOT_CONTROLLER` (41), so clients do not believe
+topics were created. OffsetCommit and OffsetFetch answer retriable `COORDINATOR_LOAD_IN_PROGRESS`
+(14). Do not read `ec=0` from a stub as durable storage. Produce is the one API that stores records
+once you configure a bridge ([#3535](https://github.com/apache/iggy/issues/3535)).
| Deliverable | Status | Location |
| ------------- | -------- | ---------- |
@@ -29,7 +34,7 @@ Expand `SUPPORTED_RANGES` only after a key/version pair is manually tested. ApiV
**Every unsupported-version case closes the connection, for every listed key** - not just above
the encoder max. `kafka_protocol`'s schema floor for each supported message happens to equal
`SUPPORTED_RANGES`' own min today (Produce 3, Fetch 4, ListOffsets 1, Metadata 0, ApiVersions 0,
-CreateTopics 2, InitProducerId 0, and 0 for the four consumer-group keys), so there is no version below an API's
+CreateTopics 2, InitProducerId 0, OffsetCommit 2, OffsetFetch 1, and 0 for the five group coordination keys), so there is no version below an API's
min that the crate can actually encode a response for either - `unsupported_version_response`
still tries, but the encode attempt fails and the connection closes rather than sending a
malformed body.
@@ -49,6 +54,8 @@ it knows the server supports flexible encoding.
| 1 | Fetch | 4 | 12 | 4, 5, 6, 7, 8, 9, 10, 11, 12 | With a bridge: shared topic probes, paged `poll_messages` per partition with records, wait up to `max_wait_ms` (max 18 s). Without one: stub returns `NOT_LEADER_OR_FOLLOWER` (6) |
| 2 | ListOffsets | 1 | 6 | 1, 2, 3, 4, 5, 6 | Decode request; stub response |
| 19 | CreateTopics | 2 | 5 | 2, 3, 4, 5 | Decode request; stub returns `NOT_CONTROLLER` (41); `-1` partitions/RF = broker default on v4+ |
+| 8 | OffsetCommit | 2 | 9 | 2 … 9 | With a bridge: stores each offset as an Iggy external group offset, and a negative offset deletes it ([`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)). Without one: `COORDINATOR_LOAD_IN_PROGRESS` (14). Flexible encoding at v8+ |
+| 9 | OffsetFetch | 1 | 9 | 1 … 9 | With a bridge: reads each offset, -1 when none is stored, and a null topic list reads every topic that holds the group. Without one: 14. Flexible encoding at v6+ |
| 10 | FindCoordinator | 0 | 4 | 0, 1, 2, 3, 4 | Answers "this gateway" for group keys; `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` (53) for the transaction key type, `INVALID_REQUEST` (42) for share; flexible encoding at v3+ |
| 11 | JoinGroup | 0 | 9 | 0 … 9 | Real membership; parks on the group's join barrier; flexible encoding at v6+ |
| 12 | Heartbeat | 0 | 4 | 0, 1, 2, 3, 4 | Refreshes a session; `REBALANCE_IN_PROGRESS` (27) drives a rejoin; flexible encoding at v4+ |
@@ -70,6 +77,8 @@ Use this table when configuring clients or generating wire fixtures with `kafka-
| 1 | Fetch | 4–12 | v12 |
| 2 | ListOffsets | 1–6 | v6 |
| 3 | Metadata | 0–9 | v9 |
+| 8 | OffsetCommit | 2–9 | v8 |
+| 9 | OffsetFetch | 1–9 | v6 |
| 10 | FindCoordinator | 0–4 | v3 |
| 11 | JoinGroup | 0–9 | v6 |
| 12 | Heartbeat | 0–4 | v4 |
@@ -87,8 +96,6 @@ All API keys not listed above close the connection (see Governance model above)
| API key | Name | Notes |
| --------- | ------ | ------- |
-| 8 | OffsetCommit | Consumer group offsets — [#3542](https://github.com/apache/iggy/issues/3542) |
-| 9 | OffsetFetch | Consumer group offsets — [#3542](https://github.com/apache/iggy/issues/3542); sent right after SyncGroup, so a joined consumer loops on it today ([`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)) |
| 15, 16 | DescribeGroups, ListGroups | Admin views — [#3548](https://github.com/apache/iggy/issues/3548) |
| 17 | SaslHandshake | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`AUTHENTICATION.md`](AUTHENTICATION.md)) |
| 29 | DescribeAcls | Implemented behind `IGGY_KAFKA_SASL_ENABLED`, advertised only while it is on ([`ACL_MAPPING.md`](ACL_MAPPING.md)) |
@@ -334,7 +341,8 @@ decoded) on the Produce/Fetch hot paths, preserving the one property this TODO w
`bounds_guard.rs` covers the DoS-bound gap the crate itself leaves open (see Governance model
above).
-- [ ] Offset-related consumer-group API keys (8, 9) and real Metadata topology remain unimplemented (see Phase 3 below) - the crate can decode them when that phase starts
+- [x] Offset-related consumer-group API keys (8, 9) - see Phase 3 below
+- [ ] Real Metadata topology
### Phase 3 — Consumer groups (~7 API keys)
@@ -347,7 +355,8 @@ Offset persistence design ([#3540](https://github.com/apache/iggy/issues/3540)):
- [x] FindCoordinator (10), JoinGroup (11), Heartbeat (12), SyncGroup (14) -
[#3541](https://github.com/apache/iggy/issues/3541); LeaveGroup (13) -
[#3543](https://github.com/apache/iggy/issues/3543); see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md)
-- [ ] OffsetCommit (8), OffsetFetch (9)
+- [x] OffsetCommit (8), OffsetFetch (9) - [#3542](https://github.com/apache/iggy/issues/3542);
+ offsets live in Iggy, see [`OFFSET_STORAGE.md`](OFFSET_STORAGE.md)
- [ ] DescribeGroups (15), ListGroups (16) as needed by target clients
### Phase 3+ — Auth, admin, tuning
diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md
index e2a006958..bfb73c95e 100644
--- a/gateways/kafka/docs/TEST_SUITE.md
+++ b/gateways/kafka/docs/TEST_SUITE.md
@@ -83,6 +83,7 @@ file under `tests/` anymore.
| [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, codes 3 and 43 | No (needs the `iggy-server` binary - see Prerequisites) |
| [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) |
| [`produce_real_bridge_tests.rs`](../tests/produce_real_bridge_tests.rs) | Produce (key 0) through the whole handler against a real, spawned `iggy-server` — records go in as Kafka wire bytes and come back through the Iggy SDK, plus one error code per partition | No (needs the `iggy-server` binary - see Prerequisites) |
+| [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) |
| [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) | [#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics → Metadata → Produce → ListOffsets over real TCP against the real compiled `iggy-gateway-kafka` binary bridged to a real spawned `iggy-server` - the only suite that spawns the gateway as a process rather than calling `KafkaGateway::run` in-process | No (needs both binaries - see Prerequisites) |
`tests/common/` holds shared helpers (`codec.rs`, `fixtures.rs`, `gateway_process.rs`, `scope.rs`,
diff --git a/gateways/kafka/docs/kafka_api_keys_reference.md b/gateways/kafka/docs/kafka_api_keys_reference.md
index 8c5582847..d2db0ee05 100644
--- a/gateways/kafka/docs/kafka_api_keys_reference.md
+++ b/gateways/kafka/docs/kafka_api_keys_reference.md
@@ -293,15 +293,16 @@ Key new minimums:
| Heartbeat | v0-v4 | v4 | current |
| LeaveGroup | v0-v5 | v5 | current |
| SyncGroup | v0-v5 | v5 | current |
+| OffsetCommit | v2-v9 | v9 | current |
+| OffsetFetch | v1-v9 | v9 | current |
-### Missing from `SUPPORTED_RANGES` (76 of the 88 API keys in this document)
+### Missing from `SUPPORTED_RANGES` (74 of the 88 API keys in this document)
Every key not in `SUPPORTED_RANGES` closes the connection - the same policy applied to every
other unlisted key, not a special case for these. The gateway declines to define a response for a
key it does not advertise, and a conforming client never sends one, so no response shape has to
be agreed. This includes:
-- **Client bootstrap blockers**: OffsetCommit (8), OffsetFetch (9)
- **New consumer group protocol**: ConsumerGroupHeartbeat (68), opt-in via `group.protocol=consumer` (the 4.0 default is still `classic`)
- **Share groups (KIP-932)**: ShareFetch, ShareGroupHeartbeat, ShareAcknowledge
- **Auth flow**: SaslHandshake (17), SaslAuthenticate (36)
@@ -309,6 +310,7 @@ be agreed. This includes:
The classic consumer group keys FindCoordinator (10), JoinGroup (11), Heartbeat (12),
LeaveGroup (13) and SyncGroup (14) are supported - see [`CONSUMER_GROUPS.md`](CONSUMER_GROUPS.md).
+OffsetCommit (8) and OffsetFetch (9) are supported too - see [`OFFSET_STORAGE.md`](OFFSET_STORAGE.md).
Remaining scope (consumer groups, auth, admin/tuning) is tracked in `SCOPE.md`'s TODO section,
not duplicated here.
diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs
index a5352adf6..da2a0dc77 100644
--- a/gateways/kafka/src/bridge/error.rs
+++ b/gateways/kafka/src/bridge/error.rs
@@ -19,10 +19,10 @@ use iggy::prelude::IggyError;
use thiserror::Error;
use crate::protocol::api::{
- ERROR_INVALID_PARTITIONS, ERROR_INVALID_REQUEST, ERROR_INVALID_TOPIC_EXCEPTION,
- ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_POLICY_VIOLATION, ERROR_REQUEST_TIMED_OUT,
- ERROR_TOPIC_ALREADY_EXISTS, ERROR_TOPIC_AUTHORIZATION_FAILED, ERROR_UNKNOWN_SERVER_ERROR,
- ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_INVALID_PARTITIONS, ERROR_INVALID_REQUEST,
+ ERROR_INVALID_TOPIC_EXCEPTION, ERROR_NOT_LEADER_OR_FOLLOWER, ERROR_POLICY_VIOLATION,
+ ERROR_REQUEST_TIMED_OUT, ERROR_TOPIC_ALREADY_EXISTS, ERROR_TOPIC_AUTHORIZATION_FAILED,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION,
};
/// Errors from the `IggyBridge`: connection lifecycle, config, and Iggy SDK calls.
@@ -124,6 +124,20 @@ impl BridgeError {
}
}
+ /// [`Self::to_kafka_error_code`] for `OffsetCommit` and `OffsetFetch`. A failure that a retry
+ /// can fix becomes `COORDINATOR_LOAD_IN_PROGRESS` (14): the Java client retries 14 in both
+ /// APIs, but fails the call on 6, and on 7 in `OffsetFetch`. Offset calls are idempotent, so
+ /// the unknown outcome behind 7 does no harm on a retry.
+ #[must_use]
+ pub const fn to_offset_error_code(&self) -> i16 {
+ match self.to_kafka_error_code() {
+ ERROR_NOT_LEADER_OR_FOLLOWER | ERROR_REQUEST_TIMED_OUT => {
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS
+ }
+ code => code,
+ }
+ }
+
/// True when Iggy rejected the bridge's own login. The Kafka client only sees -1.
#[must_use]
pub const fn is_bridge_login_rejected(&self) -> bool {
diff --git a/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
new file mode 100644
index 000000000..9b9a55fb3
--- /dev/null
+++ b/gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs
@@ -0,0 +1,195 @@
+// 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 iggy::prelude::{Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyError};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, TopicTarget, with_request_timeout};
+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.";
+
+impl IggyBridge {
+ /// Stores `offset` as the committed offset of Kafka group `group` on `partition` of `topic`.
+ ///
+ /// 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.
+ ///
+ /// The Iggy group that holds the offsets is created when a store finds it missing, so the
+ /// common commit costs one call.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses a call. [`BridgeError::Timeout`] past
+ /// `REQUEST_TIMEOUT`.
+ pub async fn commit_group_offset(
+ &self,
+ topic: &TopicTarget,
+ group: &str,
+ partition: u32,
+ offset: i64,
+ ) -> Result<(), BridgeError> {
+ let consumer = offset_consumer(group)?;
+ let Ok(offset) = u64::try_from(offset) else {
+ return match with_request_timeout(self.client.delete_consumer_offset(
+ &consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ ))
+ .await
+ {
+ Err(BridgeError::Iggy(
+ IggyError::ConsumerOffsetNotFound(_) | IggyError::ConsumerGroupNameNotFound(..),
+ )) => Ok(()),
+ result => result,
+ };
+ };
+ let store = || {
+ with_request_timeout(self.client.store_consumer_offset(
+ &consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ offset,
+ ))
+ };
+ match store().await {
+ Err(BridgeError::Iggy(IggyError::ConsumerGroupNameNotFound(..))) => {
+ self.create_offset_group(topic, group).await?;
+ store().await
+ }
+ result => result,
+ }
+ }
+
+ /// The committed offset of Kafka group `group` on `partition` of `topic`. `None` when there
+ /// is no key or no group.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic.
+ /// [`BridgeError::Timeout`] past `REQUEST_TIMEOUT`.
+ pub async fn fetch_group_offset(
+ &self,
+ topic: &TopicTarget,
+ group: &str,
+ partition: u32,
+ ) -> Result<Option<u64>, BridgeError> {
+ let consumer = offset_consumer(group)?;
+ let info = with_request_timeout(self.client.get_consumer_offset(
+ &consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ ))
+ .await?;
+ Ok(info.map(|info| info.stored_offset))
+ }
+
+ /// Every Kafka topic that holds offsets of Kafka group `group`, for an `OffsetFetch` with a
+ /// null topic list. One call per Kafka topic.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses a call. [`BridgeError::Timeout`] past
+ /// `REQUEST_TIMEOUT`.
+ pub async fn offset_group_topics(
+ &self,
+ group: &str,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let group_id = offset_group_id(group)?;
+ let mut held = Vec::new();
+ for topic in self.list_kafka_topics().await? {
+ let target = self.topic_target(&topic.kafka_topic)?;
+ let found = with_request_timeout(self.client.get_consumer_group(
+ &target.stream_id,
+ &target.topic_id,
+ &group_id,
+ ))
+ .await?;
+ if found.is_some() {
+ held.push(topic);
+ }
+ }
+ Ok(held)
+ }
+
+ /// Creates the Iggy group of Kafka group `group` on `topic`. A group that exists counts as
+ /// created, since another commit or another gateway can create it first.
+ async fn create_offset_group(
+ &self,
+ topic: &TopicTarget,
+ group: &str,
+ ) -> Result<(), BridgeError> {
+ let name = offset_group_name(group);
+ match with_request_timeout(self.client.create_consumer_group(
+ &topic.stream_id,
+ &topic.topic_id,
+ &name,
+ ))
+ .await
+ {
+ Ok(_) => {
+ info!("created Iggy consumer group '{name}' for Kafka group offsets");
+ Ok(())
+ }
+ Err(BridgeError::Iggy(IggyError::ConsumerGroupNameAlreadyExists(..))) => Ok(()),
+ Err(error) => Err(error),
+ }
+ }
+}
+
+fn offset_group_name(group: &str) -> String {
+ format!("{OFFSET_GROUP_PREFIX}{group}")
+}
+
+fn offset_group_id(group: &str) -> Result<Identifier, BridgeError> {
+ Identifier::named(&offset_group_name(group)).map_err(BridgeError::Iggy)
+}
+
+/// The Iggy key of Kafka group `group`'s offsets.
+fn offset_consumer(group: &str) -> Result<Consumer, BridgeError> {
+ offset_group_id(group).map(Consumer::external_group)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy::prelude::ConsumerKind;
+
+ #[test]
+ fn given_a_kafka_group_the_key_should_be_its_prefixed_external_group() {
+ let consumer = offset_consumer("orders").unwrap();
+
+ assert_eq!(consumer.kind, ConsumerKind::ExternalGroup);
+ assert_eq!(consumer.id, Identifier::named("kafka.cg.orders").unwrap());
+ }
+
+ /// 246 bytes of group id plus the prefix is the 255 byte cap of an Iggy name.
+ #[test]
+ fn given_the_longest_kafka_group_id_the_key_should_still_be_a_valid_name() {
+ let longest = "g".repeat(crate::group::MAX_GROUP_ID_BYTES);
+
+ assert!(offset_consumer(&longest).is_ok());
+ assert!(offset_consumer(&format!("{longest}g")).is_err());
+ }
+}
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index 0d95e6002..82f79d553 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -31,12 +31,14 @@ use crate::bridge::error::BridgeError;
use crate::bridge::topic_map::validate_kafka_topic_name;
mod fetch;
+mod group_offsets;
mod offsets;
mod produce;
mod topics;
use fetch::{FetchPool, LazyClient};
pub(crate) use fetch::{FetchSlot, PartitionProbe, TopicProbe};
+pub use group_offsets::OFFSET_GROUP_PREFIX;
pub use topics::{KafkaTopicMetadata, TopicCreationOutcome};
/// Passes attempted, after the first, before [`IggyBridge::connect`] gives up and returns `Err`.
diff --git a/gateways/kafka/src/bridge/mod.rs b/gateways/kafka/src/bridge/mod.rs
index 1c8836ac1..deab83db9 100644
--- a/gateways/kafka/src/bridge/mod.rs
+++ b/gateways/kafka/src/bridge/mod.rs
@@ -17,9 +17,10 @@
//! Iggy SDK integration layer.
//!
-//! Maps Kafka topics to Iggy streams/topics, exposes create-if-missing provisioning and
-//! high-watermark lookups, and translates Iggy errors to Kafka wire error codes. `ListOffsets`,
-//! Produce, Fetch, Metadata and `CreateTopics` call it.
+//! Maps Kafka topics to Iggy streams/topics, exposes create-if-missing provisioning,
+//! high-watermark lookups and group offsets, and translates Iggy errors to Kafka wire error codes.
+//! `ListOffsets`, Produce, Fetch, Metadata, `CreateTopics`, `OffsetCommit` and `OffsetFetch` call
+//! it.
pub mod config;
pub mod error;
@@ -28,5 +29,7 @@ pub mod topic_map;
pub use config::{DEFAULT_MAX_MESSAGE_SIZE, IggyBridgeConfig};
pub use error::BridgeError;
-pub use iggy_bridge::{IggyBridge, KafkaTopicMetadata, TopicCreationOutcome, TopicTarget};
+pub use iggy_bridge::{
+ IggyBridge, KafkaTopicMetadata, OFFSET_GROUP_PREFIX, TopicCreationOutcome, TopicTarget,
+};
pub use topic_map::{TopicMapping, TopicOverride};
diff --git a/gateways/kafka/src/protocol/handlers/offset_commit.rs b/gateways/kafka/src/protocol/handlers/offset_commit.rs
index 454eda271..f65515c68 100644
--- a/gateways/kafka/src/protocol/handlers/offset_commit.rs
+++ b/gateways/kafka/src/protocol/handlers/offset_commit.rs
@@ -22,16 +22,20 @@
//! every partition.
use bytes::Bytes;
+use iggy::prelude::IggyError;
use kafka_protocol::messages::offset_commit_response::{
OffsetCommitResponsePartition, OffsetCommitResponseTopic,
};
use kafka_protocol::messages::{OffsetCommitRequest, OffsetCommitResponse};
+use tokio::time::{Instant, timeout_at};
+use crate::bridge::{BridgeError, 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,
- GatewayState, HandleOutcome, is_supported_version,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, REQUEST_DEADLINE,
+ is_supported_version,
};
use crate::protocol::bounds_guard::validate_offset_commit_shape;
use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close};
@@ -69,17 +73,76 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
.groups
.validate_commit(&CommitRequest::from(&request))
.await;
- let code = if error == ERROR_NONE {
- ERROR_COORDINATOR_LOAD_IN_PROGRESS
- } else {
- error
+ let codes = match (&state.bridge, error) {
+ (Some(bridge), ERROR_NONE) => {
+ commit_all(bridge, &request, Instant::now() + 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, &request, &same_code(&request, code)),
+ encode_response(api_version, &request, &codes),
"OffsetCommit",
)
}
+/// Stores every requested offset, in request order. Each partition costs one call on the lockstep
+/// client, so a partition that the deadline leaves out answers 14 and makes no call.
+async fn commit_all(
+ bridge: &IggyBridge,
+ request: &OffsetCommitRequest,
+ deadline: Instant,
+) -> PartitionCodes {
+ let group = request.group_id.0.as_str();
+ let mut codes = Vec::with_capacity(request.topics.len());
+ for topic in &request.topics {
+ let target = bridge.topic_target(topic.name.0.as_str());
+ let mut topic_codes = Vec::with_capacity(topic.partitions.len());
+ for partition in &topic.partitions {
+ let code = match (&target, u32::try_from(partition.partition_index)) {
+ (Err(error), _) => error.to_offset_error_code(),
+ (Ok(_), Err(_)) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ // A call started now still goes out once, and the client retries the 14 anyway.
+ (Ok(_), Ok(_)) if Instant::now() >= deadline => ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ (Ok(target), Ok(index)) => match timeout_at(
+ deadline,
+ bridge.commit_group_offset(target, group, index, partition.committed_offset),
+ )
+ .await
+ {
+ Ok(result) => commit_code(result),
+ Err(_elapsed) => ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ },
+ };
+ topic_codes.push(code);
+ }
+ codes.push(topic_codes);
+ }
+ codes
+}
+
+/// The Kafka code for one partition's store or delete. A refusal that only the operator can fix
+/// is logged, since the client sees no more than -1.
+fn commit_code(result: core::result::Result<(), BridgeError>) -> i16 {
+ let Err(error) = result else {
+ return ERROR_NONE;
+ };
+ match &error {
+ BridgeError::Iggy(IggyError::TooManyConsumerOffsets) => tracing::error!(
+ %error,
+ "Iggy refused a Kafka offset commit: the partition holds \
+ partition.consumer_offsets_max keys"
+ ),
+ BridgeError::Iggy(IggyError::InvalidCommand) => tracing::error!(
+ %error,
+ "Iggy refused a Kafka offset commit as an invalid command, which is what a server \
+ without the external group kind answers"
+ ),
+ _ => {}
+ }
+ error.to_offset_error_code()
+}
+
/// `code` for every requested partition.
#[must_use]
pub fn same_code(request: &OffsetCommitRequest, code: i16) -> PartitionCodes {
@@ -143,7 +206,9 @@ mod tests {
use kafka_protocol::protocol::{Decodable, Encodable, StrBytes};
use super::*;
- use crate::protocol::api::{BrokerAdvertise, ERROR_UNKNOWN_MEMBER_ID};
+ use crate::protocol::api::{
+ BrokerAdvertise, ERROR_UNKNOWN_MEMBER_ID, ERROR_UNKNOWN_SERVER_ERROR,
+ };
fn request(generation_id: i32, member_id: &'static str) -> OffsetCommitRequest {
OffsetCommitRequest::default()
@@ -217,6 +282,32 @@ mod tests {
);
}
+ #[test]
+ fn given_bridge_results_the_commit_codes_should_keep_retriable_failures_retriable() {
+ assert_eq!(commit_code(Ok(())), ERROR_NONE);
+ for retriable in [
+ BridgeError::Timeout,
+ BridgeError::Iggy(IggyError::Disconnected),
+ BridgeError::Iggy(IggyError::TransientNotCommitted),
+ ] {
+ assert_eq!(
+ commit_code(Err(retriable)),
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS
+ );
+ }
+ assert_eq!(
+ commit_code(Err(BridgeError::Iggy(IggyError::TooManyConsumerOffsets))),
+ ERROR_UNKNOWN_SERVER_ERROR
+ );
+ assert_eq!(
+ commit_code(Err(BridgeError::Iggy(IggyError::TopicNameNotFound(
+ "orders".to_string(),
+ "kafka".to_string(),
+ )))),
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION
+ );
+ }
+
/// A partition left out of the answer reads as committed, so nothing short of a full answer
/// may go back.
#[tokio::test]
diff --git a/gateways/kafka/src/protocol/handlers/offset_fetch.rs b/gateways/kafka/src/protocol/handlers/offset_fetch.rs
index 9c13c14fc..507c6d84d 100644
--- a/gateways/kafka/src/protocol/handlers/offset_fetch.rs
+++ b/gateways/kafka/src/protocol/handlers/offset_fetch.rs
@@ -27,13 +27,15 @@ use kafka_protocol::messages::offset_fetch_response::{
};
use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+use crate::bridge::{BridgeError, IggyBridge};
use crate::error::Result;
use crate::group::MAX_GROUP_ID_BYTES;
use crate::protocol::api::{
API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
- ERROR_INVALID_GROUP_ID, ERROR_NONE, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
- is_supported_version,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState,
+ HandleOutcome, REQUEST_DEADLINE, UNKNOWN_OFFSET, is_supported_version,
};
use crate::protocol::bounds_guard::validate_offset_fetch_shape;
use crate::protocol::handlers::{decode_guarded, encode_message, respond_or_close};
@@ -121,10 +123,6 @@ pub struct PartitionAnswer {
pub error: i16,
}
-#[expect(
- clippy::unused_async,
- reason = "the shared handler signature, kept until this handler awaits the bridge"
-)]
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.
@@ -143,18 +141,116 @@ pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> Hand
};
let queries = GroupQuery::all(api_version, &request);
- let answers: Vec<GroupAnswer> = queries.iter().map(answer).collect();
+ let deadline = Instant::now() + REQUEST_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",
)
}
-fn answer(query: &GroupQuery) -> GroupAnswer {
+/// 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 query.group_id.is_empty() || query.group_id.len() > MAX_GROUP_ID_BYTES {
return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
}
- GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS)
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ let read = async {
+ match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics).await,
+ None => read_every_topic(bridge, &query.group_id).await,
+ }
+ };
+ match timeout_at(deadline, read).await {
+ Ok(Ok(topics)) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Ok(Err(error)) => GroupAnswer::error(error.to_offset_error_code()),
+ Err(_elapsed) => GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ }
+}
+
+/// Every requested partition, in request order. A partition with no offset answers -1.
+async fn read_topics(
+ bridge: &IggyBridge,
+ group: &str,
+ topics: &[(TopicName, Vec<i32>)],
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let mut answers = Vec::with_capacity(topics.len());
+ for (name, indexes) in topics {
+ // A name that cannot map to Iggy names no topic, so nothing is committed under it.
+ let target = bridge.topic_target(name.0.as_str()).ok();
+ let mut partitions = Vec::with_capacity(indexes.len());
+ for &index in indexes {
+ let offset = match (&target, u32::try_from(index)) {
+ (Some(target), Ok(partition)) => {
+ committed(bridge.fetch_group_offset(target, group, partition).await)?
+ }
+ _ => UNKNOWN_OFFSET,
+ };
+ partitions.push(PartitionAnswer {
+ index,
+ offset,
+ error: ERROR_NONE,
+ });
+ }
+ answers.push(TopicAnswer {
+ name: name.clone(),
+ partitions,
+ });
+ }
+ 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,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let mut answers = Vec::new();
+ for topic in bridge.offset_group_topics(group).await? {
+ let target = bridge.topic_target(&topic.kafka_topic)?;
+ let mut partitions = Vec::new();
+ for partition in 0..topic.partitions_count {
+ let offset = committed(bridge.fetch_group_offset(&target, group, partition).await)?;
+ if offset != UNKNOWN_OFFSET {
+ partitions.push(PartitionAnswer {
+ index: i32::try_from(partition).unwrap_or(i32::MAX),
+ offset,
+ error: ERROR_NONE,
+ });
+ }
+ }
+ if !partitions.is_empty() {
+ answers.push(TopicAnswer {
+ name: TopicName(StrBytes::from_string(topic.kafka_topic)),
+ partitions,
+ });
+ }
+ }
+ Ok(answers)
+}
+
+/// 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.
@@ -267,6 +363,7 @@ pub fn encode_response(
#[cfg(test)]
mod tests {
use bytes::BytesMut;
+ use iggy::prelude::IggyError;
use kafka_protocol::messages::offset_fetch_request::{
OffsetFetchRequestGroup, OffsetFetchRequestTopic, OffsetFetchRequestTopics,
};
@@ -349,6 +446,26 @@ mod tests {
);
}
+ /// Only "there is no such offset" may read as -1. A lost read must fail the group instead,
+ /// or the consumer resets its position.
+ #[test]
+ fn given_bridge_reads_only_a_missing_offset_should_answer_minus_one() {
+ assert_eq!(committed(Ok(Some(42))).unwrap(), 42);
+ assert_eq!(committed(Ok(None)).unwrap(), UNKNOWN_OFFSET);
+ assert_eq!(
+ committed(Err(BridgeError::Iggy(IggyError::ResourceNotFound(
+ String::new()
+ ))))
+ .unwrap(),
+ UNKNOWN_OFFSET
+ );
+ let lost = committed(Err(BridgeError::Iggy(IggyError::Disconnected))).unwrap_err();
+ assert_eq!(
+ lost.to_offset_error_code(),
+ ERROR_COORDINATOR_LOAD_IN_PROGRESS
+ );
+ }
+
#[tokio::test]
async fn given_an_unreadable_request_should_close_rather_than_answer() {
let state = GatewayState::stub(BrokerAdvertise::default(), 8 * 1024 * 1024);
diff --git a/gateways/kafka/tests/common/wire.rs b/gateways/kafka/tests/common/wire.rs
index a9a525292..8238f9adc 100644
--- a/gateways/kafka/tests/common/wire.rs
+++ b/gateways/kafka/tests/common/wire.rs
@@ -666,6 +666,37 @@ pub fn build_offset_fetch_request(version: i16, group_id: &str, topic: &str) ->
enc.freeze()
}
+/// `OffsetFetch` request with a null topic list, which asks for every offset of the group.
+/// Version 2 or later.
+pub fn build_offset_fetch_all_request(version: i16, group_id: &str) -> Bytes {
+ let flexible = version >= 6;
+ let mut enc = Encoder::with_capacity(32);
+
+ if version >= 8 {
+ write_array_count(&mut enc, true, 1); // groups
+ }
+ write_string(&mut enc, flexible, Some(group_id));
+ if version >= 9 {
+ write_string(&mut enc, true, None); // member_id
+ enc.write_i32(-1); // member_epoch
+ }
+ if flexible {
+ enc.write_varint(0); // topics: null
+ } else {
+ enc.write_i32(-1); // topics: null
+ }
+ if version >= 8 {
+ enc.write_empty_tagged_fields(); // group
+ }
+ if version >= 7 {
+ enc.write_bool(false); // require_stable
+ }
+ if flexible {
+ enc.write_empty_tagged_fields(); // request
+ }
+ enc.freeze()
+}
+
/// `LeaveGroup` request. Each member is `(member_id, group_instance_id, reason)`. Below v3 the
/// body carries one member id and no identities array, so only `members[0].0` is written.
pub fn build_leave_group_request(
diff --git a/gateways/kafka/tests/offsets_real_bridge_tests.rs b/gateways/kafka/tests/offsets_real_bridge_tests.rs
new file mode 100644
index 000000000..e4ee39905
--- /dev/null
+++ b/gateways/kafka/tests/offsets_real_bridge_tests.rs
@@ -0,0 +1,295 @@
+// 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_all_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, 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");
+ let _metadata = d.read_compact_nullable_string().expect("metadata");
+ let error = d.read_i16().expect("error_code");
+ d.read_tagged_fields().expect("partition tagged fields");
+ partitions.push((name.clone(), index, offset, error));
+ }
+ d.read_tagged_fields().expect("topic tagged fields");
+ }
+ let group_error = d.read_i16().expect("group error_code");
+ (group_error, partitions)
+}
+
+fn iggy_id(name: &str) -> Identifier {
+ Identifier::named(name).expect("valid Iggy name")
+}
+
+/// A Java consumer on defaults commits 0 on an empty partition, and a Kafka commit can sit past
+/// the last message. Both are stored as sent.
+#[tokio::test]
+#[serial]
+async fn given_an_empty_partition_when_a_group_commits_should_store_each_offset_as_sent() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+
+ for offset in [0, 42, 1 << 40] {
+ assert_eq!(commit(&state, "orders", offset).await, ERROR_NONE);
+ assert_eq!(
+ fetch(&state, GROUP, "orders").await,
+ (ERROR_NONE, offset, ERROR_NONE)
+ );
+ }
+}
+
+/// The offset sits under the external group kind of Iggy group `kafka.cg.<group>`, apart from
+/// that group's own offsets.
+#[tokio::test]
+#[serial]
+async fn given_a_commit_the_offset_should_live_under_an_external_group_in_iggy() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+
+ assert_eq!(commit(&state, "orders", 7).await, ERROR_NONE);
+
+ let raw = iggy_server::raw_client(&server).await;
+ let group = iggy_id(&format!("{OFFSET_GROUP_PREFIX}{GROUP}"));
+ let (stream, topic) = (iggy_id("kafka"), iggy_id("orders"));
+ assert!(
+ raw.get_consumer_group(&stream, &topic, &group)
+ .await
+ .expect("read the group")
+ .is_some(),
+ "the first commit creates the Iggy group"
+ );
+ let stored = |consumer: Consumer| {
+ let (raw, stream, topic) = (&raw, &stream, &topic);
+ async move {
+ raw.get_consumer_offset(&consumer, stream, topic, Some(0))
+ .await
+ .expect("read the offset")
+ .map(|info| info.stored_offset)
+ }
+ };
+ assert_eq!(
+ stored(Consumer::external_group(group.clone())).await,
+ Some(7)
+ );
+ assert_eq!(stored(Consumer::group(group)).await, None);
+}
+
+/// Kafka consumers read any negative offset as none, so a negative commit deletes the key. A key
+/// or a group that is not there counts as deleted.
+#[tokio::test]
+#[serial]
+async fn given_a_negative_commit_the_offset_should_be_deleted() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+
+ assert_eq!(commit(&state, "orders", -1).await, ERROR_NONE);
+ assert_eq!(commit(&state, "orders", 7).await, ERROR_NONE);
+ assert_eq!(commit(&state, "orders", -1).await, ERROR_NONE);
+ assert_eq!(
+ fetch(&state, GROUP, "orders").await,
+ (ERROR_NONE, UNKNOWN_OFFSET, ERROR_NONE)
+ );
+ assert_eq!(commit(&state, "orders", -1).await, ERROR_NONE);
+}
+
+/// No key, no group and no topic all read as -1 with no error, as Kafka answers them.
+#[tokio::test]
+#[serial]
+async fn given_nothing_committed_fetch_should_answer_minus_one_without_error() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+ let state = connected_state(&server).await;
+
+ assert_eq!(
+ fetch(&state, "never-committed", "orders").await,
+ (ERROR_NONE, UNKNOWN_OFFSET, ERROR_NONE)
+ );
+ assert_eq!(
+ fetch(&state, GROUP, "no-such-topic").await,
+ (ERROR_NONE, UNKNOWN_OFFSET, ERROR_NONE)
+ );
+}
+
+/// The acceptance rule of #3542: committed offsets survive a gateway restart. A new gateway is a
+/// new bridge with no memory of the old one.
+#[tokio::test]
+#[serial]
+async fn given_a_restarted_gateway_fetch_should_return_the_committed_offset() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ seed_topic(&server, "orders").await;
+
+ let first = connected_state(&server).await;
+ assert_eq!(commit(&first, "orders", 1_234).await, ERROR_NONE);
+ drop(first);
+
+ let restarted = connected_state(&server).await;
+ assert_eq!(
+ fetch(&restarted, GROUP, "orders").await,
+ (ERROR_NONE, 1_234, ERROR_NONE)
+ );
+}
+
+/// The Java `Admin.listConsumerGroupOffsets` asks for a whole group with a null topic list. Only
+/// the topics that hold the group come back, with only their committed partitions.
+#[tokio::test]
+#[serial]
+async fn given_a_null_topic_list_fetch_should_answer_every_committed_partition() {
+ let data_dir = tempfile::tempdir().expect("tempdir");
+ let server = TestServer::spawn(data_dir.path()).await;
+ for topic in ["orders", "payments", "untouched"] {
+ seed_topic(&server, topic).await;
+ }
+ let state = connected_state(&server).await;
+ assert_eq!(commit(&state, "orders", 3).await, ERROR_NONE);
+ assert_eq!(commit(&state, "payments", 5).await, ERROR_NONE);
+
+ let body = build_offset_fetch_all_request(FETCH_VERSION, GROUP);
+ let response = offset_fetch::handle(&state, FETCH_VERSION, body)
+ .await
+ .expect_response("OffsetFetch answers");
+ let (group_error, mut partitions) = decode_fetch(response);
+
+ partitions.sort();
+ assert_eq!(group_error, ERROR_NONE);
+ assert_eq!(
+ partitions,
+ vec![
+ ("orders".to_string(), 0, 3, ERROR_NONE),
+ ("payments".to_string(), 0, 5, ERROR_NONE),
+ ]
+ );
+}