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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
