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