You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by "krishvishal (via GitHub)" <gi...@apache.org> on 2026/10/06 11:00:38 UTC
[PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
krishvishal opened a new pull request, #4415:
URL: https://github.com/apache/iggy/pull/4415
Closes #3542
With `IGGY_KAFKA_BRIDGE_ENABLED=true`, OffsetCommit (8) and OffsetFetch (9) now keep group offsets in Iggy, using the `ExternalGroup` kind from #4392. A group resumes from its last commit, also after a gateway restart, and `kafka-console-consumer.sh --group` commits and resumes.
### How offsets are stored
- key: the partition, plus Iggy consumer group `kafka.cg.<group>` on the mapped topic. The gateway creates that group on the first commit and never joins it
- value: the Kafka offset as sent, no range check. 0 on an empty partition and one past the last message both work
- a negative offset deletes the key. Metadata and leader epoch are dropped, OffsetFetch returns `""` and -1
- nothing deletes these groups on its own. `OFFSET_STORAGE.md` has the `iggy consumer-group delete` command for cleanup
### Commit and fetch
- OffsetCommit v2-v9 checks the committer the way Kafka's `validateOffsetCommit` does. After a gateway restart a member gets `UNKNOWN_MEMBER_ID` and rejoins. Admin tools commit with a negative generation, which works on an empty group
- OffsetFetch v1-v9, batched groups from v8. A null topic list (`Admin.listConsumerGroupOffsets`) reads every topic that holds the group
- no key, no group or no topic reads as -1 with no error. Any other failed read fails the whole group instead, because a consumer that gets -1 resets its position
- retriable failures go back as 14. Java fails offset calls on 6, and OffsetFetch on 7
- anything sent as -1 is logged at `error!` with the group and the cause
### Iggy calls
- offset calls have their own 4 Iggy clients and don't wait behind Produce
- each partition sticks to one client, which keeps its calls in order even when a caller gives up
- a request gives up after half the min session timeout (3s by default). Heartbeats on the same connection wait behind it. Partitions cut off by that answer 14 without a call
### Server change
`get_consumer_offset` answered a read that timed out on the owning shard with an empty body. The SDK reads that as "no offset", and a consumer would reset. It now answers `TransientNotAccepted`, which the Rust SDK retries. This is in `core/server/src/dispatch/partition.rs` and applies to every client.
### Known limits
- every Iggy server needs a build with kind 3 (#4392) before the first commit. Nothing enforces that yet, and in a partly upgraded cluster one commit can stop a partition from committing
- in a cluster an offset read can land on a node that lags. Right after a group's first commit that node answers -1, and a consumer on `auto.offset.reset=latest` skips records. Reads from the primary are #4409
- delivery stays at-least-once
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on PR #4415:
URL: https://github.com/apache/iggy/pull/4415#issuecomment-6029436076
/author
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on code in PR #4415:
URL: https://github.com/apache/iggy/pull/4415#discussion_r4203181217
##########
gateways/kafka/src/protocol/handlers/offset_commit.rs:
##########
@@ -0,0 +1,314 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetCommit` (API key 8).
+//!
+//! Checks the committer against its group, then stores each offset in Iggy
+//! (`docs/OFFSET_STORAGE.md`). The response has no top-level error, so a group answer goes on
+//! every partition.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_commit_response::{
+ OffsetCommitResponsePartition, OffsetCommitResponseTopic,
+};
+use kafka_protocol::messages::{OffsetCommitRequest, OffsetCommitResponse};
+use tokio::time::Instant;
+
+use crate::bridge::IggyBridge;
+use crate::error::Result;
+use crate::group::CommitRequest;
+use crate::protocol::api::{
+ API_KEY_OFFSET_COMMIT, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_commit_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_COMMIT,
+ min_version: 2,
+ max_version: 9,
+};
+
+/// Partition index, error code and tagged fields of one echoed partition.
+const PER_PARTITION_BYTES: usize = 7;
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A client reads a partition left out of the answer as committed, so a request that cannot be
+ // answered partition by partition closes the connection instead.
+ if !is_supported_version(API_KEY_OFFSET_COMMIT, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetCommitRequest>(api_version, body, |version, body| {
+ validate_offset_commit_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetCommit request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let error = state
+ .groups
+ .validate_commit(&CommitRequest::from(&request))
+ .await;
+ let topics = match (&state.bridge, error) {
+ (Some(bridge), ERROR_NONE) => {
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ commit_all(bridge, &request, deadline).await
+ }
+ (None, ERROR_NONE) => same_code(&request, ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ (_, error) => same_code(&request, error),
+ };
+ respond_or_close(encode_response(api_version, topics), "OffsetCommit")
+}
+
+/// Stores every requested offset. The partitions run at once across the offset slots, and a
+/// partition that the deadline leaves out answers 14 and makes no call.
+async fn commit_all(
+ bridge: &IggyBridge,
+ request: &OffsetCommitRequest,
+ deadline: Instant,
+) -> Vec<OffsetCommitResponseTopic> {
+ let group = request.group_id.0.as_str();
+ let targets: Vec<_> = request
+ .topics
+ .iter()
+ .map(|topic| bridge.topic_target(topic.name.0.as_str()))
+ .collect();
+ let mut topics = same_code(request, ERROR_NONE);
+ let mut commits = Vec::new();
+ // Where each commit's code goes: (topic, partition) in `topics`.
+ let mut positions = Vec::new();
+ for (topic_at, (topic, target)) in request.topics.iter().zip(&targets).enumerate() {
+ for (partition_at, partition) in topic.partitions.iter().enumerate() {
+ let code = match (target, u32::try_from(partition.partition_index)) {
+ (Err(error), _) => error.to_offset_error_code(),
+ (Ok(_), Err(_)) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ (Ok(target), Ok(index)) => {
+ commits.push((target, index, partition.committed_offset));
+ positions.push((topic_at, partition_at));
+ continue;
+ }
+ };
+ topics[topic_at].partitions[partition_at].error_code = code;
+ }
+ }
+
+ let results = bridge.commit_group_offsets(group, &commits, deadline).await;
+ let mut refused = 0_usize;
+ let mut first_refusal = None;
+ for ((topic_at, partition_at), result) in positions.into_iter().zip(results) {
+ let Err(error) = result else {
+ continue;
+ };
+ let code = error.to_offset_error_code();
Review Comment:
RequestAlreadyApplied counts as done now for store, delete and the group create. fwiw since #4387 the server answers an aged out replay with RequestTooOld, so offset calls map that to 14 too
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,449 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ let mut started = Vec::with_capacity(commits.len());
+ for &(topic, partition, offset) in commits {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, commit)
+ .await,
+ );
+ }
+ finish_all(started, deadline).await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let mut started = Vec::with_capacity(reads.len());
+ for &(topic, partition) in reads {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(&names.consumer, &stream_id, &topic_id, Some(partition))
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, read)
+ .await,
+ );
+ }
+ finish_all(started, deadline).await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let (stream_id, topic_id) = (topic.stream_id.clone(), topic.topic_id.clone());
+ let lookup = move |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.start_offset_call(topic, 0, deadline, lookup)
+ .await
+ .finish(deadline)
+ .await
+ }
+
+ /// Starts `call` with the client of the slot of `partition` of `topic`, and returns without
+ /// waiting for it to end. A call that has not started by `deadline` never starts.
+ async fn start_offset_call<T, F>(
+ &self,
+ topic: &TopicTarget,
+ partition: u32,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Started<T>
+ where
+ T: Send + 'static,
+ F: Future<Output = Result<T, BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(topic, partition, deadline).await else {
+ return Started(Err(BridgeError::Timeout));
+ };
+ match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(Ok(client)) => Started(Ok(spawn_holding(turn, call(client)))),
+ Ok(Err(error)) => Started(Err(error)),
+ Err(_elapsed) => Started(Err(BridgeError::Timeout)),
+ }
+ }
+}
+
+/// The offset slots. A partition always takes the same slot, and a call keeps its slot until the
+/// SDK lets go of it, so the calls of one partition reach Iggy in the order they started.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// The slot of `partition` of `topic`, with its client. `None` if it is not free by
+ /// `deadline`.
+ async fn take(
+ &self,
+ topic: &TopicTarget,
+ partition: u32,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot_of(topic, partition)];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
+ usize::from(hasher.finish().to_le_bytes()[0]) % OFFSET_SLOTS
+}
+
+/// Runs `call` in a task that keeps `turn` until `call` ends, even after its caller stops
+/// waiting, so the next call on the slot cannot overtake it.
+fn spawn_holding<T: Send + 'static>(
+ turn: OwnedSemaphorePermit,
+ call: impl Future<Output = Result<T, BridgeError>> + Send + 'static,
+) -> oneshot::Receiver<Result<T, BridgeError>> {
+ let (sender, receiver) = oneshot::channel();
+ tokio::spawn(async move {
+ let ended = timeout(SLOT_LIMIT, call)
+ .await
+ .unwrap_or(Err(BridgeError::Timeout));
+ // The caller may have stopped waiting.
+ let _ = sender.send(ended);
+ drop(turn);
+ });
+ receiver
+}
+
+/// An offset call that keeps its slot, or why it never started.
+struct Started<T>(Result<oneshot::Receiver<Result<T, BridgeError>>, BridgeError>);
+
+impl<T> Started<T> {
+ /// The call's result. [`BridgeError::Timeout`] if it does not end by `deadline`.
+ async fn finish(self, deadline: Instant) -> Result<T, BridgeError> {
+ match timeout_at(deadline, self.0?).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
Review Comment:
yeah cross-gateway only. an epoch needs a conditional store on the server, that's a wire change. I'd rather do a follow-up issue, documented for now
##########
gateways/kafka/docs/kafka_api_keys_reference.md:
##########
Review Comment:
fixed, both are Bridge now and the stub count is 11
##########
gateways/kafka/README.md:
##########
Review Comment:
added the fold next to the Fetch line
##########
gateways/kafka/tests/offsets_real_bridge_tests.rs:
##########
@@ -0,0 +1,293 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetCommit` and `OffsetFetch` against a real `iggy-server` process, through the handlers
+//! with a connected [`GatewayState`]. The offsets live in Iggy as external group offsets, so they
+//! outlive the gateway. Responses are decoded by hand for the reason that
+//! `create_topics_real_bridge_tests.rs` gives.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use iggy::prelude::{Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier};
+use serial_test::serial;
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{IggyBridge, OFFSET_GROUP_PREFIX};
+use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::api::{BrokerAdvertise, ERROR_NONE, GatewayState};
+use iggy_gateway_kafka::protocol::handlers::{offset_commit, offset_fetch};
+
+use codec::Decoder;
+use iggy_server::TestServer;
+use wire::{build_offset_commit_request, build_offset_fetch_request};
+
+const COMMIT_VERSION: i16 = 8;
+const FETCH_VERSION: i16 = 8;
+const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+const GROUP: &str = "orders-app";
+const UNKNOWN_OFFSET: i64 = -1;
+
+async fn connected_state(server: &TestServer) -> GatewayState {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ TEST_MAX_FRAME_SIZE,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+}
+
+async fn seed_topic(server: &TestServer, topic: &str) {
+ IggyBridge::connect(server.test_config())
+ .await
+ .expect("seed bridge should connect")
+ .ensure_stream_and_topic(topic, 1)
+ .await
+ .expect("seed the topic");
+}
+
+/// Commits `offset` for partition 0 of `topic`, outside the group protocol, and returns its code.
+async fn commit(state: &GatewayState, topic: &str, offset: i64) -> i16 {
+ let body = build_offset_commit_request(COMMIT_VERSION, GROUP, topic, offset);
+ let response = offset_commit::handle(state, COMMIT_VERSION, body)
+ .await
+ .expect_response("OffsetCommit answers");
+ let codes = decode_commit_codes(response);
+ assert_eq!(codes.len(), 1, "one partition was committed");
+ codes[0]
+}
+
+/// The group error and partition 0's `(offset, error)` for `group` on `topic`.
+async fn fetch(state: &GatewayState, group: &str, topic: &str) -> (i16, i64, i16) {
+ let body = build_offset_fetch_request(FETCH_VERSION, group, Some(topic));
+ let response = offset_fetch::handle(state, FETCH_VERSION, body)
+ .await
+ .expect_response("OffsetFetch answers");
+ let (group_error, partitions) = decode_fetch(response);
+ assert_eq!(partitions.len(), 1, "partition 0 was asked for");
+ let (_, index, offset, error) = &partitions[0];
+ assert_eq!(*index, 0);
+ (group_error, *offset, *error)
+}
+
+/// Decodes a v8 `OffsetCommit` response into one code per partition, in wire order.
+fn decode_commit_codes(body: Bytes) -> Vec<i16> {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let mut codes = Vec::new();
+ for _ in 1..d.read_varint().expect("topics") {
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ for _ in 1..d.read_varint().expect("partitions") {
+ let _index = d.read_i32().expect("partition_index");
+ codes.push(d.read_i16().expect("error_code"));
+ d.read_tagged_fields().expect("partition tagged fields");
+ }
+ d.read_tagged_fields().expect("topic tagged fields");
+ }
+ codes
+}
+
+/// Decodes a one-group v8 `OffsetFetch` response into the group error and every partition as
+/// `(topic, index, offset, error)`.
+fn decode_fetch(body: Bytes) -> (i16, Vec<(String, i32, i64, i16)>) {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ assert_eq!(d.read_varint().expect("groups"), 2, "one group");
+ let _group_id = d.read_compact_nullable_string().expect("group_id");
+ let mut partitions = Vec::new();
+ for _ in 1..d.read_varint().expect("topics") {
+ let name = d
+ .read_compact_nullable_string()
+ .expect("topic name")
+ .expect("a topic name is never null");
+ for _ in 1..d.read_varint().expect("partitions") {
+ let index = d.read_i32().expect("partition_index");
+ let offset = d.read_i64().expect("committed_offset");
+ let _leader_epoch = d.read_i32().expect("committed_leader_epoch");
Review Comment:
done, decode asserts epoch -1 and empty metadata now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #4415:
URL: https://github.com/apache/iggy/pull/4415#issuecomment-6015239043
## [Codecov](https://app.codecov.io/gh/apache/iggy/pull/4415?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `98.91304%` with `1 line` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 75.83%. Comparing base ([`f6dfbdb`](https://app.codecov.io/gh/apache/iggy/commit/f6dfbdbb379be2140c453c0702975c256bc4f995?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`86ae086`](https://app.codecov.io/gh/apache/iggy/commit/86ae086d43223bba550396a78e5c57ee599e716a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4415?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [core/server/src/dispatch/partition.rs](https://app.codecov.io/gh/apache/iggy/pull/4415?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Fpartition.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL3BhcnRpdGlvbi5ycw==) | 98.91% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/iggy/pull/4415?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #4415 +/- ##
=============================================
- Coverage 87.91% 75.83% -12.09%
- Complexity 1578 1580 +2
=============================================
Files 1298 1298
Lines 233811 233857 +46
Branches 194562 194606 +44
=============================================
- Hits 205563 177348 -28215
- Misses 23339 51579 +28240
- Partials 4909 4930 +21
```
| [Components](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [Rust Core](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.29% <98.91%> (-14.85%)` | :arrow_down: |
| [Java SDK](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `68.80% <ø> (+0.08%)` | :arrow_up: |
| [C# SDK](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `77.67% <ø> (+0.06%)` | :arrow_up: |
| [Python SDK](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `91.24% <ø> (ø)` | |
| [PHP SDK](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `85.67% <ø> (ø)` | |
| [Node SDK](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `96.59% <ø> (+1.70%)` | :arrow_up: |
| [Go SDK](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.29% <ø> (+0.02%)` | :arrow_up: |
| [C++ SDK](https://app.codecov.io/gh/apache/iggy/pull/4415/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.60% <ø> (ø)` | |
| [Files with missing lines](https://app.codecov.io/gh/apache/iggy/pull/4415?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [core/server/src/dispatch/partition.rs](https://app.codecov.io/gh/apache/iggy/pull/4415?src=pr&el=tree&filepath=core%2Fserver%2Fsrc%2Fdispatch%2Fpartition.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29yZS9zZXJ2ZXIvc3JjL2Rpc3BhdGNoL3BhcnRpdGlvbi5ycw==) | `83.59% <98.91%> (-11.69%)` | :arrow_down: |
... and [424 files with indirect coverage changes](https://app.codecov.io/gh/apache/iggy/pull/4415/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "ryerraguntla (via GitHub)" <gi...@apache.org>.
ryerraguntla commented on code in PR #4415:
URL: https://github.com/apache/iggy/pull/4415#discussion_r4202258240
##########
core/server/src/dispatch/partition.rs:
##########
@@ -729,6 +730,24 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
.await;
return;
}
+ // A missing reply says nothing about the offset, so it must
+ // not read as "no stored offset". A read moves no progress,
+ // so the client may retry it.
+ None => {
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ "consumer offset read reply unavailable; replying not accepted"
+ );
+ send_non_replicated_deny(
+ shard,
+ request,
+ transport_client_id,
+ IggyError::TransientNotAccepted.as_code(),
+ )
+ .await;
+ return;
+ }
_ => Bytes::new(),
Review Comment:
NotFound becomes an empty success body, so OffsetFetch answers -1 and auto.offset.reset=latest skips. Fix: deny NotFound with TransientNotAccepted. Empty body only for ConsumerOffset
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
Review Comment:
v8 groups share one deadline in series. Tail groups get retriable 14 and never call Iggy. Java retries 14, so a head group that eats the 3s budget livelocks the tail. Fix: start the group reads together under the same deadline.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,449 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ let mut started = Vec::with_capacity(commits.len());
+ for &(topic, partition, offset) in commits {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, commit)
+ .await,
Review Comment:
partition offset calls start one at a time, so a busy slot blocks partitions whose slots are free. Fix: start every partition, then wait in finish. The slot semaphore already keeps order.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
Review Comment:
the handler holds the connection until the offset deadline, so the heartbeat waits. A fetch that starts with 3s or less left burns the rest of the session and the member expires. Fix: cap the hold by the time left on that member session, or read the heartbeat while the offset call runs.
##########
gateways/kafka/README.md:
##########
Review Comment:
[nit] - the error map records the Fetch fold (7→6) and omits the OffsetCommit/OffsetFetch fold (6 and 7 → 14 via to_offset_error_code). Fix: add that fold beside the Fetch sentence.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,449 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ let mut started = Vec::with_capacity(commits.len());
+ for &(topic, partition, offset) in commits {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, commit)
+ .await,
+ );
+ }
+ finish_all(started, deadline).await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let mut started = Vec::with_capacity(reads.len());
+ for &(topic, partition) in reads {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(&names.consumer, &stream_id, &topic_id, Some(partition))
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, read)
+ .await,
+ );
+ }
+ finish_all(started, deadline).await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let (stream_id, topic_id) = (topic.stream_id.clone(), topic.topic_id.clone());
+ let lookup = move |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&stream_id, &topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.start_offset_call(topic, 0, deadline, lookup)
+ .await
+ .finish(deadline)
+ .await
+ }
+
+ /// Starts `call` with the client of the slot of `partition` of `topic`, and returns without
+ /// waiting for it to end. A call that has not started by `deadline` never starts.
+ async fn start_offset_call<T, F>(
+ &self,
+ topic: &TopicTarget,
+ partition: u32,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Started<T>
+ where
+ T: Send + 'static,
+ F: Future<Output = Result<T, BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(topic, partition, deadline).await else {
+ return Started(Err(BridgeError::Timeout));
+ };
+ match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(Ok(client)) => Started(Ok(spawn_holding(turn, call(client)))),
+ Ok(Err(error)) => Started(Err(error)),
+ Err(_elapsed) => Started(Err(BridgeError::Timeout)),
+ }
+ }
+}
+
+/// The offset slots. A partition always takes the same slot, and a call keeps its slot until the
+/// SDK lets go of it, so the calls of one partition reach Iggy in the order they started.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// The slot of `partition` of `topic`, with its client. `None` if it is not free by
+ /// `deadline`.
+ async fn take(
+ &self,
+ topic: &TopicTarget,
+ partition: u32,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot_of(topic, partition)];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
+ usize::from(hasher.finish().to_le_bytes()[0]) % OFFSET_SLOTS
+}
+
+/// Runs `call` in a task that keeps `turn` until `call` ends, even after its caller stops
+/// waiting, so the next call on the slot cannot overtake it.
+fn spawn_holding<T: Send + 'static>(
+ turn: OwnedSemaphorePermit,
+ call: impl Future<Output = Result<T, BridgeError>> + Send + 'static,
+) -> oneshot::Receiver<Result<T, BridgeError>> {
+ let (sender, receiver) = oneshot::channel();
+ tokio::spawn(async move {
+ let ended = timeout(SLOT_LIMIT, call)
+ .await
+ .unwrap_or(Err(BridgeError::Timeout));
+ // The caller may have stopped waiting.
+ let _ = sender.send(ended);
+ drop(turn);
+ });
+ receiver
+}
+
+/// An offset call that keeps its slot, or why it never started.
+struct Started<T>(Result<oneshot::Receiver<Result<T, BridgeError>>, BridgeError>);
+
+impl<T> Started<T> {
+ /// The call's result. [`BridgeError::Timeout`] if it does not end by `deadline`.
+ async fn finish(self, deadline: Instant) -> Result<T, BridgeError> {
+ match timeout_at(deadline, self.0?).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
Review Comment:
cross-gateway only. finish returns Timeout while spawn_holding still stores or deletes. Another gateway can land a newer commit first, and this write rewinds or deletes it. A deleted key fetches as -1. Fix: store a commit epoch and reject a stale write. Same-gateway rewind does not happen: the slot permit is held until the call ends.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
+ answers.push(answer(state.bridge.as_deref(), query, deadline).await);
+ }
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ let read = async {
+ match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ }
+ };
+ match timeout_at(deadline, read).await {
Review Comment:
one timeout_at wraps the whole group read. Topics already read are dropped, and the slot stays held up to 45s, so a Java retry of 14 cannot enter. Fix: per-item deadlines, keep finished topics, release the slot when the fetch is dropped.
##########
gateways/kafka/docs/kafka_api_keys_reference.md:
##########
Review Comment:
[nit] - Group 3 still marks OffsetCommit and OffsetFetch as required stubs. Line 271 still counts keys 8–14 as stubs. Line 313 says both are supported. Fix: mark them implemented and drop them from the stub count.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
+ answers.push(answer(state.bridge.as_deref(), query, deadline).await);
+ }
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ let read = async {
+ match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ }
+ };
+ match timeout_at(deadline, read).await {
+ Ok(Ok(topics)) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Ok(Err(error)) => GroupAnswer::error(refusal(&query.group_id, &error)),
+ Err(_elapsed) => GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ }
+}
+
+/// The group code for a failed read. A -1 is logged, since the client sees no more.
+fn refusal(group: &str, error: &BridgeError) -> i16 {
+ let code = error.to_offset_error_code();
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ tracing::error!(
+ group,
+ %error,
+ cause = offset_refusal_cause(error),
+ "Iggy refused a Kafka offset read"
+ );
+ } else {
+ tracing::debug!(group, %error, "Iggy refused a Kafka offset read");
+ }
+ code
+}
+
+/// Every requested partition, in request order. A partition with no offset answers -1. The reads
+/// run at once across the offset slots.
+async fn read_topics(
+ bridge: &IggyBridge,
+ group: &str,
+ topics: &[(TopicName, Vec<i32>)],
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ // A name that cannot map to Iggy names no topic, so nothing is committed under it.
+ let targets: Vec<Option<TopicTarget>> = topics
+ .iter()
+ .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok())
+ .collect();
+ let mut answers: Vec<TopicAnswer> = topics
+ .iter()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: ERROR_NONE,
+ })
+ .collect(),
+ })
+ .collect();
+ let mut reads = Vec::new();
+ // Where each read's offset goes: (topic, partition) in `answers`.
+ let mut positions = Vec::new();
+ for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() {
+ let Some(target) = target else {
+ continue;
+ };
+ for (partition_at, &index) in indexes.iter().enumerate() {
+ if let Ok(partition) = u32::try_from(index) {
+ reads.push((target, partition));
+ positions.push((topic_at, partition_at));
+ }
+ }
+ }
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) {
+ answers[topic_at].partitions[partition_at].offset = committed(read)?;
+ }
+ Ok(answers)
+}
+
+/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only
+/// the partitions with an offset, on every topic that holds the group.
+async fn read_every_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let mut answers = Vec::new();
+ for topic in bridge.list_kafka_topics().await? {
Review Comment:
a null topic list uses the lockstep Produce client, so OffsetFetch blocks Produce. Docs say offset calls do not. Fix: list topics on an offset-slot client or probe_client.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -1112,6 +1119,61 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// Kafka's `ClassicGroup.validateOffsetCommit`. Never creates a group.
+///
+/// A group with no members is Kafka's `Empty`: a negative generation commits without a member. A
+/// gateway restart empties every group, so a member's next commit gets `UNKNOWN_MEMBER_ID` and it
+/// rejoins. Kafka's v9 answer, `GROUP_ID_NOT_FOUND`, is fatal in the Java consumer.
+pub fn commit_step(groups: &mut Groups, request: &CommitRequest, now: Instant) -> i16 {
+ if !is_valid_group_id(&request.group_id) {
+ return ERROR_INVALID_GROUP_ID;
+ }
+ let empty = || {
+ if request.generation_id < 0 {
Review Comment:
generation -1 on a gateway with no local group returns ERROR_NONE and then stores on the shared kafka.cg.<group>, overwriting another gateway's live offsets. Fix: refuse generation < 0 unless this process has the group, or fence the stale write.
##########
gateways/kafka/src/protocol/handlers/offset_commit.rs:
##########
@@ -0,0 +1,314 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetCommit` (API key 8).
+//!
+//! Checks the committer against its group, then stores each offset in Iggy
+//! (`docs/OFFSET_STORAGE.md`). The response has no top-level error, so a group answer goes on
+//! every partition.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_commit_response::{
+ OffsetCommitResponsePartition, OffsetCommitResponseTopic,
+};
+use kafka_protocol::messages::{OffsetCommitRequest, OffsetCommitResponse};
+use tokio::time::Instant;
+
+use crate::bridge::IggyBridge;
+use crate::error::Result;
+use crate::group::CommitRequest;
+use crate::protocol::api::{
+ API_KEY_OFFSET_COMMIT, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS, ERROR_NONE,
+ ERROR_UNKNOWN_SERVER_ERROR, ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_commit_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_COMMIT,
+ min_version: 2,
+ max_version: 9,
+};
+
+/// Partition index, error code and tagged fields of one echoed partition.
+const PER_PARTITION_BYTES: usize = 7;
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A client reads a partition left out of the answer as committed, so a request that cannot be
+ // answered partition by partition closes the connection instead.
+ if !is_supported_version(API_KEY_OFFSET_COMMIT, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetCommitRequest>(api_version, body, |version, body| {
+ validate_offset_commit_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetCommit request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let error = state
+ .groups
+ .validate_commit(&CommitRequest::from(&request))
+ .await;
+ let topics = match (&state.bridge, error) {
+ (Some(bridge), ERROR_NONE) => {
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ commit_all(bridge, &request, deadline).await
+ }
+ (None, ERROR_NONE) => same_code(&request, ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ (_, error) => same_code(&request, error),
+ };
+ respond_or_close(encode_response(api_version, topics), "OffsetCommit")
+}
+
+/// Stores every requested offset. The partitions run at once across the offset slots, and a
+/// partition that the deadline leaves out answers 14 and makes no call.
+async fn commit_all(
+ bridge: &IggyBridge,
+ request: &OffsetCommitRequest,
+ deadline: Instant,
+) -> Vec<OffsetCommitResponseTopic> {
+ let group = request.group_id.0.as_str();
+ let targets: Vec<_> = request
+ .topics
+ .iter()
+ .map(|topic| bridge.topic_target(topic.name.0.as_str()))
+ .collect();
+ let mut topics = same_code(request, ERROR_NONE);
+ let mut commits = Vec::new();
+ // Where each commit's code goes: (topic, partition) in `topics`.
+ let mut positions = Vec::new();
+ for (topic_at, (topic, target)) in request.topics.iter().zip(&targets).enumerate() {
+ for (partition_at, partition) in topic.partitions.iter().enumerate() {
+ let code = match (target, u32::try_from(partition.partition_index)) {
+ (Err(error), _) => error.to_offset_error_code(),
+ (Ok(_), Err(_)) => ERROR_UNKNOWN_TOPIC_OR_PARTITION,
+ (Ok(target), Ok(index)) => {
+ commits.push((target, index, partition.committed_offset));
+ positions.push((topic_at, partition_at));
+ continue;
+ }
+ };
+ topics[topic_at].partitions[partition_at].error_code = code;
+ }
+ }
+
+ let results = bridge.commit_group_offsets(group, &commits, deadline).await;
+ let mut refused = 0_usize;
+ let mut first_refusal = None;
+ for ((topic_at, partition_at), result) in positions.into_iter().zip(results) {
+ let Err(error) = result else {
+ continue;
+ };
+ let code = error.to_offset_error_code();
Review Comment:
a replayed store or delete that already landed (RequestAlreadyApplied) becomes fatal -1. Classic OffsetCommit treats -1 as non-retriable. Fix: map IggyError::RequestAlreadyApplied to ERROR_NONE on the commit path.
##########
gateways/kafka/tests/offsets_real_bridge_tests.rs:
##########
@@ -0,0 +1,293 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetCommit` and `OffsetFetch` against a real `iggy-server` process, through the handlers
+//! with a connected [`GatewayState`]. The offsets live in Iggy as external group offsets, so they
+//! outlive the gateway. Responses are decoded by hand for the reason that
+//! `create_topics_real_bridge_tests.rs` gives.
+
+#[path = "common/codec.rs"]
+mod codec;
+#[path = "common/iggy_server.rs"]
+mod iggy_server;
+#[path = "common/wire.rs"]
+mod wire;
+
+use std::sync::Arc;
+
+use bytes::Bytes;
+use iggy::prelude::{Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier};
+use serial_test::serial;
+use tokio_util::sync::CancellationToken;
+
+use iggy_gateway_kafka::bridge::{IggyBridge, OFFSET_GROUP_PREFIX};
+use iggy_gateway_kafka::group::{GroupCoordinator, GroupCoordinatorConfig};
+use iggy_gateway_kafka::protocol::api::{BrokerAdvertise, ERROR_NONE, GatewayState};
+use iggy_gateway_kafka::protocol::handlers::{offset_commit, offset_fetch};
+
+use codec::Decoder;
+use iggy_server::TestServer;
+use wire::{build_offset_commit_request, build_offset_fetch_request};
+
+const COMMIT_VERSION: i16 = 8;
+const FETCH_VERSION: i16 = 8;
+const TEST_MAX_FRAME_SIZE: usize = 8 * 1024 * 1024;
+const GROUP: &str = "orders-app";
+const UNKNOWN_OFFSET: i64 = -1;
+
+async fn connected_state(server: &TestServer) -> GatewayState {
+ let bridge = IggyBridge::connect(server.test_config())
+ .await
+ .expect("bridge should connect to a ready server");
+ GatewayState::new(
+ BrokerAdvertise::default(),
+ Some(Arc::new(bridge)),
+ TEST_MAX_FRAME_SIZE,
+ false,
+ 0,
+ GroupCoordinator::new(GroupCoordinatorConfig::default(), CancellationToken::new()),
+ )
+}
+
+async fn seed_topic(server: &TestServer, topic: &str) {
+ IggyBridge::connect(server.test_config())
+ .await
+ .expect("seed bridge should connect")
+ .ensure_stream_and_topic(topic, 1)
+ .await
+ .expect("seed the topic");
+}
+
+/// Commits `offset` for partition 0 of `topic`, outside the group protocol, and returns its code.
+async fn commit(state: &GatewayState, topic: &str, offset: i64) -> i16 {
+ let body = build_offset_commit_request(COMMIT_VERSION, GROUP, topic, offset);
+ let response = offset_commit::handle(state, COMMIT_VERSION, body)
+ .await
+ .expect_response("OffsetCommit answers");
+ let codes = decode_commit_codes(response);
+ assert_eq!(codes.len(), 1, "one partition was committed");
+ codes[0]
+}
+
+/// The group error and partition 0's `(offset, error)` for `group` on `topic`.
+async fn fetch(state: &GatewayState, group: &str, topic: &str) -> (i16, i64, i16) {
+ let body = build_offset_fetch_request(FETCH_VERSION, group, Some(topic));
+ let response = offset_fetch::handle(state, FETCH_VERSION, body)
+ .await
+ .expect_response("OffsetFetch answers");
+ let (group_error, partitions) = decode_fetch(response);
+ assert_eq!(partitions.len(), 1, "partition 0 was asked for");
+ let (_, index, offset, error) = &partitions[0];
+ assert_eq!(*index, 0);
+ (group_error, *offset, *error)
+}
+
+/// Decodes a v8 `OffsetCommit` response into one code per partition, in wire order.
+fn decode_commit_codes(body: Bytes) -> Vec<i16> {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ let mut codes = Vec::new();
+ for _ in 1..d.read_varint().expect("topics") {
+ let _name = d.read_compact_nullable_string().expect("topic name");
+ for _ in 1..d.read_varint().expect("partitions") {
+ let _index = d.read_i32().expect("partition_index");
+ codes.push(d.read_i16().expect("error_code"));
+ d.read_tagged_fields().expect("partition tagged fields");
+ }
+ d.read_tagged_fields().expect("topic tagged fields");
+ }
+ codes
+}
+
+/// Decodes a one-group v8 `OffsetFetch` response into the group error and every partition as
+/// `(topic, index, offset, error)`.
+fn decode_fetch(body: Bytes) -> (i16, Vec<(String, i32, i64, i16)>) {
+ let mut d = Decoder::new(body);
+ let _throttle_time_ms = d.read_i32().expect("throttle_time_ms");
+ assert_eq!(d.read_varint().expect("groups"), 2, "one group");
+ let _group_id = d.read_compact_nullable_string().expect("group_id");
+ let mut partitions = Vec::new();
+ for _ in 1..d.read_varint().expect("topics") {
+ let name = d
+ .read_compact_nullable_string()
+ .expect("topic name")
+ .expect("a topic name is never null");
+ for _ in 1..d.read_varint().expect("partitions") {
+ let index = d.read_i32().expect("partition_index");
+ let offset = d.read_i64().expect("committed_offset");
+ let _leader_epoch = d.read_i32().expect("committed_leader_epoch");
Review Comment:
[nit] - decode drops committed_leader_epoch and metadata. Epoch 0 still passes. Java treats 0 as a real epoch. Fix: assert epoch -1 and metadata empty.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on code in PR #4415:
URL: https://github.com/apache/iggy/pull/4415#discussion_r4211081142
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
Review Comment:
fixed. each slot has a read client and a commit client now, and a stuck commit only holds up other commits.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
+ usize::from(hasher.finish().to_le_bytes()[0]) % OFFSET_SLOTS
+}
+
+/// Runs `call` in a task that keeps `turn` until `call` ends, even after its caller stops
+/// waiting, so the next call on the slot cannot overtake it.
+fn spawn_holding<T: Send + 'static>(
+ turn: OwnedSemaphorePermit,
+ call: impl Future<Output = Result<T, BridgeError>> + Send + 'static,
+) -> oneshot::Receiver<Result<T, BridgeError>> {
+ let (sender, receiver) = oneshot::channel();
+ tokio::spawn(async move {
+ let ended = timeout(SLOT_LIMIT, call)
Review Comment:
yeah, that promise was wrong. reworded it in OFFSET_STORAGE.md. reads have their own client now and can't drop the commit one. a late commit just waits behind the stuck one inside the sdk.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,553 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::Instant;
+
+use crate::bridge::{BridgeError, IggyBridge, KafkaTopicMetadata, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_deadline, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = offset_deadline(state, connection).await;
+ // The groups read at once, so a slow group cannot spend the time of the others.
+ let answers = join_all(
+ queries
+ .iter()
+ .map(|query| answer(state.bridge.as_deref(), query, deadline)),
+ )
+ .await;
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ // Each Iggy call stops at `deadline` on its own, so none is cut off holding its slot.
+ let read = match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ };
+ match read {
+ Ok(topics) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Err(error) => GroupAnswer::error(refusal(&query.group_id, &error)),
+ }
+}
+
+/// The group code for a failed read. A -1 is logged, since the client sees no more.
+fn refusal(group: &str, error: &BridgeError) -> i16 {
+ let code = error.to_offset_error_code();
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ tracing::error!(
+ group,
+ %error,
+ cause = offset_refusal_cause(error),
+ "Iggy refused a Kafka offset read"
+ );
+ } else {
+ tracing::debug!(group, %error, "Iggy refused a Kafka offset read");
+ }
+ code
+}
+
+/// Every requested partition, in request order. A partition with no offset answers -1. The reads
+/// run at once across the offset slots.
+async fn read_topics(
+ bridge: &IggyBridge,
+ group: &str,
+ topics: &[(TopicName, Vec<i32>)],
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ // A name that cannot map to Iggy names no topic, so nothing is committed under it.
+ let targets: Vec<Option<TopicTarget>> = topics
+ .iter()
+ .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok())
+ .collect();
+ let mut answers: Vec<TopicAnswer> = topics
+ .iter()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: ERROR_NONE,
+ })
+ .collect(),
+ })
+ .collect();
+ let mut reads = Vec::new();
+ // Where each read's offset goes: (topic, partition) in `answers`.
+ let mut positions = Vec::new();
+ for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() {
+ let Some(target) = target else {
+ continue;
+ };
+ for (partition_at, &index) in indexes.iter().enumerate() {
+ if let Ok(partition) = u32::try_from(index) {
+ reads.push((target, partition));
+ positions.push((topic_at, partition_at));
+ }
+ }
+ }
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) {
+ answers[topic_at].partitions[partition_at].offset = committed(read)?;
+ }
+ Ok(answers)
+}
+
+/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only
+/// the partitions with an offset, on every topic that holds the group. The topics read at once.
+async fn read_every_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let topics = bridge.list_kafka_topics_until(deadline).await?;
+ let answers = join_all(
Review Comment:
added OffsetCalls, one per kafka request. it allows one waiting call per slot, other requests get in after one of its calls. commits use it too.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
Review Comment:
done, the hash is stream + topic now and the partition index is added on top. 4 partitions get 4 slots.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
Review Comment:
added permit_by in the bridge mod, produce uses it too.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -724,12 +725,32 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
stored: Some(stored_offset),
current_offset,
}) => build_consumer_offset_body(partition_id, current_offset, stored_offset),
+ Some(PartitionReadReply::ConsumerOffset { stored: None, .. }) => Bytes::new(),
Some(PartitionReadReply::Rejected(error)) => {
send_non_replicated_deny(shard, request, transport_client_id, error.as_code())
.await;
return;
}
- _ => Bytes::new(),
+ // No reply, or an owner that does not hold the partition yet,
+ // says nothing about the offset, so it must not read as "no
+ // stored offset". A read moves no progress, so the client may
+ // retry it.
+ reply => {
+ warn!(
Review Comment:
NotFound logs at debug now, timeout stays at warn.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,553 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::Instant;
+
+use crate::bridge::{BridgeError, IggyBridge, KafkaTopicMetadata, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_deadline, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = offset_deadline(state, connection).await;
+ // The groups read at once, so a slow group cannot spend the time of the others.
+ let answers = join_all(
+ queries
+ .iter()
+ .map(|query| answer(state.bridge.as_deref(), query, deadline)),
+ )
+ .await;
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ // Each Iggy call stops at `deadline` on its own, so none is cut off holding its slot.
+ let read = match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ };
+ match read {
+ Ok(topics) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Err(error) => GroupAnswer::error(refusal(&query.group_id, &error)),
+ }
+}
+
+/// The group code for a failed read. A -1 is logged, since the client sees no more.
+fn refusal(group: &str, error: &BridgeError) -> i16 {
+ let code = error.to_offset_error_code();
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ tracing::error!(
+ group,
+ %error,
+ cause = offset_refusal_cause(error),
+ "Iggy refused a Kafka offset read"
+ );
+ } else {
+ tracing::debug!(group, %error, "Iggy refused a Kafka offset read");
+ }
+ code
+}
+
+/// Every requested partition, in request order. A partition with no offset answers -1. The reads
+/// run at once across the offset slots.
+async fn read_topics(
+ bridge: &IggyBridge,
+ group: &str,
+ topics: &[(TopicName, Vec<i32>)],
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ // A name that cannot map to Iggy names no topic, so nothing is committed under it.
+ let targets: Vec<Option<TopicTarget>> = topics
+ .iter()
+ .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok())
+ .collect();
+ let mut answers: Vec<TopicAnswer> = topics
+ .iter()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: ERROR_NONE,
+ })
+ .collect(),
+ })
+ .collect();
+ let mut reads = Vec::new();
+ // Where each read's offset goes: (topic, partition) in `answers`.
+ let mut positions = Vec::new();
+ for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() {
+ let Some(target) = target else {
+ continue;
+ };
+ for (partition_at, &index) in indexes.iter().enumerate() {
+ if let Ok(partition) = u32::try_from(index) {
+ reads.push((target, partition));
+ positions.push((topic_at, partition_at));
+ }
+ }
+ }
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) {
+ answers[topic_at].partitions[partition_at].offset = committed(read)?;
+ }
+ Ok(answers)
+}
+
+/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only
+/// the partitions with an offset, on every topic that holds the group. The topics read at once.
+async fn read_every_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let topics = bridge.list_kafka_topics_until(deadline).await?;
+ let answers = join_all(
+ topics
+ .into_iter()
+ .map(|topic| read_listed_topic(bridge, group, topic, deadline)),
+ )
+ .await;
+ answers
+ .into_iter()
+ .filter_map(core::result::Result::transpose)
+ .collect()
+}
+
+/// The partitions of `topic` with an offset of `group`. `None` when there are none.
+async fn read_listed_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ topic: KafkaTopicMetadata,
+ deadline: Instant,
+) -> core::result::Result<Option<TopicAnswer>, BridgeError> {
+ let target = bridge.topic_target(&topic.kafka_topic)?;
+ match bridge.holds_offset_group(group, &target, deadline).await {
+ Ok(true) => {}
+ Ok(false) => return Ok(None),
+ // A topic deleted since the listing holds no offsets.
+ Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => {
+ return Ok(None);
+ }
+ Err(error) => return Err(error),
+ }
+ let reads: Vec<_> = (0..topic.partitions_count)
+ .map(|partition| (&target, partition))
+ .collect();
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ let mut partitions = Vec::new();
+ for (partition, read) in (0..topic.partitions_count).zip(offsets) {
+ let offset = committed(read)?;
+ if offset != UNKNOWN_OFFSET {
+ partitions.push(PartitionAnswer {
+ index: i32::try_from(partition).unwrap_or(i32::MAX),
+ offset,
+ error: ERROR_NONE,
+ });
+ }
+ }
+ Ok((!partitions.is_empty()).then(|| TopicAnswer {
+ name: TopicName(StrBytes::from_string(topic.kafka_topic)),
+ partitions,
+ }))
+}
+
+/// A read as Kafka answers it: -1 for no offset and for a topic or partition that does not exist.
+fn committed(
+ read: core::result::Result<Option<u64>, BridgeError>,
+) -> core::result::Result<i64, BridgeError> {
+ match read {
+ Ok(Some(offset)) => Ok(i64::try_from(offset).unwrap_or(i64::MAX)),
+ Ok(None) => Ok(UNKNOWN_OFFSET),
+ Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => {
+ Ok(UNKNOWN_OFFSET)
+ }
+ Err(error) => Err(error),
+ }
+}
+
+/// One entry per query, in request order. Below v8 there is exactly one.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+fn encode_response(version: i16, queries: &[GroupQuery], answers: &[GroupAnswer]) -> Result<Bytes> {
+ let partitions: usize = answers
+ .iter()
+ .flat_map(|answer| &answer.topics)
+ .map(|topic| topic.name.0.len() + topic.partitions.len() * PER_PARTITION_BYTES)
+ .sum();
+ let groups: usize = queries.iter().map(|query| query.group_id.len() + 8).sum();
+ let capacity = 16 + groups + partitions;
+
+ if version >= FIRST_BATCHED_VERSION {
+ let groups = queries
+ .iter()
+ .zip(answers)
+ .map(|(query, answer)| {
+ OffsetFetchResponseGroup::default()
+ .with_group_id(GroupId(query.group_id.clone()))
+ .with_error_code(answer.error)
+ .with_topics(
+ answer
+ .topics
+ .iter()
+ .map(|topic| {
+ OffsetFetchResponseTopics::default()
+ .with_name(topic.name.clone())
+ .with_partitions(
+ topic
+ .partitions
+ .iter()
+ .map(|partition| {
+ OffsetFetchResponsePartitions::default()
+ .with_partition_index(partition.index)
+ .with_committed_offset(partition.offset)
+ .with_error_code(partition.error)
+ })
+ .collect(),
+ )
+ })
+ .collect(),
+ )
+ })
+ .collect();
+ return encode_message(
+ &OffsetFetchResponse::default().with_groups(groups),
+ version,
+ capacity,
+ );
+ }
+
+ let (Some(query), Some(answer)) = (queries.first(), answers.first()) else {
+ return encode_message(&OffsetFetchResponse::default(), version, capacity);
+ };
+ let topics = if version < FIRST_GROUP_ERROR_VERSION && answer.error != ERROR_NONE {
+ // No group error field yet: the error goes on every partition asked for.
+ query
+ .topics
+ .iter()
+ .flatten()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: answer.error,
+ })
+ .collect(),
+ })
+ .collect()
+ } else {
+ answer.topics.clone()
Review Comment:
encode_response takes the answers by value now, no clone.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on code in PR #4415:
URL: https://github.com/apache/iggy/pull/4415#discussion_r4211083008
##########
gateways/kafka/README.md:
##########
@@ -1,11 +1,13 @@
# Kafka gateway (`iggy-gateway-kafka`)
-Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata and CreateTopics use Iggy. InitProducerId and consumer group coordination work with or without one.
-
-> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata and
-> CreateTopics use Iggy. With the bridge off (the default), those five are stubs: Produce, Fetch and
-> ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6), CreateTopics answers `NOT_CONTROLLER`
-> (41), and Metadata reports every requested topic unknown. **CreateTopics runs as the bridge's own
+Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, OffsetCommit and OffsetFetch use Iggy. InitProducerId and consumer group coordination work with or without one.
+
+> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata,
+> CreateTopics, OffsetCommit and OffsetFetch use Iggy. With the bridge off (the default), those
+> seven are stubs: Produce, Fetch and ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6),
+> CreateTopics answers `NOT_CONTROLLER` (41), OffsetCommit and OffsetFetch answer retriable
+> `COORDINATOR_LOAD_IN_PROGRESS` (14), and Metadata reports every requested topic unknown.
+> **CreateTopics runs as the bridge's own
Review Comment:
added it next to CreateTopics, with a link to the cleanup commands.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -49,6 +49,13 @@ use crate::protocol::api::{
/// group id this gateway admits must leave room for that prefix (`docs/OFFSET_STORAGE.md`).
pub const MAX_GROUP_ID_BYTES: usize = 246;
+/// Whether every group API admits `group_id`. Kafka refuses an empty one, and a longer one than
Review Comment:
went with the doc fix and kept refusing "" for 8 and 9. it's only there for old clients. also noted it in OFFSET_STORAGE.md.
##########
gateways/kafka/docs/CONSUMER_GROUPS.md:
##########
@@ -168,17 +168,15 @@ lock, and stock clients list one or two assignors.
## What a real consumer still cannot do
Review Comment:
renamed.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -724,12 +725,32 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
stored: Some(stored_offset),
current_offset,
}) => build_consumer_offset_body(partition_id, current_offset, stored_offset),
+ Some(PartitionReadReply::ConsumerOffset { stored: None, .. }) => Bytes::new(),
Some(PartitionReadReply::Rejected(error)) => {
send_non_replicated_deny(shard, request, transport_client_id, error.as_code())
.await;
return;
}
- _ => Bytes::new(),
+ // No reply, or an owner that does not hold the partition yet,
+ // says nothing about the offset, so it must not read as "no
+ // stored offset". A read moves no progress, so the client may
+ // retry it.
+ reply => {
Review Comment:
I'd rather keep it in this PR since the gateway needs it. the body has a server change section for it.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
+ usize::from(hasher.finish().to_le_bytes()[0]) % OFFSET_SLOTS
+}
+
+/// Runs `call` in a task that keeps `turn` until `call` ends, even after its caller stops
+/// waiting, so the next call on the slot cannot overtake it.
+fn spawn_holding<T: Send + 'static>(
Review Comment:
done. commits go through in_slot with timeout(SLOT_LIMIT, ..) like produce, spawn_holding is gone. in_slot's doc has the line about commits keeping their client.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
Review Comment:
moved it to LazyClient::close_all, both pools call it.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -1112,6 +1119,83 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// Kafka's `ClassicGroup.validateOffsetCommit`. Never creates a group.
+///
+/// A group with no members is Kafka's `Empty`: a negative generation commits without a member. A
+/// gateway restart empties every group, so a member's next commit gets `UNKNOWN_MEMBER_ID` and it
+/// rejoins. Kafka's v9 answer, `GROUP_ID_NOT_FOUND`, is fatal in the Java consumer.
+pub fn commit_step(groups: &mut Groups, request: &CommitRequest, now: Instant) -> i16 {
+ if !is_valid_group_id(&request.group_id) {
+ return ERROR_INVALID_GROUP_ID;
+ }
+ let empty = || {
+ if request.generation_id < 0 {
+ ERROR_NONE
+ } else {
+ ERROR_UNKNOWN_MEMBER_ID
+ }
+ };
+ if !tick_group(groups, &request.group_id, now) {
+ return empty();
+ }
+ let Some(group) = groups.get_mut(&request.group_id) else {
+ return empty();
+ };
+ if group.members.is_empty() {
+ return empty();
+ }
+ if request.generation_id < 0
+ && request.member_id.is_empty()
+ && request.group_instance_id.is_none()
+ {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ if let Some(instance_id) = &request.group_instance_id {
+ let holds = |member: &Member| member.group_instance_id.as_ref() == Some(instance_id);
+ if !group.members.values().any(holds) {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ if !group.members.get(&request.member_id).is_some_and(holds) {
+ return ERROR_FENCED_INSTANCE_ID;
+ }
+ }
+ let (generation_id, phase) = (group.generation_id, group.phase);
+ let Some(member) = group.members.get_mut(&request.member_id) else {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ };
+ if request.generation_id != generation_id {
+ return ERROR_ILLEGAL_GENERATION;
+ }
+ // A member commits right before it rejoins, so `PreparingRebalance` accepts.
+ if phase == Phase::CompletingRebalance {
+ return ERROR_REBALANCE_IN_PROGRESS;
+ }
+ member.session_deadline = now + member.session_timeout;
+ ERROR_NONE
+}
+
+/// How long an offset call may hold the connection that `member` heartbeats on: `hold`, and at
+/// most half of what is left of that member's session, so a heartbeat that waits behind the call
+/// still lands in time. A member that is not there, or whose session has ended, leaves `hold`.
+pub fn offset_hold(
Review Comment:
takes &GroupMember now. dropped the None branch and its assert.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "hubcio (via GitHub)" <gi...@apache.org>.
hubcio commented on code in PR #4415:
URL: https://github.com/apache/iggy/pull/4415#discussion_r4209806933
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
Review Comment:
warning: since commits hold the slot for ordering, one stuck commit blocks every group's offset reads on that slot for up to `SLOT_LIMIT`, and they all get 14. give reads their own client per slot so only commits wait on commits.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,553 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::Instant;
+
+use crate::bridge::{BridgeError, IggyBridge, KafkaTopicMetadata, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_deadline, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = offset_deadline(state, connection).await;
+ // The groups read at once, so a slow group cannot spend the time of the others.
+ let answers = join_all(
+ queries
+ .iter()
+ .map(|query| answer(state.bridge.as_deref(), query, deadline)),
+ )
+ .await;
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ // Each Iggy call stops at `deadline` on its own, so none is cut off holding its slot.
+ let read = match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ };
+ match read {
+ Ok(topics) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Err(error) => GroupAnswer::error(refusal(&query.group_id, &error)),
+ }
+}
+
+/// The group code for a failed read. A -1 is logged, since the client sees no more.
+fn refusal(group: &str, error: &BridgeError) -> i16 {
+ let code = error.to_offset_error_code();
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ tracing::error!(
+ group,
+ %error,
+ cause = offset_refusal_cause(error),
+ "Iggy refused a Kafka offset read"
+ );
+ } else {
+ tracing::debug!(group, %error, "Iggy refused a Kafka offset read");
+ }
+ code
+}
+
+/// Every requested partition, in request order. A partition with no offset answers -1. The reads
+/// run at once across the offset slots.
+async fn read_topics(
+ bridge: &IggyBridge,
+ group: &str,
+ topics: &[(TopicName, Vec<i32>)],
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ // A name that cannot map to Iggy names no topic, so nothing is committed under it.
+ let targets: Vec<Option<TopicTarget>> = topics
+ .iter()
+ .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok())
+ .collect();
+ let mut answers: Vec<TopicAnswer> = topics
+ .iter()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: ERROR_NONE,
+ })
+ .collect(),
+ })
+ .collect();
+ let mut reads = Vec::new();
+ // Where each read's offset goes: (topic, partition) in `answers`.
+ let mut positions = Vec::new();
+ for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() {
+ let Some(target) = target else {
+ continue;
+ };
+ for (partition_at, &index) in indexes.iter().enumerate() {
+ if let Ok(partition) = u32::try_from(index) {
+ reads.push((target, partition));
+ positions.push((topic_at, partition_at));
+ }
+ }
+ }
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) {
+ answers[topic_at].partitions[partition_at].offset = committed(read)?;
+ }
+ Ok(answers)
+}
+
+/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only
+/// the partitions with an offset, on every topic that holds the group. The topics read at once.
+async fn read_every_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let topics = bridge.list_kafka_topics_until(deadline).await?;
+ let answers = join_all(
Review Comment:
warning: a null-topic fetch queues one call per topic and partition on the offset slots, so one admin `listConsumerGroupOffsets` can stall all other offset calls for up to 3 s. cap each request at one queued call per slot.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
Review Comment:
nit: hashing the partition index puts two partitions of most 4-partition topics on one slot, so their commits run one after another. hash only stream and topic, then add the partition index.
##########
gateways/kafka/README.md:
##########
@@ -1,11 +1,13 @@
# Kafka gateway (`iggy-gateway-kafka`)
-Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata and CreateTopics use Iggy. InitProducerId and consumer group coordination work with or without one.
-
-> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata and
-> CreateTopics use Iggy. With the bridge off (the default), those five are stubs: Produce, Fetch and
-> ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6), CreateTopics answers `NOT_CONTROLLER`
-> (41), and Metadata reports every requested topic unknown. **CreateTopics runs as the bridge's own
+Foundation layer for [apache/iggy#3421](https://github.com/apache/iggy/issues/3421): a TCP listener on the Kafka wire port that decodes requests and validates scoped API keys and versions. With a bridge, Produce, Fetch, ListOffsets, Metadata, CreateTopics, OffsetCommit and OffsetFetch use Iggy. InitProducerId and consumer group coordination work with or without one.
+
+> **Stub warning:** with `IGGY_KAFKA_BRIDGE_ENABLED=true`, Produce, Fetch, ListOffsets, Metadata,
+> CreateTopics, OffsetCommit and OffsetFetch use Iggy. With the bridge off (the default), those
+> seven are stubs: Produce, Fetch and ListOffsets answer retriable `NOT_LEADER_OR_FOLLOWER` (6),
+> CreateTopics answers `NOT_CONTROLLER` (41), OffsetCommit and OffsetFetch answer retriable
+> `COORDINATOR_LOAD_IN_PROGRESS` (14), and Metadata reports every requested topic unknown.
+> **CreateTopics runs as the bridge's own
Review Comment:
nit: `OffsetCommit` creates iggy resources too - a `kafka.cg.<group>` consumer group per group id and topic, and nothing deletes them. name it next to `CreateTopics` and point at the cleanup commands in `OFFSET_STORAGE.md`.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
Review Comment:
nit: this repeats the permit and deadline guard from `send_in_slot` in `produce.rs`. a small `permit_by(turn, deadline)` helper keeps one copy of the deadline rule.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -724,12 +725,32 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
stored: Some(stored_offset),
current_offset,
}) => build_consumer_offset_body(partition_id, current_offset, stored_offset),
+ Some(PartitionReadReply::ConsumerOffset { stored: None, .. }) => Bytes::new(),
Some(PartitionReadReply::Rejected(error)) => {
send_non_replicated_deny(shard, request, transport_client_id, error.as_code())
.await;
return;
}
- _ => Bytes::new(),
+ // No reply, or an owner that does not hold the partition yet,
+ // says nothing about the offset, so it must not read as "no
+ // stored offset". A read moves no progress, so the client may
+ // retry it.
+ reply => {
Review Comment:
nit: the body covers this server contract change, but the squash title will only say `feat(gateways)`. split it into a `fix(server)` PR or name it in the title.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -1112,6 +1119,83 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// Kafka's `ClassicGroup.validateOffsetCommit`. Never creates a group.
+///
+/// A group with no members is Kafka's `Empty`: a negative generation commits without a member. A
+/// gateway restart empties every group, so a member's next commit gets `UNKNOWN_MEMBER_ID` and it
+/// rejoins. Kafka's v9 answer, `GROUP_ID_NOT_FOUND`, is fatal in the Java consumer.
+pub fn commit_step(groups: &mut Groups, request: &CommitRequest, now: Instant) -> i16 {
+ if !is_valid_group_id(&request.group_id) {
+ return ERROR_INVALID_GROUP_ID;
+ }
+ let empty = || {
+ if request.generation_id < 0 {
+ ERROR_NONE
+ } else {
+ ERROR_UNKNOWN_MEMBER_ID
+ }
+ };
+ if !tick_group(groups, &request.group_id, now) {
+ return empty();
+ }
+ let Some(group) = groups.get_mut(&request.group_id) else {
+ return empty();
+ };
+ if group.members.is_empty() {
+ return empty();
+ }
+ if request.generation_id < 0
+ && request.member_id.is_empty()
+ && request.group_instance_id.is_none()
+ {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ if let Some(instance_id) = &request.group_instance_id {
+ let holds = |member: &Member| member.group_instance_id.as_ref() == Some(instance_id);
+ if !group.members.values().any(holds) {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ }
+ if !group.members.get(&request.member_id).is_some_and(holds) {
+ return ERROR_FENCED_INSTANCE_ID;
+ }
+ }
+ let (generation_id, phase) = (group.generation_id, group.phase);
+ let Some(member) = group.members.get_mut(&request.member_id) else {
+ return ERROR_UNKNOWN_MEMBER_ID;
+ };
+ if request.generation_id != generation_id {
+ return ERROR_ILLEGAL_GENERATION;
+ }
+ // A member commits right before it rejoins, so `PreparingRebalance` accepts.
+ if phase == Phase::CompletingRebalance {
+ return ERROR_REBALANCE_IN_PROGRESS;
+ }
+ member.session_deadline = now + member.session_timeout;
+ ERROR_NONE
+}
+
+/// How long an offset call may hold the connection that `member` heartbeats on: `hold`, and at
+/// most half of what is left of that member's session, so a heartbeat that waits behind the call
+/// still lands in time. A member that is not there, or whose session has ended, leaves `hold`.
+pub fn offset_hold(
Review Comment:
simplification: the only production caller already returns on `None` before it calls this, so take `&GroupMember` and drop the `None` branch and its test assertion.
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,553 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use futures::future::join_all;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::Instant;
+
+use crate::bridge::{BridgeError, IggyBridge, KafkaTopicMetadata, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ConnectionState, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_deadline, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(
+ state: &GatewayState,
+ connection: &ConnectionState,
+ api_version: i16,
+ body: Bytes,
+) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = offset_deadline(state, connection).await;
+ // The groups read at once, so a slow group cannot spend the time of the others.
+ let answers = join_all(
+ queries
+ .iter()
+ .map(|query| answer(state.bridge.as_deref(), query, deadline)),
+ )
+ .await;
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ // Each Iggy call stops at `deadline` on its own, so none is cut off holding its slot.
+ let read = match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ };
+ match read {
+ Ok(topics) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Err(error) => GroupAnswer::error(refusal(&query.group_id, &error)),
+ }
+}
+
+/// The group code for a failed read. A -1 is logged, since the client sees no more.
+fn refusal(group: &str, error: &BridgeError) -> i16 {
+ let code = error.to_offset_error_code();
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ tracing::error!(
+ group,
+ %error,
+ cause = offset_refusal_cause(error),
+ "Iggy refused a Kafka offset read"
+ );
+ } else {
+ tracing::debug!(group, %error, "Iggy refused a Kafka offset read");
+ }
+ code
+}
+
+/// Every requested partition, in request order. A partition with no offset answers -1. The reads
+/// run at once across the offset slots.
+async fn read_topics(
+ bridge: &IggyBridge,
+ group: &str,
+ topics: &[(TopicName, Vec<i32>)],
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ // A name that cannot map to Iggy names no topic, so nothing is committed under it.
+ let targets: Vec<Option<TopicTarget>> = topics
+ .iter()
+ .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok())
+ .collect();
+ let mut answers: Vec<TopicAnswer> = topics
+ .iter()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: ERROR_NONE,
+ })
+ .collect(),
+ })
+ .collect();
+ let mut reads = Vec::new();
+ // Where each read's offset goes: (topic, partition) in `answers`.
+ let mut positions = Vec::new();
+ for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() {
+ let Some(target) = target else {
+ continue;
+ };
+ for (partition_at, &index) in indexes.iter().enumerate() {
+ if let Ok(partition) = u32::try_from(index) {
+ reads.push((target, partition));
+ positions.push((topic_at, partition_at));
+ }
+ }
+ }
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) {
+ answers[topic_at].partitions[partition_at].offset = committed(read)?;
+ }
+ Ok(answers)
+}
+
+/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only
+/// the partitions with an offset, on every topic that holds the group. The topics read at once.
+async fn read_every_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let topics = bridge.list_kafka_topics_until(deadline).await?;
+ let answers = join_all(
+ topics
+ .into_iter()
+ .map(|topic| read_listed_topic(bridge, group, topic, deadline)),
+ )
+ .await;
+ answers
+ .into_iter()
+ .filter_map(core::result::Result::transpose)
+ .collect()
+}
+
+/// The partitions of `topic` with an offset of `group`. `None` when there are none.
+async fn read_listed_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ topic: KafkaTopicMetadata,
+ deadline: Instant,
+) -> core::result::Result<Option<TopicAnswer>, BridgeError> {
+ let target = bridge.topic_target(&topic.kafka_topic)?;
+ match bridge.holds_offset_group(group, &target, deadline).await {
+ Ok(true) => {}
+ Ok(false) => return Ok(None),
+ // A topic deleted since the listing holds no offsets.
+ Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => {
+ return Ok(None);
+ }
+ Err(error) => return Err(error),
+ }
+ let reads: Vec<_> = (0..topic.partitions_count)
+ .map(|partition| (&target, partition))
+ .collect();
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ let mut partitions = Vec::new();
+ for (partition, read) in (0..topic.partitions_count).zip(offsets) {
+ let offset = committed(read)?;
+ if offset != UNKNOWN_OFFSET {
+ partitions.push(PartitionAnswer {
+ index: i32::try_from(partition).unwrap_or(i32::MAX),
+ offset,
+ error: ERROR_NONE,
+ });
+ }
+ }
+ Ok((!partitions.is_empty()).then(|| TopicAnswer {
+ name: TopicName(StrBytes::from_string(topic.kafka_topic)),
+ partitions,
+ }))
+}
+
+/// A read as Kafka answers it: -1 for no offset and for a topic or partition that does not exist.
+fn committed(
+ read: core::result::Result<Option<u64>, BridgeError>,
+) -> core::result::Result<i64, BridgeError> {
+ match read {
+ Ok(Some(offset)) => Ok(i64::try_from(offset).unwrap_or(i64::MAX)),
+ Ok(None) => Ok(UNKNOWN_OFFSET),
+ Err(error) if error.to_kafka_error_code() == ERROR_UNKNOWN_TOPIC_OR_PARTITION => {
+ Ok(UNKNOWN_OFFSET)
+ }
+ Err(error) => Err(error),
+ }
+}
+
+/// One entry per query, in request order. Below v8 there is exactly one.
+///
+/// # Errors
+///
+/// Returns an error when `kafka_protocol` cannot encode the response at `version`.
+fn encode_response(version: i16, queries: &[GroupQuery], answers: &[GroupAnswer]) -> Result<Bytes> {
+ let partitions: usize = answers
+ .iter()
+ .flat_map(|answer| &answer.topics)
+ .map(|topic| topic.name.0.len() + topic.partitions.len() * PER_PARTITION_BYTES)
+ .sum();
+ let groups: usize = queries.iter().map(|query| query.group_id.len() + 8).sum();
+ let capacity = 16 + groups + partitions;
+
+ if version >= FIRST_BATCHED_VERSION {
+ let groups = queries
+ .iter()
+ .zip(answers)
+ .map(|(query, answer)| {
+ OffsetFetchResponseGroup::default()
+ .with_group_id(GroupId(query.group_id.clone()))
+ .with_error_code(answer.error)
+ .with_topics(
+ answer
+ .topics
+ .iter()
+ .map(|topic| {
+ OffsetFetchResponseTopics::default()
+ .with_name(topic.name.clone())
+ .with_partitions(
+ topic
+ .partitions
+ .iter()
+ .map(|partition| {
+ OffsetFetchResponsePartitions::default()
+ .with_partition_index(partition.index)
+ .with_committed_offset(partition.offset)
+ .with_error_code(partition.error)
+ })
+ .collect(),
+ )
+ })
+ .collect(),
+ )
+ })
+ .collect();
+ return encode_message(
+ &OffsetFetchResponse::default().with_groups(groups),
+ version,
+ capacity,
+ );
+ }
+
+ let (Some(query), Some(answer)) = (queries.first(), answers.first()) else {
+ return encode_message(&OffsetFetchResponse::default(), version, capacity);
+ };
+ let topics = if version < FIRST_GROUP_ERROR_VERSION && answer.error != ERROR_NONE {
+ // No group error field yet: the error goes on every partition asked for.
+ query
+ .topics
+ .iter()
+ .flatten()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: answer.error,
+ })
+ .collect(),
+ })
+ .collect()
+ } else {
+ answer.topics.clone()
Review Comment:
nit: below v8 this clones every topic answer just to read it. take the answers by value in `encode_response` and move `topics` out - no clone needed.
##########
gateways/kafka/src/group/state.rs:
##########
@@ -49,6 +49,13 @@ use crate::protocol::api::{
/// group id this gateway admits must leave room for that prefix (`docs/OFFSET_STORAGE.md`).
pub const MAX_GROUP_ID_BYTES: usize = 246;
+/// Whether every group API admits `group_id`. Kafka refuses an empty one, and a longer one than
Review Comment:
nit: kafka refuses an empty group id only for the membership APIs - `commitOffsets` and `fetchOffsets` in `GroupCoordinatorService` accept "" for backwards compatibility. fix the doc, or accept "" for keys 8 and 9.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
+ usize::from(hasher.finish().to_le_bytes()[0]) % OFFSET_SLOTS
+}
+
+/// Runs `call` in a task that keeps `turn` until `call` ends, even after its caller stops
+/// waiting, so the next call on the slot cannot overtake it.
+fn spawn_holding<T: Send + 'static>(
+ turn: OwnedSemaphorePermit,
+ call: impl Future<Output = Result<T, BridgeError>> + Send + 'static,
+) -> oneshot::Receiver<Result<T, BridgeError>> {
+ let (sender, receiver) = oneshot::channel();
+ tokio::spawn(async move {
+ let ended = timeout(SLOT_LIMIT, call)
Review Comment:
warning: the permit isn't held until the call ends - this frees it at `SLOT_LIMIT` while the SDK attempt still runs, and a timed-out read drops the client, so the next commit opens a new session. reword the order promise in `OFFSET_STORAGE.md`.
##########
gateways/kafka/docs/CONSUMER_GROUPS.md:
##########
@@ -168,17 +168,15 @@ lock, and stock clients list one or two assignors.
## What a real consumer still cannot do
Review Comment:
nit: the bridge paragraph below lists things a consumer can do now, so this heading no longer fits. rename it to "what a real consumer can and cannot do".
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
Review Comment:
simplification: `OffsetPool::close` is a line-for-line copy of `FetchPool::close`. one shared helper over the `LazyClient` list covers both.
##########
core/server/src/dispatch/partition.rs:
##########
@@ -724,12 +725,32 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
stored: Some(stored_offset),
current_offset,
}) => build_consumer_offset_body(partition_id, current_offset, stored_offset),
+ Some(PartitionReadReply::ConsumerOffset { stored: None, .. }) => Bytes::new(),
Some(PartitionReadReply::Rejected(error)) => {
send_non_replicated_deny(shard, request, transport_client_id, error.as_code())
.await;
return;
}
- _ => Bytes::new(),
+ // No reply, or an owner that does not hold the partition yet,
+ // says nothing about the offset, so it must not read as "no
+ // stored offset". A read moves no progress, so the client may
+ // retry it.
+ reply => {
+ warn!(
Review Comment:
nit: the SDK retries `TransientNotAccepted` every 50 ms, so a read for a partition the owner shard doesn't hold yet logs this `warn!` on every retry. log `NotFound` at `debug!` and keep `warn!` for the timeout.
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,574 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use futures::future::join_all;
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, KafkaTopicMetadata, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+/// The slot of a call that belongs to no partition. Any slot will do.
+const LISTING_SLOT: usize = 0;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it. A replay that Iggy already
+ /// applied counts as done.
+ ///
+ /// Each commit starts once its slot is free. Commits on one slot start in the order given.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ // `join_all` polls each call once, in order, before any again, so calls on one slot
+ // queue for it in order.
+ join_all(commits.iter().map(|&(topic, partition, offset)| {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ self.commit_in_slot(slot_of(topic, partition), deadline, commit)
+ }))
+ .await
+ }
+
+ /// The committed offset of Kafka group `group` for each `(topic, partition)` of `reads`, in
+ /// order. `None` when there is no key or no group. The reads run at once.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::commit_group_offsets`]. A missing topic is [`BridgeError::Iggy`] too.
+ pub async fn fetch_group_offsets(
+ &self,
+ group: &str,
+ reads: &[(&TopicTarget, u32)],
+ deadline: Instant,
+ ) -> Vec<Result<Option<u64>, BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => names,
+ Err(error) => return refused_all(reads.len(), &error),
+ };
+ let consumer = &names.consumer;
+ join_all(reads.iter().map(|&(topic, partition)| {
+ let read = move |client: Arc<IggyClient>| async move {
+ let info = client
+ .get_consumer_offset(
+ consumer,
+ &topic.stream_id,
+ &topic.topic_id,
+ Some(partition),
+ )
+ .await
+ .map_err(BridgeError::Iggy)?;
+ Ok(info.map(|info| info.stored_offset))
+ };
+ self.read_in_slot(slot_of(topic, partition), deadline, read)
+ }))
+ .await
+ }
+
+ /// Whether Kafka group `group` holds offsets on `topic`.
+ ///
+ /// # Errors
+ ///
+ /// [`BridgeError::Iggy`] if Iggy refuses the call, for example for a missing topic, or `group`
+ /// names no valid Iggy group. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn holds_offset_group(
+ &self,
+ group: &str,
+ topic: &TopicTarget,
+ deadline: Instant,
+ ) -> Result<bool, BridgeError> {
+ let group_id = OffsetNames::new(group)
+ .map_err(BridgeError::Iggy)?
+ .consumer
+ .id;
+ let lookup = |client: Arc<IggyClient>| async move {
+ client
+ .get_consumer_group(&topic.stream_id, &topic.topic_id, &group_id)
+ .await
+ .map(|found| found.is_some())
+ .map_err(BridgeError::Iggy)
+ };
+ // Any slot will do. Partition 0's keeps the choice fixed.
+ self.read_in_slot(slot_of(topic, 0), deadline, lookup).await
+ }
+
+ /// [`Self::list_kafka_topics`] on an offset slot's client, so it never waits behind Produce.
+ ///
+ /// # Errors
+ ///
+ /// As [`Self::list_kafka_topics`]. [`BridgeError::Timeout`] past `deadline`.
+ pub async fn list_kafka_topics_until(
+ &self,
+ deadline: Instant,
+ ) -> Result<Vec<KafkaTopicMetadata>, BridgeError> {
+ let listing =
+ |client: Arc<IggyClient>| async move { self.list_kafka_topics_on(&client).await };
+ self.read_in_slot(LISTING_SLOT, deadline, listing).await
+ }
+
+ /// Runs `call` with the client of `slot`, and waits for it until `deadline`. The call keeps
+ /// the slot until it ends, even after its caller stops waiting, so the next call on the slot
+ /// cannot overtake it. A call that has not started by `deadline` never starts.
+ async fn commit_in_slot<F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<(), BridgeError>
+ where
+ F: Future<Output = Result<(), BridgeError>> + Send + 'static,
+ {
+ let Some((turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let client = match timeout_at(deadline, client.connected(&self.config)).await {
+ Ok(connected) => connected?,
+ Err(_elapsed) => return Err(BridgeError::Timeout),
+ };
+ match timeout_at(deadline, spawn_holding(turn, call(client))).await {
+ Ok(Ok(ended)) => ended,
+ // A task that ends with no answer leaves the outcome unknown too.
+ Ok(Err(_)) | Err(_) => Err(BridgeError::Timeout),
+ }
+ }
+
+ /// Runs `call` with the client of `slot`, and frees the slot when it ends or at `deadline`.
+ /// A read that `deadline` cuts off leaves its client behind, so the next call on the slot
+ /// does not wait behind it in the SDK. A call that has not started by `deadline` never starts.
+ async fn read_in_slot<T, F>(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ call: impl FnOnce(Arc<IggyClient>) -> F,
+ ) -> Result<T, BridgeError>
+ where
+ F: Future<Output = Result<T, BridgeError>>,
+ {
+ let Some((_turn, client)) = self.offset_pool.take(slot, deadline).await else {
+ return Err(BridgeError::Timeout);
+ };
+ let read = async { call(client.connected(&self.config).await?).await };
+ client.until(deadline, read).await
+ }
+}
+
+/// The offset slots. A partition always takes the same slot. A commit keeps its slot until the
+/// SDK lets go of it, so the commits of one partition reach Iggy in the order they started, and a
+/// read that comes after them waits for them. A read frees its slot by its deadline.
+pub(super) struct OffsetPool {
+ slots: Vec<OffsetSlot>,
+}
+
+struct OffsetSlot {
+ client: LazyClient,
+ /// One call at a time.
+ turn: Arc<Semaphore>,
+}
+
+impl OffsetPool {
+ pub(super) fn new() -> Self {
+ Self {
+ slots: (0..OFFSET_SLOTS)
+ .map(|_| OffsetSlot {
+ client: LazyClient::default(),
+ turn: Arc::new(Semaphore::new(1)),
+ })
+ .collect(),
+ }
+ }
+
+ /// Slot `slot`, with its client. `None` if it is not free by `deadline`.
+ async fn take(
+ &self,
+ slot: usize,
+ deadline: Instant,
+ ) -> Option<(OwnedSemaphorePermit, &LazyClient)> {
+ let slot = &self.slots[slot];
+ match timeout_at(deadline, Arc::clone(&slot.turn).acquire_owned()).await {
+ Ok(Ok(turn)) if Instant::now() < deadline => Some((turn, &slot.client)),
+ _ => None,
+ }
+ }
+
+ /// Shuts down each client that connected, and returns the first error. A client that fails to
+ /// shut down stops none of the others.
+ pub(super) async fn close(&self) -> Result<(), BridgeError> {
+ let mut closed = Ok(());
+ for slot in &self.slots {
+ let shutdown = slot.client.close().await;
+ closed = closed.and(shutdown);
+ }
+ closed
+ }
+}
+
+/// The slot of `partition` of `topic`, the same at every call.
+fn slot_of(topic: &TopicTarget, partition: u32) -> usize {
+ let mut hasher = DefaultHasher::new();
+ (&topic.stream_id, &topic.topic_id, partition).hash(&mut hasher);
+ usize::from(hasher.finish().to_le_bytes()[0]) % OFFSET_SLOTS
+}
+
+/// Runs `call` in a task that keeps `turn` until `call` ends, even after its caller stops
+/// waiting, so the next call on the slot cannot overtake it.
+fn spawn_holding<T: Send + 'static>(
Review Comment:
simplification: this repeats `in_slot` plus the `SLOT_LIMIT` wrapper from `send_in_slot`. pass `timeout(SLOT_LIMIT, call(client))` to `in_slot` like produce does and drop `spawn_holding` - commits keep their client, so the `in_slot` doc needs a line on that.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on code in PR #4415:
URL: https://github.com/apache/iggy/pull/4415#discussion_r4203180025
##########
core/server/src/dispatch/partition.rs:
##########
@@ -729,6 +730,24 @@ pub(in crate::dispatch) async fn handle_get_consumer_offset<B, MJ, S, SB>(
.await;
return;
}
+ // A missing reply says nothing about the offset, so it must
+ // not read as "no stored offset". A read moves no progress,
+ // so the client may retry it.
+ None => {
+ warn!(
+ transport_client_id,
+ namespace = namespace.inner(),
+ "consumer offset read reply unavailable; replying not accepted"
+ );
+ send_non_replicated_deny(
+ shard,
+ request,
+ transport_client_id,
+ IggyError::TransientNotAccepted.as_code(),
+ )
+ .await;
+ return;
+ }
_ => Bytes::new(),
Review Comment:
fixed. only `ConsumerOffset { stored: None }` gets the empty body now, NotFound / no reply / anything else answers TransientNotAccepted. added a test
##########
gateways/kafka/src/bridge/iggy_bridge/group_offsets.rs:
##########
@@ -0,0 +1,449 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Kafka group offsets, kept as Iggy external group offsets. See `docs/OFFSET_STORAGE.md`.
+
+use std::future::Future;
+use std::hash::{DefaultHasher, Hash, Hasher};
+use std::sync::Arc;
+
+use iggy::prelude::{
+ Consumer, ConsumerGroupClient, ConsumerOffsetClient, Identifier, IggyClient, IggyError,
+};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot};
+use tokio::time::{Instant, timeout, timeout_at};
+use tracing::info;
+
+use super::{IggyBridge, LazyClient, SLOT_LIMIT, TopicTarget};
+use crate::bridge::error::BridgeError;
+
+/// Prefix of the Iggy consumer group that holds a Kafka group's offsets. It keeps a Kafka group
+/// apart from a native Iggy group of the same name.
+pub const OFFSET_GROUP_PREFIX: &str = "kafka.cg.";
+
+/// Offset calls that run at once, each slot on a client of its own, so offset calls never wait
+/// behind a Produce send.
+const OFFSET_SLOTS: usize = 4;
+
+impl IggyBridge {
+ /// Stores each `(topic, partition, offset)` of `commits` as a committed offset of Kafka group
+ /// `group`. One result per commit, in order.
+ ///
+ /// A negative offset deletes the key instead. Kafka consumers read any negative offset as
+ /// none, and Iggy cannot store one. A key or a group that is not there counts as deleted. A
+ /// store that finds no Iggy group for the offsets creates it.
+ ///
+ /// # Errors
+ ///
+ /// Per commit: [`BridgeError::Iggy`] if Iggy refuses it or `group` names no valid Iggy group,
+ /// and [`BridgeError::Timeout`] if it does not end by `deadline`. A commit that has not
+ /// started by then never starts.
+ pub async fn commit_group_offsets(
+ &self,
+ group: &str,
+ commits: &[(&TopicTarget, u32, i64)],
+ deadline: Instant,
+ ) -> Vec<Result<(), BridgeError>> {
+ let names = match OffsetNames::new(group) {
+ Ok(names) => Arc::new(names),
+ Err(error) => return refused_all(commits.len(), &error),
+ };
+ let mut started = Vec::with_capacity(commits.len());
+ for &(topic, partition, offset) in commits {
+ let (names, stream_id, topic_id) = (
+ Arc::clone(&names),
+ topic.stream_id.clone(),
+ topic.topic_id.clone(),
+ );
+ let commit = move |client: Arc<IggyClient>| async move {
+ commit_on(&client, &names, &stream_id, &topic_id, partition, offset).await
+ };
+ started.push(
+ self.start_offset_call(topic, partition, deadline, commit)
+ .await,
Review Comment:
done, commits all start together through join_all. it polls them in order first, so same-slot calls keep request order
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
Review Comment:
fixed, groups read together under one deadline. added a real server test with 3 groups in one v8 request
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
+ answers.push(answer(state.bridge.as_deref(), query, deadline).await);
+ }
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ let read = async {
+ match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ }
+ };
+ match timeout_at(deadline, read).await {
+ Ok(Ok(topics)) => GroupAnswer {
+ error: ERROR_NONE,
+ topics,
+ },
+ Ok(Err(error)) => GroupAnswer::error(refusal(&query.group_id, &error)),
+ Err(_elapsed) => GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS),
+ }
+}
+
+/// The group code for a failed read. A -1 is logged, since the client sees no more.
+fn refusal(group: &str, error: &BridgeError) -> i16 {
+ let code = error.to_offset_error_code();
+ if code == ERROR_UNKNOWN_SERVER_ERROR {
+ tracing::error!(
+ group,
+ %error,
+ cause = offset_refusal_cause(error),
+ "Iggy refused a Kafka offset read"
+ );
+ } else {
+ tracing::debug!(group, %error, "Iggy refused a Kafka offset read");
+ }
+ code
+}
+
+/// Every requested partition, in request order. A partition with no offset answers -1. The reads
+/// run at once across the offset slots.
+async fn read_topics(
+ bridge: &IggyBridge,
+ group: &str,
+ topics: &[(TopicName, Vec<i32>)],
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ // A name that cannot map to Iggy names no topic, so nothing is committed under it.
+ let targets: Vec<Option<TopicTarget>> = topics
+ .iter()
+ .map(|(name, _)| bridge.topic_target(name.0.as_str()).ok())
+ .collect();
+ let mut answers: Vec<TopicAnswer> = topics
+ .iter()
+ .map(|(name, indexes)| TopicAnswer {
+ name: name.clone(),
+ partitions: indexes
+ .iter()
+ .map(|&index| PartitionAnswer {
+ index,
+ offset: UNKNOWN_OFFSET,
+ error: ERROR_NONE,
+ })
+ .collect(),
+ })
+ .collect();
+ let mut reads = Vec::new();
+ // Where each read's offset goes: (topic, partition) in `answers`.
+ let mut positions = Vec::new();
+ for (topic_at, ((_, indexes), target)) in topics.iter().zip(&targets).enumerate() {
+ let Some(target) = target else {
+ continue;
+ };
+ for (partition_at, &index) in indexes.iter().enumerate() {
+ if let Ok(partition) = u32::try_from(index) {
+ reads.push((target, partition));
+ positions.push((topic_at, partition_at));
+ }
+ }
+ }
+ let offsets = bridge.fetch_group_offsets(group, &reads, deadline).await;
+ for ((topic_at, partition_at), read) in positions.into_iter().zip(offsets) {
+ answers[topic_at].partitions[partition_at].offset = committed(read)?;
+ }
+ Ok(answers)
+}
+
+/// A null topic list, as the Java `Admin.listConsumerGroupOffsets` sends for a whole group: only
+/// the partitions with an offset, on every topic that holds the group.
+async fn read_every_topic(
+ bridge: &IggyBridge,
+ group: &str,
+ deadline: Instant,
+) -> core::result::Result<Vec<TopicAnswer>, BridgeError> {
+ let mut answers = Vec::new();
+ for topic in bridge.list_kafka_topics().await? {
Review Comment:
moved the listing to an offset slot client, with the request deadline
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
+ let mut answers = Vec::with_capacity(queries.len());
+ for query in &queries {
+ answers.push(answer(state.bridge.as_deref(), query, deadline).await);
+ }
+ respond_or_close(
+ encode_response(api_version, &queries, &answers),
+ "OffsetFetch",
+ )
+}
+
+/// One failed read fails the whole group, so a client never reads a lost answer as "nothing
+/// committed" and resets its position.
+async fn answer(bridge: Option<&IggyBridge>, query: &GroupQuery, deadline: Instant) -> GroupAnswer {
+ if !is_valid_group_id(&query.group_id) {
+ return GroupAnswer::error(ERROR_INVALID_GROUP_ID);
+ }
+ let Some(bridge) = bridge else {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ };
+ // A group the deadline leaves out costs no Iggy call.
+ if Instant::now() >= deadline {
+ return GroupAnswer::error(ERROR_COORDINATOR_LOAD_IN_PROGRESS);
+ }
+ let read = async {
+ match &query.topics {
+ Some(topics) => read_topics(bridge, &query.group_id, topics, deadline).await,
+ None => read_every_topic(bridge, &query.group_id, deadline).await,
+ }
+ };
+ match timeout_at(deadline, read).await {
Review Comment:
dropped the outer timeout_at, every call has its own deadline now. a read that hits it frees its slot and drops its client, the retry gets a fresh one. commits still hold the slot till the SDK is done, that's what keeps them ordered. no partial topics though: one failed read still fails the group, since a missing partition reads as nothing committed
##########
gateways/kafka/src/protocol/handlers/offset_fetch.rs:
##########
@@ -0,0 +1,526 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! `OffsetFetch` (API key 9).
+//!
+//! Reads committed offsets from Iggy (`docs/OFFSET_STORAGE.md`). A failure goes at group level
+//! from v2: the Java consumer treats most partition errors as fatal, even retriable ones.
+
+use bytes::Bytes;
+use kafka_protocol::messages::offset_fetch_response::{
+ OffsetFetchResponseGroup, OffsetFetchResponsePartition, OffsetFetchResponsePartitions,
+ OffsetFetchResponseTopic, OffsetFetchResponseTopics,
+};
+use kafka_protocol::messages::{GroupId, OffsetFetchRequest, OffsetFetchResponse, TopicName};
+use kafka_protocol::protocol::StrBytes;
+use tokio::time::{Instant, timeout_at};
+
+use crate::bridge::{BridgeError, IggyBridge, TopicTarget};
+use crate::error::Result;
+use crate::group::is_valid_group_id;
+use crate::protocol::api::{
+ API_KEY_OFFSET_FETCH, ApiVersionRange, ERROR_COORDINATOR_LOAD_IN_PROGRESS,
+ ERROR_INVALID_GROUP_ID, ERROR_NONE, ERROR_UNKNOWN_SERVER_ERROR,
+ ERROR_UNKNOWN_TOPIC_OR_PARTITION, GatewayState, HandleOutcome, UNKNOWN_OFFSET,
+ is_supported_version,
+};
+use crate::protocol::bounds_guard::validate_offset_fetch_shape;
+use crate::protocol::handlers::{
+ decode_guarded, encode_message, offset_refusal_cause, respond_or_close,
+};
+
+pub const RANGE: ApiVersionRange = ApiVersionRange {
+ api_key: API_KEY_OFFSET_FETCH,
+ min_version: 1,
+ max_version: 9,
+};
+
+/// From this version a request names several groups and each answer carries its own error.
+const FIRST_BATCHED_VERSION: i16 = 8;
+/// Below this version there is no group error field.
+const FIRST_GROUP_ERROR_VERSION: i16 = 2;
+/// Index, offset, leader epoch, empty metadata, error code and tagged fields of one partition.
+const PER_PARTITION_BYTES: usize = 24;
+
+/// One group a request asks about, across wire versions.
+#[derive(Debug, Clone)]
+struct GroupQuery {
+ group_id: StrBytes,
+ /// `None` asks for every committed offset of the group.
+ topics: Option<Vec<(TopicName, Vec<i32>)>>,
+}
+
+impl GroupQuery {
+ /// One query per group the request names: one below v8, the `groups` array from v8.
+ fn all(version: i16, request: &OffsetFetchRequest) -> Vec<Self> {
+ if version >= FIRST_BATCHED_VERSION {
+ return request
+ .groups
+ .iter()
+ .map(|group| Self {
+ group_id: group.group_id.0.clone(),
+ topics: group.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ })
+ .collect();
+ }
+ vec![Self {
+ group_id: request.group_id.0.clone(),
+ topics: request.topics.as_ref().map(|topics| {
+ topics
+ .iter()
+ .map(|topic| (topic.name.clone(), topic.partition_indexes.clone()))
+ .collect()
+ }),
+ }]
+ }
+}
+
+/// The answer for one group. A group error carries no topics.
+#[derive(Debug, Clone)]
+struct GroupAnswer {
+ error: i16,
+ topics: Vec<TopicAnswer>,
+}
+
+impl GroupAnswer {
+ const fn error(error: i16) -> Self {
+ Self {
+ error,
+ topics: Vec::new(),
+ }
+ }
+}
+
+#[derive(Debug, Clone)]
+struct TopicAnswer {
+ name: TopicName,
+ partitions: Vec<PartitionAnswer>,
+}
+
+/// `offset` is -1 when nothing is committed.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct PartitionAnswer {
+ index: i32,
+ offset: i64,
+ error: i16,
+}
+
+pub async fn handle(state: &GatewayState, api_version: i16, body: Bytes) -> HandleOutcome {
+ // A v1 answer has nowhere to put a request error, and a partition left out reads as "nothing
+ // committed", so an unreadable request closes the connection at every version.
+ if !is_supported_version(API_KEY_OFFSET_FETCH, api_version) {
+ return HandleOutcome::Close;
+ }
+ let request = match decode_guarded::<OffsetFetchRequest>(api_version, body, |version, body| {
+ validate_offset_fetch_shape(version, body, state.max_frame_size)
+ }) {
+ Ok(request) => request,
+ Err(error) => {
+ // debug!, not warn!: attacker-controlled, not operator-actionable.
+ tracing::debug!(%error, api_version, "Failed to decode OffsetFetch request");
+ return HandleOutcome::Close;
+ }
+ };
+
+ let queries = GroupQuery::all(api_version, &request);
+ let deadline = Instant::now() + state.groups.offset_deadline();
Review Comment:
went with the cap. the gateway remembers who heartbeats on each connection, and an offset call there takes at most half of what's left of that member's session. back to back calls keep halving, the heartbeat behind them still makes it
##########
gateways/kafka/src/group/state.rs:
##########
@@ -1112,6 +1119,61 @@ pub fn leave_step(groups: &mut Groups, request: &LeaveRequest, now: Instant) ->
}
}
+/// Kafka's `ClassicGroup.validateOffsetCommit`. Never creates a group.
+///
+/// A group with no members is Kafka's `Empty`: a negative generation commits without a member. A
+/// gateway restart empties every group, so a member's next commit gets `UNKNOWN_MEMBER_ID` and it
+/// rejoins. Kafka's v9 answer, `GROUP_ID_NOT_FOUND`, is fatal in the Java consumer.
+pub fn commit_step(groups: &mut Groups, request: &CommitRequest, now: Instant) -> i16 {
+ if !is_valid_group_id(&request.group_id) {
+ return ERROR_INVALID_GROUP_ID;
+ }
+ let empty = || {
+ if request.generation_id < 0 {
Review Comment:
refusing gen < 0 without a local group breaks assign() consumers and offset resets after a restart, kafka allows both on an empty group. a real fix needs group state shared between gateways, and we only support one gateway per bootstrap endpoint (CONSUMER_GROUPS.md). documented it in OFFSET_STORAGE.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on PR #4415:
URL: https://github.com/apache/iggy/pull/4415#issuecomment-6045267475
the two outside the diff are in a85d90f:
- http answers 503 TransientNotAccepted for NotFound now, like tcp. doc fixed, small test added
- get_consumer_offset doc has the TransientNotAccepted line now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal commented on PR #4415:
URL: https://github.com/apache/iggy/pull/4415#issuecomment-6055194459
/ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(gateways): answer OffsetCommit and OffsetFetch [iggy]
Posted by "krishvishal (via GitHub)" <gi...@apache.org>.
krishvishal merged PR #4415:
URL: https://github.com/apache/iggy/pull/4415
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@iggy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org