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