You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/06/25 08:31:35 UTC
(iggy) branch master updated: feat(server-ng): port consumer group (#3527)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 1932014c7 feat(server-ng): port consumer group (#3527)
1932014c7 is described below
commit 1932014c7973cdd98adae7da7a681c99a047cbd4
Author: Grzegorz Koszyk <11...@users.noreply.github.com>
AuthorDate: Thu Jun 25 10:31:30 2026 +0200
feat(server-ng): port consumer group (#3527)
This PR ports the consumer group mechanism to server-ng, maintaining the
cooperative rebalancing mechanism and extending it to accommodate for
the fact that the server is distributed.
---
core/binary_protocol/src/codes.rs | 2 +
core/binary_protocol/src/consensus/header.rs | 4 +
core/binary_protocol/src/consensus/operation.rs | 21 +-
core/binary_protocol/src/dispatch.rs | 22 +-
core/binary_protocol/src/lib.rs | 2 +-
core/binary_protocol/src/primitives/consumer.rs | 4 +-
.../src/requests/consumer_groups/mod.rs | 2 +
.../consumer_groups/sync_consumer_group.rs | 100 +++
.../src/responses/consumer_groups/mod.rs | 2 +
.../consumer_groups/sync_consumer_group.rs | 114 +++
core/common/src/consumer_group_client_state.rs | 219 +++++
core/common/src/error/iggy_error.rs | 4 +
core/common/src/lib.rs | 13 +
.../src/traits/binary_impls/consumer_groups.rs | 13 +
core/common/src/traits/binary_impls/messages.rs | 237 +++++
core/common/src/traits/binary_impls/system.rs | 5 +
core/common/src/traits/binary_transport.rs | 8 +
core/common/src/traits/system_client.rs | 7 +
core/consensus/src/observability.rs | 4 +
core/integration/src/harness/port_reserver.rs | 180 ++--
core/integration/tests/data_integrity/mod.rs | 13 +-
core/integration/tests/mod.rs | 28 +-
core/integration/tests/server/cg_vsr.rs | 85 ++
core/integration/tests/server/mod.rs | 5 +-
.../server/scenarios/authentication_scenario.rs | 12 +-
...onsumer_group_duplicate_name_create_scenario.rs | 118 +++
core/integration/tests/server/scenarios/mod.rs | 13 +-
core/metadata/src/impls/metadata.rs | 153 +++-
core/metadata/src/stm/consumer_group.rs | 971 +++++++++++++++------
core/metadata/src/stm/mux.rs | 23 +-
core/metadata/src/stm/snapshot.rs | 12 +-
core/metadata/src/stm/stream.rs | 536 +++++++++++-
core/partitions/src/iggy_partition.rs | 57 ++
.../src/client_wrappers/binary_system_client.rs | 12 +
core/sdk/src/clients/client.rs | 7 +
core/sdk/src/quic/quic_client.rs | 9 +
core/sdk/src/tcp/tcp_client.rs | 9 +
core/sdk/src/vsr.rs | 1 +
core/sdk/src/websocket/websocket_client.rs | 9 +
core/server-ng/src/bootstrap.rs | 41 +-
core/server-ng/src/consumer_group.rs | 257 ++++++
core/server-ng/src/dispatch.rs | 356 +++++++-
core/server-ng/src/lib.rs | 1 +
core/server-ng/src/partition_reconciler.rs | 386 +++++++-
core/server-ng/src/responses.rs | 221 ++++-
core/server-ng/src/session_manager.rs | 55 +-
core/shard/src/lib.rs | 41 +
core/simulator/src/deps.rs | 3 +-
core/simulator/src/replica.rs | 4 +-
49 files changed, 3929 insertions(+), 472 deletions(-)
diff --git a/core/binary_protocol/src/codes.rs b/core/binary_protocol/src/codes.rs
index ae33a9814..3a538130c 100644
--- a/core/binary_protocol/src/codes.rs
+++ b/core/binary_protocol/src/codes.rs
@@ -90,6 +90,7 @@ pub const CREATE_CONSUMER_GROUP_CODE: u32 = 602;
pub const DELETE_CONSUMER_GROUP_CODE: u32 = 603;
pub const JOIN_CONSUMER_GROUP_CODE: u32 = 604;
pub const LEAVE_CONSUMER_GROUP_CODE: u32 = 605;
+pub const SYNC_CONSUMER_GROUP_CODE: u32 = 606;
/// Lookup the human-readable name for a command code.
///
@@ -158,6 +159,7 @@ mod tests {
DELETE_CONSUMER_GROUP_CODE,
JOIN_CONSUMER_GROUP_CODE,
LEAVE_CONSUMER_GROUP_CODE,
+ SYNC_CONSUMER_GROUP_CODE,
];
#[test]
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index 773a2bb17..7d0ad045b 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -369,6 +369,9 @@ pub enum EvictionReason {
InvalidToken = 10,
UserInactive = 11,
SessionError = 12,
+ /// Client missed heartbeats past the configured threshold; the server
+ /// evicted the session (and dropped it from any consumer groups).
+ StaleClient = 13,
}
// EvictionHeader - primary -> client (session-terminal, no body)
@@ -1075,6 +1078,7 @@ mod tests {
assert_eq!(EvictionReason::InvalidToken as u8, 10);
assert_eq!(EvictionReason::UserInactive as u8, 11);
assert_eq!(EvictionReason::SessionError as u8, 12);
+ assert_eq!(EvictionReason::StaleClient as u8, 13);
}
#[test]
diff --git a/core/binary_protocol/src/consensus/operation.rs b/core/binary_protocol/src/consensus/operation.rs
index 2a6f72b3a..7d38f3bac 100644
--- a/core/binary_protocol/src/consensus/operation.rs
+++ b/core/binary_protocol/src/consensus/operation.rs
@@ -45,6 +45,15 @@ pub enum Operation {
// Internal metadata operations (journal / replica-only)
CreateTopicWithAssignments = 64,
CreatePartitionsWithAssignments = 65,
+ /// Server-originated: drop a disconnected client from every consumer group
+ /// it joined and rebalance. Applied as a deterministic side-effect of the
+ /// `Logout` commit (not its own client command), so it has no wire code.
+ RemoveConsumerGroupMember = 66,
+
+ /// Reconciler-originated: complete a pending cooperative revocation, moving
+ /// a drained partition to its target member. Submitted through metadata
+ /// consensus by the partition reconciler; no client wire code.
+ CompleteConsumerGroupRevocation = 67,
// Metadata operations (shard 0)
CreateStream = 128,
@@ -70,6 +79,8 @@ pub enum Operation {
UpdatePermissions = 145,
CreatePersonalAccessToken = 146,
DeletePersonalAccessToken = 147,
+ JoinConsumerGroup = 148,
+ LeaveConsumerGroup = 149,
// Partition operations (routed by namespace)
SendMessages = 160,
@@ -121,6 +132,8 @@ impl Operation {
| Self::UpdatePermissions
| Self::CreatePersonalAccessToken
| Self::DeletePersonalAccessToken
+ | Self::JoinConsumerGroup
+ | Self::LeaveConsumerGroup
)
}
@@ -160,7 +173,9 @@ impl Operation {
| Self::Logout
| Self::NonReplicated
| Self::CreateTopicWithAssignments
- | Self::CreatePartitionsWithAssignments => None,
+ | Self::CreatePartitionsWithAssignments
+ | Self::RemoveConsumerGroupMember
+ | Self::CompleteConsumerGroupRevocation => None,
Self::CreateStream
| Self::UpdateStream
| Self::DeleteStream
@@ -181,6 +196,8 @@ impl Operation {
| Self::UpdatePermissions
| Self::CreatePersonalAccessToken
| Self::DeletePersonalAccessToken
+ | Self::JoinConsumerGroup
+ | Self::LeaveConsumerGroup
| Self::SendMessages
| Self::StoreConsumerOffset
| Self::DeleteConsumerOffset
@@ -231,6 +248,8 @@ mod tests {
Operation::UpdatePermissions,
Operation::CreatePersonalAccessToken,
Operation::DeletePersonalAccessToken,
+ Operation::JoinConsumerGroup,
+ Operation::LeaveConsumerGroup,
Operation::SendMessages,
Operation::StoreConsumerOffset,
Operation::DeleteConsumerOffset,
diff --git a/core/binary_protocol/src/dispatch.rs b/core/binary_protocol/src/dispatch.rs
index c8d7b307b..d455d0dea 100644
--- a/core/binary_protocol/src/dispatch.rs
+++ b/core/binary_protocol/src/dispatch.rs
@@ -180,8 +180,17 @@ pub const COMMAND_TABLE: &[CommandMeta] = &[
"consumer_group.delete",
Operation::DeleteConsumerGroup,
),
- CommandMeta::non_replicated(JOIN_CONSUMER_GROUP_CODE, "consumer_group.join"),
- CommandMeta::non_replicated(LEAVE_CONSUMER_GROUP_CODE, "consumer_group.leave"),
+ CommandMeta::replicated(
+ JOIN_CONSUMER_GROUP_CODE,
+ "consumer_group.join",
+ Operation::JoinConsumerGroup,
+ ),
+ CommandMeta::replicated(
+ LEAVE_CONSUMER_GROUP_CODE,
+ "consumer_group.leave",
+ Operation::LeaveConsumerGroup,
+ ),
+ CommandMeta::non_replicated(SYNC_CONSUMER_GROUP_CODE, "consumer_group.sync"),
// Login + Register (PAT - Personal Access Token variant)
CommandMeta::non_replicated(LOGIN_REGISTER_WITH_PAT_CODE, "user.login_register_with_pat"),
];
@@ -245,7 +254,8 @@ pub const fn lookup_command(code: u32) -> Option<&'static CommandMeta> {
DELETE_CONSUMER_GROUP_CODE => 47,
JOIN_CONSUMER_GROUP_CODE => 48,
LEAVE_CONSUMER_GROUP_CODE => 49,
- LOGIN_REGISTER_WITH_PAT_CODE => 50,
+ SYNC_CONSUMER_GROUP_CODE => 50,
+ LOGIN_REGISTER_WITH_PAT_CODE => 51,
_ => return None,
};
Some(&COMMAND_TABLE[idx])
@@ -272,6 +282,8 @@ pub const fn lookup_by_operation(op: Operation) -> Option<&'static CommandMeta>
Operation::DeleteSegments => 43,
Operation::CreateConsumerGroup => 46,
Operation::DeleteConsumerGroup => 47,
+ Operation::JoinConsumerGroup => 48,
+ Operation::LeaveConsumerGroup => 49,
Operation::CreateUser => 9,
Operation::UpdateUser => 11,
Operation::DeleteUser => 10,
@@ -286,6 +298,8 @@ pub const fn lookup_by_operation(op: Operation) -> Option<&'static CommandMeta>
Operation::DeleteConsumerOffset2 => 28,
Operation::CreateTopicWithAssignments
| Operation::CreatePartitionsWithAssignments
+ | Operation::RemoveConsumerGroupMember
+ | Operation::CompleteConsumerGroupRevocation
| Operation::Reserved
| Operation::Register
| Operation::Logout
@@ -409,6 +423,8 @@ mod tests {
Operation::UpdatePermissions,
Operation::CreatePersonalAccessToken,
Operation::DeletePersonalAccessToken,
+ Operation::JoinConsumerGroup,
+ Operation::LeaveConsumerGroup,
Operation::SendMessages,
Operation::StoreConsumerOffset,
Operation::DeleteConsumerOffset,
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index 00e2e5653..cd2b14a33 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -82,7 +82,7 @@ pub use message_view::{
WireMessageIterator, WireMessageIteratorMut, WireMessageView, WireMessageViewMut,
};
pub use primitives::ack_level::AckLevel;
-pub use primitives::consumer::WireConsumer;
+pub use primitives::consumer::{KIND_CONSUMER_GROUP, WireConsumer};
pub use primitives::identifier::{MAX_WIRE_NAME_LENGTH, WireIdentifier, WireName};
pub use primitives::partition_assignment::CreatedPartitionAssignment;
pub use primitives::partitioning::{MAX_MESSAGES_KEY_LENGTH, WirePartitioning};
diff --git a/core/binary_protocol/src/primitives/consumer.rs b/core/binary_protocol/src/primitives/consumer.rs
index 255b6a38a..202de84f1 100644
--- a/core/binary_protocol/src/primitives/consumer.rs
+++ b/core/binary_protocol/src/primitives/consumer.rs
@@ -21,7 +21,9 @@ use crate::codec::{WireDecode, WireEncode, read_u8};
use bytes::{BufMut, BytesMut};
const KIND_CONSUMER: u8 = 1;
-const KIND_CONSUMER_GROUP: u8 = 2;
+/// Wire discriminant for a consumer-group consumer (vs a single `Consumer`).
+/// Public so the server dispatch can match on it by name instead of a raw `2`.
+pub const KIND_CONSUMER_GROUP: u8 = 2;
/// Wire consumer type. Identifies either a single consumer or a consumer group.
///
diff --git a/core/binary_protocol/src/requests/consumer_groups/mod.rs b/core/binary_protocol/src/requests/consumer_groups/mod.rs
index e82c728e8..f02433f49 100644
--- a/core/binary_protocol/src/requests/consumer_groups/mod.rs
+++ b/core/binary_protocol/src/requests/consumer_groups/mod.rs
@@ -21,6 +21,7 @@ pub mod get_consumer_group;
pub mod get_consumer_groups;
pub mod join_consumer_group;
pub mod leave_consumer_group;
+pub mod sync_consumer_group;
pub use create_consumer_group::CreateConsumerGroupRequest;
pub use delete_consumer_group::DeleteConsumerGroupRequest;
@@ -28,3 +29,4 @@ pub use get_consumer_group::GetConsumerGroupRequest;
pub use get_consumer_groups::GetConsumerGroupsRequest;
pub use join_consumer_group::JoinConsumerGroupRequest;
pub use leave_consumer_group::LeaveConsumerGroupRequest;
+pub use sync_consumer_group::SyncConsumerGroupRequest;
diff --git a/core/binary_protocol/src/requests/consumer_groups/sync_consumer_group.rs b/core/binary_protocol/src/requests/consumer_groups/sync_consumer_group.rs
new file mode 100644
index 000000000..33befcf4a
--- /dev/null
+++ b/core/binary_protocol/src/requests/consumer_groups/sync_consumer_group.rs
@@ -0,0 +1,100 @@
+// 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.
+
+use crate::WireError;
+use crate::WireIdentifier;
+use crate::codec::{WireDecode, WireEncode};
+use bytes::BytesMut;
+
+/// `SyncConsumerGroup` request.
+///
+/// A member asks the coordinator for its current partition assignment (and the
+/// group generation) so it can select partitions to poll client-side.
+///
+/// Wire format: `[stream_id][topic_id][group_id]`
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SyncConsumerGroupRequest {
+ pub stream_id: WireIdentifier,
+ pub topic_id: WireIdentifier,
+ pub group_id: WireIdentifier,
+}
+
+impl WireEncode for SyncConsumerGroupRequest {
+ fn encoded_size(&self) -> usize {
+ self.stream_id.encoded_size() + self.topic_id.encoded_size() + self.group_id.encoded_size()
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ self.stream_id.encode(buf);
+ self.topic_id.encode(buf);
+ self.group_id.encode(buf);
+ }
+}
+
+impl WireDecode for SyncConsumerGroupRequest {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
+ let mut pos = 0;
+ let (stream_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ let (topic_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ let (group_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ Ok((
+ Self {
+ stream_id,
+ topic_id,
+ group_id,
+ },
+ pos,
+ ))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn roundtrip_named() {
+ let req = SyncConsumerGroupRequest {
+ stream_id: WireIdentifier::named("stream-1").unwrap(),
+ topic_id: WireIdentifier::named("topic-1").unwrap(),
+ group_id: WireIdentifier::named("group-1").unwrap(),
+ };
+ let bytes = req.to_bytes();
+ let (decoded, consumed) = SyncConsumerGroupRequest::decode(&bytes).unwrap();
+ assert_eq!(consumed, bytes.len());
+ assert_eq!(decoded, req);
+ }
+
+ #[test]
+ fn truncated_returns_error() {
+ let req = SyncConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(1),
+ topic_id: WireIdentifier::numeric(2),
+ group_id: WireIdentifier::numeric(3),
+ };
+ let bytes = req.to_bytes();
+ for i in 0..bytes.len() {
+ assert!(
+ SyncConsumerGroupRequest::decode(&bytes[..i]).is_err(),
+ "expected error for truncation at byte {i}"
+ );
+ }
+ }
+}
diff --git a/core/binary_protocol/src/responses/consumer_groups/mod.rs b/core/binary_protocol/src/responses/consumer_groups/mod.rs
index 17c72a8a2..5abd62fc0 100644
--- a/core/binary_protocol/src/responses/consumer_groups/mod.rs
+++ b/core/binary_protocol/src/responses/consumer_groups/mod.rs
@@ -22,6 +22,7 @@ pub mod get_consumer_group;
pub mod get_consumer_groups;
mod join_consumer_group;
mod leave_consumer_group;
+pub mod sync_consumer_group;
pub use super::EmptyResponse;
pub use consumer_group_response::ConsumerGroupResponse;
@@ -31,3 +32,4 @@ pub use get_consumer_group::{ConsumerGroupDetailsResponse, ConsumerGroupMemberRe
pub use get_consumer_groups::GetConsumerGroupsResponse;
pub use join_consumer_group::JoinConsumerGroupResponse;
pub use leave_consumer_group::LeaveConsumerGroupResponse;
+pub use sync_consumer_group::SyncConsumerGroupResponse;
diff --git a/core/binary_protocol/src/responses/consumer_groups/sync_consumer_group.rs b/core/binary_protocol/src/responses/consumer_groups/sync_consumer_group.rs
new file mode 100644
index 000000000..9f9c3283a
--- /dev/null
+++ b/core/binary_protocol/src/responses/consumer_groups/sync_consumer_group.rs
@@ -0,0 +1,114 @@
+// 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.
+
+use crate::WireError;
+use crate::codec::{WireDecode, WireEncode, capped_capacity, read_u32_le, read_u64_le};
+use bytes::{BufMut, BytesMut};
+
+/// `SyncConsumerGroup` response.
+///
+/// The requesting member's current partition assignment plus the group
+/// generation it belongs to. The client caches this and re-syncs when the
+/// generation advances (rebalance) or a poll is fenced.
+///
+/// Wire format: `[generation:8][partitions_count:4][partition_id:4]*`
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SyncConsumerGroupResponse {
+ pub generation: u64,
+ pub partitions: Vec<u32>,
+}
+
+impl WireEncode for SyncConsumerGroupResponse {
+ fn encoded_size(&self) -> usize {
+ 8 + 4 + self.partitions.len() * 4
+ }
+
+ #[allow(clippy::cast_possible_truncation)]
+ fn encode(&self, buf: &mut BytesMut) {
+ buf.put_u64_le(self.generation);
+ buf.put_u32_le(self.partitions.len() as u32);
+ for &partition_id in &self.partitions {
+ buf.put_u32_le(partition_id);
+ }
+ }
+}
+
+impl WireDecode for SyncConsumerGroupResponse {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
+ let generation = read_u64_le(buf, 0)?;
+ let partitions_count = read_u32_le(buf, 8)?;
+ let remaining = buf.len().saturating_sub(12);
+ let mut partitions =
+ Vec::with_capacity(capped_capacity(partitions_count as usize, remaining, 4));
+ let mut offset = 12;
+ for _ in 0..partitions_count {
+ partitions.push(read_u32_le(buf, offset)?);
+ offset += 4;
+ }
+ Ok((
+ Self {
+ generation,
+ partitions,
+ },
+ offset,
+ ))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn roundtrip() {
+ let response = SyncConsumerGroupResponse {
+ generation: 7,
+ partitions: vec![0, 2, 4],
+ };
+ let bytes = response.to_bytes();
+ let (decoded, consumed) = SyncConsumerGroupResponse::decode(&bytes).unwrap();
+ assert_eq!(consumed, bytes.len());
+ assert_eq!(decoded, response);
+ }
+
+ #[test]
+ fn roundtrip_empty_assignment() {
+ let response = SyncConsumerGroupResponse {
+ generation: 1,
+ partitions: vec![],
+ };
+ let bytes = response.to_bytes();
+ let (decoded, consumed) = SyncConsumerGroupResponse::decode(&bytes).unwrap();
+ assert_eq!(consumed, bytes.len());
+ assert_eq!(decoded, response);
+ }
+
+ #[test]
+ fn truncated_returns_error() {
+ let response = SyncConsumerGroupResponse {
+ generation: 7,
+ partitions: vec![1, 2],
+ };
+ let bytes = response.to_bytes();
+ for i in 0..bytes.len() {
+ assert!(
+ SyncConsumerGroupResponse::decode(&bytes[..i]).is_err(),
+ "expected error for truncation at byte {i}"
+ );
+ }
+ }
+}
diff --git a/core/common/src/consumer_group_client_state.rs b/core/common/src/consumer_group_client_state.rs
new file mode 100644
index 000000000..74a612ebb
--- /dev/null
+++ b/core/common/src/consumer_group_client_state.rs
@@ -0,0 +1,219 @@
+// 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.
+
+//! Client-side consumer-group + partitioning state for the VSR transport.
+//!
+//! Under VSR the client routes partition ops by namespace, so it resolves
+//! partitioning locally (the broker never picks a partition, matching Kafka):
+//! - consumer-group polls select the next of the member's assigned partitions
+//! (round-robin) from the cached assignment synced from the coordinator;
+//! - `Balanced` produce round-robins per topic; `MessagesKey` hashes the key.
+//!
+//! Cursors must persist across calls, so this lives on the long-lived
+//! transport. The coordinator fences stale selections (rebalance), prompting a
+//! re-sync that resets the cursor.
+
+use crate::Identifier;
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+#[derive(Debug, Default, Clone)]
+struct GroupAssignment {
+ partitions: Vec<u32>,
+ generation: u64,
+ cursor: usize,
+}
+
+/// Per-transport cache of consumer-group assignments + partition counts used to
+/// resolve partitioning client-side. Keys are `stream|topic[|group]` strings
+/// built from the request identifiers.
+#[derive(Debug, Default)]
+pub struct ConsumerGroupClientState {
+ assignments: Mutex<HashMap<String, GroupAssignment>>,
+ balanced_cursors: Mutex<HashMap<String, usize>>,
+ partition_counts: Mutex<HashMap<String, u32>>,
+ /// Identifiers of the joined groups, so the heartbeat can rebuild a sync
+ /// request without re-deriving them from the cache key. Keyed by the same
+ /// `stream|topic|group` string as `assignments`.
+ joined_groups: Mutex<HashMap<String, (Identifier, Identifier, Identifier)>>,
+}
+
+impl ConsumerGroupClientState {
+ #[must_use]
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// True if a non-empty assignment is cached for the group.
+ #[must_use]
+ pub fn has_assignment(&self, key: &str) -> bool {
+ self.assignments
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .get(key)
+ .is_some_and(|assignment| !assignment.partitions.is_empty())
+ }
+
+ /// Replace the cached assignment for a group. A generation change (a
+ /// rebalance) resets the round-robin cursor so selection restarts cleanly.
+ pub fn set_assignment(&self, key: String, generation: u64, partitions: Vec<u32>) {
+ let mut map = self
+ .assignments
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let entry = map.entry(key).or_default();
+ if entry.generation != generation {
+ entry.cursor = 0;
+ }
+ entry.generation = generation;
+ entry.partitions = partitions;
+ }
+
+ /// Drop a group's cached assignment (after a fence rejection) so the next
+ /// poll re-syncs.
+ pub fn invalidate_assignment(&self, key: &str) {
+ self.assignments
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(key);
+ }
+
+ /// The next assigned partition for a group poll, advancing the cursor.
+ /// `None` when nothing is cached / the assignment is empty.
+ #[must_use]
+ pub fn next_group_partition(&self, key: &str) -> Option<u32> {
+ let mut map = self
+ .assignments
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let assignment = map.get_mut(key)?;
+ if assignment.partitions.is_empty() {
+ return None;
+ }
+ let index = assignment.cursor % assignment.partitions.len();
+ assignment.cursor = assignment.cursor.wrapping_add(1);
+ Some(assignment.partitions[index])
+ }
+
+ /// The next `Balanced` produce partition for a topic, advancing the cursor.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn next_balanced_partition(&self, key: &str, partition_count: u32) -> u32 {
+ if partition_count == 0 {
+ return 0;
+ }
+ let mut map = self
+ .balanced_cursors
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner);
+ let cursor = map.entry(key.to_owned()).or_default();
+ let partition = (*cursor % partition_count as usize) as u32;
+ *cursor = cursor.wrapping_add(1);
+ partition
+ }
+
+ #[must_use]
+ pub fn partition_count(&self, key: &str) -> Option<u32> {
+ self.partition_counts
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .get(key)
+ .copied()
+ }
+
+ pub fn set_partition_count(&self, key: String, partition_count: u32) {
+ self.partition_counts
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .insert(key, partition_count);
+ }
+
+ /// Record a joined group's identifiers so the heartbeat can re-sync it.
+ pub fn register_group(
+ &self,
+ key: String,
+ stream_id: Identifier,
+ topic_id: Identifier,
+ group_id: Identifier,
+ ) {
+ self.joined_groups
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .insert(key, (stream_id, topic_id, group_id));
+ }
+
+ /// Forget a group (after leave / delete) so the heartbeat stops re-syncing.
+ pub fn deregister_group(&self, key: &str) {
+ self.joined_groups
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .remove(key);
+ }
+
+ /// Identifiers of every group the client has joined on this transport.
+ #[must_use]
+ pub fn registered_groups(&self) -> Vec<(Identifier, Identifier, Identifier)> {
+ self.joined_groups
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .values()
+ .cloned()
+ .collect()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn group_partition_round_robins_then_wraps() {
+ let state = ConsumerGroupClientState::new();
+ state.set_assignment("s|t|g".to_owned(), 1, vec![0, 1, 2]);
+ let picks: Vec<u32> = (0..4)
+ .map(|_| state.next_group_partition("s|t|g").unwrap())
+ .collect();
+ assert_eq!(picks, vec![0, 1, 2, 0]);
+ }
+
+ #[test]
+ fn generation_change_resets_cursor() {
+ let state = ConsumerGroupClientState::new();
+ state.set_assignment("s|t|g".to_owned(), 1, vec![0, 1, 2]);
+ assert_eq!(state.next_group_partition("s|t|g"), Some(0));
+ assert_eq!(state.next_group_partition("s|t|g"), Some(1));
+ // Rebalance: new generation, one partition, cursor reset.
+ state.set_assignment("s|t|g".to_owned(), 2, vec![5]);
+ assert_eq!(state.next_group_partition("s|t|g"), Some(5));
+ }
+
+ #[test]
+ fn balanced_round_robins() {
+ let state = ConsumerGroupClientState::new();
+ let picks: Vec<u32> = (0..4)
+ .map(|_| state.next_balanced_partition("s|t", 3))
+ .collect();
+ assert_eq!(picks, vec![0, 1, 2, 0]);
+ }
+
+ #[test]
+ fn missing_assignment_yields_none() {
+ let state = ConsumerGroupClientState::new();
+ assert!(!state.has_assignment("s|t|g"));
+ assert_eq!(state.next_group_partition("s|t|g"), None);
+ }
+}
diff --git a/core/common/src/error/iggy_error.rs b/core/common/src/error/iggy_error.rs
index 816d8fac5..ced7b81e2 100644
--- a/core/common/src/error/iggy_error.rs
+++ b/core/common/src/error/iggy_error.rs
@@ -439,6 +439,10 @@ pub enum IggyError {
"Failed to delete consumer group info file for ID: {0} for topic with ID: {1} for stream with ID: {2}."
)]
CannotDeleteConsumerGroupInfo(usize, Identifier, Identifier) = 5008,
+ #[error(
+ "Consumer group member with client ID: {0} does not own partition: {1} at the current generation (rebalance in progress)."
+ )]
+ ConsumerGroupPartitionNotOwned(u32, u32) = 5009,
#[error("Base offset is missing")]
MissingBaseOffsetRetainedMessageBatch = 6000,
#[error("Last offset delta is missing")]
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index 966e0335d..9b8032043 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+#[cfg(feature = "vsr")]
+pub mod consumer_group_client_state;
mod error;
pub mod http;
mod macros;
@@ -28,6 +30,17 @@ pub use error::iggy_error::{IggyError, IggyErrorDiscriminants};
// Locking is feature gated, thus only mod level re-export.
pub mod locking;
pub use chrono::{DateTime, Duration as ChronoDuration, Utc};
+#[cfg(feature = "vsr")]
+pub use consumer_group_client_state::ConsumerGroupClientState;
+
+/// Sentinel `partition_id` in an otherwise-empty poll reply that tells the
+/// client its cached consumer-group assignment is stale (a generation fence
+/// rejected the partition it polled) and it must re-sync before retrying. The
+/// VSR reply header carries no status field yet, so this rides the empty-poll
+/// body -- an application payload, not a wire-format header. Never a real
+/// partition id (those are small, dense, zero-based), so it can't collide with
+/// a genuine end-of-partition empty poll, which echoes the real partition id.
+pub const RESYNC_REQUIRED_PARTITION_SENTINEL: u32 = u32::MAX;
pub use http::consumer_groups::*;
pub use http::consumer_offsets::*;
pub use http::messages::*;
diff --git a/core/common/src/traits/binary_impls/consumer_groups.rs b/core/common/src/traits/binary_impls/consumer_groups.rs
index bcf3ddb37..a47dbb97a 100644
--- a/core/common/src/traits/binary_impls/consumer_groups.rs
+++ b/core/common/src/traits/binary_impls/consumer_groups.rs
@@ -156,6 +156,12 @@ impl<B: BinaryClient> ConsumerGroupClient for B {
.to_bytes(),
)
.await?;
+ // Joining changes this member's assignment (and the group generation);
+ // drop any cached assignment so the next group poll re-syncs. Key
+ // format matches `binary_impls::messages::group_cache_key`.
+ #[cfg(feature = "vsr")]
+ self.consumer_group_state()
+ .invalidate_assignment(&format!("{stream_id}|{topic_id}|{group_id}"));
Ok(())
}
@@ -179,6 +185,13 @@ impl<B: BinaryClient> ConsumerGroupClient for B {
.to_bytes(),
)
.await?;
+ #[cfg(feature = "vsr")]
+ {
+ let key = format!("{stream_id}|{topic_id}|{group_id}");
+ self.consumer_group_state().invalidate_assignment(&key);
+ // Stop the heartbeat from re-syncing a group we left.
+ self.consumer_group_state().deregister_group(&key);
+ }
Ok(())
}
}
diff --git a/core/common/src/traits/binary_impls/messages.rs b/core/common/src/traits/binary_impls/messages.rs
index ee62480b7..0ed3e67cb 100644
--- a/core/common/src/traits/binary_impls/messages.rs
+++ b/core/common/src/traits/binary_impls/messages.rs
@@ -24,14 +24,222 @@ use crate::{
Consumer, Identifier, IggyError, IggyMessage, MessageClient, Partitioning, PolledMessages,
PollingStrategy,
};
+#[cfg(feature = "vsr")]
+use crate::{ConsumerKind, PartitioningKind, TopicClient, calculate_32};
use bytes::BytesMut;
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::codec::WireDecode;
use iggy_binary_protocol::codec::WireEncode;
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::codes::SYNC_CONSUMER_GROUP_CODE;
use iggy_binary_protocol::codes::{
FLUSH_UNSAVED_BUFFER_CODE, POLL_MESSAGES_CODE, SEND_MESSAGES_CODE,
};
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::requests::consumer_groups::SyncConsumerGroupRequest;
use iggy_binary_protocol::requests::messages::{
FlushUnsavedBufferRequest, PollMessagesRequest, RawMessage, SendMessagesEncoder,
};
+#[cfg(feature = "vsr")]
+use iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse;
+
+/// Max attempts to resolve a fenced consumer-group poll: one re-sync after the
+/// coordinator rejects a stale assignment, then retry once.
+#[cfg(feature = "vsr")]
+const GROUP_POLL_MAX_ATTEMPTS: usize = 2;
+
+#[cfg(feature = "vsr")]
+fn group_cache_key(stream_id: &Identifier, topic_id: &Identifier, group_id: &Identifier) -> String {
+ format!("{stream_id}|{topic_id}|{group_id}")
+}
+
+#[cfg(feature = "vsr")]
+fn topic_cache_key(stream_id: &Identifier, topic_id: &Identifier) -> String {
+ format!("{stream_id}|{topic_id}")
+}
+
+/// Sync the requesting member's assignment from the coordinator into the
+/// transport cache. An empty reply means the member holds no partitions.
+#[cfg(feature = "vsr")]
+async fn sync_group_assignment<B: BinaryClient>(
+ client: &B,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ group_id: &Identifier,
+) -> Result<(), IggyError> {
+ let request = SyncConsumerGroupRequest {
+ stream_id: identifier_to_wire(stream_id)?,
+ topic_id: identifier_to_wire(topic_id)?,
+ group_id: identifier_to_wire(group_id)?,
+ };
+ let response = client
+ .send_raw_with_response(SYNC_CONSUMER_GROUP_CODE, request.to_bytes())
+ .await?;
+ let key = group_cache_key(stream_id, topic_id, group_id);
+ if response.is_empty() {
+ // Empty reply = the client is not a member of this group (or holds no
+ // partitions). Record the empty assignment but do NOT `register_group`:
+ // registering would make the heartbeat re-sync it for the connection's
+ // lifetime and leak a `joined_groups` entry per distinct non-member
+ // group ever polled. Only a real membership (non-empty reply) registers.
+ client
+ .consumer_group_state()
+ .set_assignment(key, 0, Vec::new());
+ return Ok(());
+ }
+ let (assignment, _) =
+ SyncConsumerGroupResponse::decode(&response).map_err(|_| IggyError::InvalidCommand)?;
+ client.consumer_group_state().register_group(
+ key.clone(),
+ stream_id.clone(),
+ topic_id.clone(),
+ group_id.clone(),
+ );
+ client
+ .consumer_group_state()
+ .set_assignment(key, assignment.generation, assignment.partitions);
+ Ok(())
+}
+
+/// Re-sync every joined group's assignment from the coordinator. Heartbeat
+/// driven so a member picks up a widened assignment (e.g. after a
+/// partition-count change) without first hitting an ownership fence. A failed
+/// per-group sync is logged and skipped so one bad group can't stall the rest.
+#[cfg(feature = "vsr")]
+pub(crate) async fn refresh_group_assignments<B: BinaryClient>(client: &B) {
+ for (stream_id, topic_id, group_id) in client.consumer_group_state().registered_groups() {
+ if let Err(error) = sync_group_assignment(client, &stream_id, &topic_id, &group_id).await {
+ tracing::warn!(
+ "Failed to refresh consumer-group assignment for {stream_id}|{topic_id}|{group_id}: {error}"
+ );
+ }
+ }
+}
+
+/// Resolve (and cache) the topic's partition count for client-side produce
+/// partitioning.
+#[cfg(feature = "vsr")]
+async fn topic_partition_count<B: BinaryClient>(
+ client: &B,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+) -> Result<u32, IggyError> {
+ let key = topic_cache_key(stream_id, topic_id);
+ if let Some(count) = client.consumer_group_state().partition_count(&key) {
+ return Ok(count);
+ }
+ let details = TopicClient::get_topic(client, stream_id, topic_id)
+ .await?
+ .ok_or_else(|| IggyError::TopicIdNotFound(topic_id.clone(), stream_id.clone()))?;
+ client
+ .consumer_group_state()
+ .set_partition_count(key, details.partitions_count);
+ Ok(details.partitions_count)
+}
+
+/// Resolve `Balanced` / `MessagesKey` to an explicit `PartitionId` client-side
+/// (the VSR broker only routes explicit partitions, matching Kafka).
+#[cfg(feature = "vsr")]
+async fn resolve_partitioning<B: BinaryClient>(
+ client: &B,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ partitioning: &Partitioning,
+) -> Result<Partitioning, IggyError> {
+ match partitioning.kind {
+ PartitioningKind::PartitionId => Ok(partitioning.clone()),
+ PartitioningKind::Balanced => {
+ let count = topic_partition_count(client, stream_id, topic_id).await?;
+ if count == 0 {
+ return Err(IggyError::TopicIdNotFound(
+ topic_id.clone(),
+ stream_id.clone(),
+ ));
+ }
+ let key = topic_cache_key(stream_id, topic_id);
+ let partition = client
+ .consumer_group_state()
+ .next_balanced_partition(&key, count);
+ Ok(Partitioning::partition_id(partition))
+ }
+ PartitioningKind::MessagesKey => {
+ let count = topic_partition_count(client, stream_id, topic_id).await?;
+ if count == 0 {
+ return Err(IggyError::TopicIdNotFound(
+ topic_id.clone(),
+ stream_id.clone(),
+ ));
+ }
+ let partition = calculate_32(&partitioning.value) % count;
+ Ok(Partitioning::partition_id(partition))
+ }
+ }
+}
+
+/// Poll a consumer group: select one of the member's assigned partitions
+/// (round-robin) and send an explicit-partition poll. A coordinator fence
+/// rejection (stale assignment after a rebalance) triggers one re-sync + retry.
+#[cfg(feature = "vsr")]
+async fn poll_group_messages<B: BinaryClient>(
+ client: &B,
+ stream_id: &Identifier,
+ topic_id: &Identifier,
+ consumer: &Consumer,
+ strategy: &PollingStrategy,
+ count: u32,
+ auto_commit: bool,
+) -> Result<PolledMessages, IggyError> {
+ let key = group_cache_key(stream_id, topic_id, &consumer.id);
+ if !client.consumer_group_state().has_assignment(&key) {
+ sync_group_assignment(client, stream_id, topic_id, &consumer.id).await?;
+ }
+ for _ in 0..GROUP_POLL_MAX_ATTEMPTS {
+ let Some(partition_id) = client.consumer_group_state().next_group_partition(&key) else {
+ // Synced but holds no partitions: not a member, or none assigned.
+ return Ok(PolledMessages::empty());
+ };
+ let request = PollMessagesRequest {
+ consumer: consumer_to_wire(consumer)?,
+ stream_id: identifier_to_wire(stream_id)?,
+ topic_id: identifier_to_wire(topic_id)?,
+ partition_id: Some(partition_id),
+ strategy: polling_strategy_to_wire(strategy),
+ count,
+ auto_commit,
+ };
+ match client
+ .send_raw_with_response(POLL_MESSAGES_CODE, request.to_bytes())
+ .await
+ {
+ Ok(response) => {
+ let polled = PolledMessages::from_bytes(response)?;
+ // The coordinator can't yet signal a generation fence as a typed
+ // error (no reply-header status), so it rides the empty-poll body
+ // as a sentinel partition id. Re-sync and retry, same as the
+ // typed error below; a genuine empty poll echoes the real id.
+ if polled.messages.is_empty()
+ && polled.partition_id == crate::RESYNC_REQUIRED_PARTITION_SENTINEL
+ {
+ client.consumer_group_state().invalidate_assignment(&key);
+ sync_group_assignment(client, stream_id, topic_id, &consumer.id).await?;
+ continue;
+ }
+ return Ok(polled);
+ }
+ Err(IggyError::ConsumerGroupPartitionNotOwned(..)) => {
+ client.consumer_group_state().invalidate_assignment(&key);
+ sync_group_assignment(client, stream_id, topic_id, &consumer.id).await?;
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ // Exhausted the retry budget on back-to-back fences (a rebalance landed on
+ // every attempt) -- rare, and the cursor is already re-synced. Surface an
+ // empty poll rather than `ConsumerGroupPartitionNotOwned(0, 0)`: the (0, 0)
+ // ids are fabricated and a normal rebalance must not look like a hard error
+ // to a CG app that doesn't special-case 5009. The caller just re-polls.
+ Ok(PolledMessages::empty())
+}
#[async_trait::async_trait]
impl<B: BinaryClient> MessageClient for B {
@@ -46,6 +254,22 @@ impl<B: BinaryClient> MessageClient for B {
auto_commit: bool,
) -> Result<PolledMessages, IggyError> {
fail_if_not_authenticated(self).await?;
+ // VSR: a consumer-group poll without an explicit partition is resolved
+ // client-side from the member's cached assignment (the broker routes
+ // explicit partitions only).
+ #[cfg(feature = "vsr")]
+ if consumer.kind == ConsumerKind::ConsumerGroup && partition_id.is_none() {
+ return poll_group_messages(
+ self,
+ stream_id,
+ topic_id,
+ consumer,
+ strategy,
+ count,
+ auto_commit,
+ )
+ .await;
+ }
let req = PollMessagesRequest {
consumer: consumer_to_wire(consumer)?,
stream_id: identifier_to_wire(stream_id)?,
@@ -69,6 +293,19 @@ impl<B: BinaryClient> MessageClient for B {
messages: &mut [IggyMessage],
) -> Result<(), IggyError> {
fail_if_not_authenticated(self).await?;
+ // VSR: resolve Balanced/MessagesKey to an explicit partition client-side.
+ // An explicit `PartitionId` needs no resolution, so borrow the input
+ // directly on that fast path instead of cloning its `value: Vec<u8>`.
+ #[cfg(feature = "vsr")]
+ let resolved_partitioning;
+ #[cfg(feature = "vsr")]
+ let partitioning = if partitioning.kind == PartitioningKind::PartitionId {
+ partitioning
+ } else {
+ resolved_partitioning =
+ resolve_partitioning(self, stream_id, topic_id, partitioning).await?;
+ &resolved_partitioning
+ };
let wire_stream_id = identifier_to_wire(stream_id)?;
let wire_topic_id = identifier_to_wire(topic_id)?;
let wire_partitioning = partitioning_to_wire(partitioning)?;
diff --git a/core/common/src/traits/binary_impls/system.rs b/core/common/src/traits/binary_impls/system.rs
index 8cf1d35bd..f3868fc90 100644
--- a/core/common/src/traits/binary_impls/system.rs
+++ b/core/common/src/traits/binary_impls/system.rs
@@ -87,6 +87,11 @@ impl<B: BinaryClient> SystemClient for B {
self.get_heartbeat_interval()
}
+ #[cfg(feature = "vsr")]
+ async fn refresh_consumer_group_assignments(&self) {
+ super::messages::refresh_group_assignments(self).await;
+ }
+
async fn snapshot(
&self,
compression: SnapshotCompression,
diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs
index 6eff73f4d..233d3825b 100644
--- a/core/common/src/traits/binary_transport.rs
+++ b/core/common/src/traits/binary_transport.rs
@@ -18,6 +18,8 @@
use crate::{ClientState, DiagnosticEvent, IggyDuration, IggyError};
use async_trait::async_trait;
use bytes::Bytes;
+#[cfg(feature = "vsr")]
+use std::sync::Arc;
#[async_trait]
pub trait BinaryTransport {
@@ -28,6 +30,12 @@ pub trait BinaryTransport {
async fn publish_event(&self, event: DiagnosticEvent);
async fn send_raw_with_response(&self, code: u32, payload: Bytes) -> Result<Bytes, IggyError>;
fn get_heartbeat_interval(&self) -> IggyDuration;
+
+ /// Per-transport consumer-group + partitioning cache used to resolve
+ /// partitioning client-side under VSR (the broker never picks a
+ /// partition). Shared via `Arc` so a refresh task can hold it.
+ #[cfg(feature = "vsr")]
+ fn consumer_group_state(&self) -> Arc<crate::ConsumerGroupClientState>;
}
/// Sealed marker. Downstream crates cannot implement
diff --git a/core/common/src/traits/system_client.rs b/core/common/src/traits/system_client.rs
index 149be8f3f..4ccee4de8 100644
--- a/core/common/src/traits/system_client.rs
+++ b/core/common/src/traits/system_client.rs
@@ -43,6 +43,13 @@ pub trait SystemClient {
/// Ping the server to check if it's alive.
async fn ping(&self) -> Result<(), IggyError>;
async fn heartbeat_interval(&self) -> IggyDuration;
+ /// Re-sync the cached consumer-group assignments from the coordinator.
+ ///
+ /// Driven off the heartbeat so a member picks up a new generation (e.g. a
+ /// partition-count change widened its assignment) without waiting for an
+ /// ownership-fence rejection. No-op for transports that resolve
+ /// partitioning server-side.
+ async fn refresh_consumer_group_assignments(&self) {}
/// Capture and package the current system state as a snapshot.
///
/// Authentication is required.
diff --git a/core/consensus/src/observability.rs b/core/consensus/src/observability.rs
index ce012d49c..662f4b128 100644
--- a/core/consensus/src/observability.rs
+++ b/core/consensus/src/observability.rs
@@ -620,6 +620,8 @@ pub const fn operation_as_str(operation: Operation) -> &'static str {
Operation::Reserved => "reserved",
Operation::CreateTopicWithAssignments => "create_topic_with_assignments",
Operation::CreatePartitionsWithAssignments => "create_partitions_with_assignments",
+ Operation::RemoveConsumerGroupMember => "remove_consumer_group_member",
+ Operation::CompleteConsumerGroupRevocation => "complete_consumer_group_revocation",
Operation::CreateStream => "create_stream",
Operation::UpdateStream => "update_stream",
Operation::DeleteStream => "delete_stream",
@@ -633,6 +635,8 @@ pub const fn operation_as_str(operation: Operation) -> &'static str {
Operation::DeleteSegments => "delete_segments",
Operation::CreateConsumerGroup => "create_consumer_group",
Operation::DeleteConsumerGroup => "delete_consumer_group",
+ Operation::JoinConsumerGroup => "join_consumer_group",
+ Operation::LeaveConsumerGroup => "leave_consumer_group",
Operation::CreateUser => "create_user",
Operation::UpdateUser => "update_user",
Operation::DeleteUser => "delete_user",
diff --git a/core/integration/src/harness/port_reserver.rs b/core/integration/src/harness/port_reserver.rs
index d4103d5d5..b65c0f161 100644
--- a/core/integration/src/harness/port_reserver.rs
+++ b/core/integration/src/harness/port_reserver.rs
@@ -28,7 +28,36 @@
use crate::harness::config::{IpAddrKind, TestServerConfig};
use crate::harness::error::TestBinaryError;
use socket2::{Domain, Protocol, Socket, Type};
+use std::collections::HashSet;
use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::sync::Mutex;
+
+/// Ports already handed out by any reservation in this process.
+///
+/// The reservation sockets set `SO_REUSEPORT` so the server can later bind the
+/// same port the reserver held. But `SO_REUSEPORT` also lets the kernel assign
+/// the *same* ephemeral port to two `bind(0)` calls from different threads, so
+/// under a multi-threaded test run two clusters can reserve the same port and
+/// the second server aborts on the cluster config's port-conflict validator.
+/// Tracking every handed-out port and rebinding on collision makes reservations
+/// unique across the whole process. Ports are claimed for the process lifetime
+/// (never returned to the pool): a release frees the socket so the server can
+/// bind, but re-handing the number out could still collide with that server.
+static RESERVED_PORTS: Mutex<Option<HashSet<u16>>> = Mutex::new(None);
+
+/// Bind retries before giving up. The ephemeral range dwarfs the ports a single
+/// test run consumes, so a fresh port is found almost immediately; the cap only
+/// guards against a pathological kernel that keeps returning claimed ports.
+const MAX_BIND_ATTEMPTS: usize = 100;
+
+/// Claim `port` process-wide. Returns `false` if it was already handed out.
+fn claim_port(port: u16) -> bool {
+ RESERVED_PORTS
+ .lock()
+ .expect("reserved-ports mutex poisoned")
+ .get_or_insert_with(HashSet::new)
+ .insert(port)
+}
/// A socket bound to a specific port, held to prevent reuse until released.
struct ReservedPort {
@@ -38,109 +67,84 @@ struct ReservedPort {
impl ReservedPort {
fn tcp(ip_kind: IpAddrKind) -> Result<Self, TestBinaryError> {
- let domain = match ip_kind {
- IpAddrKind::V4 => Domain::IPV4,
- IpAddrKind::V6 => Domain::IPV6,
- };
-
- let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP)).map_err(|e| {
- TestBinaryError::InvalidState {
- message: format!("Failed to create TCP socket: {e}"),
- }
- })?;
-
- socket
- .set_reuse_address(true)
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to set SO_REUSEADDR: {e}"),
- })?;
-
- #[cfg(unix)]
- socket
- .set_reuse_port(true)
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to set SO_REUSEPORT: {e}"),
- })?;
-
- let bind_addr: SocketAddr = match ip_kind {
- IpAddrKind::V4 => SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0),
- IpAddrKind::V6 => SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 0),
- };
-
- socket
- .bind(&bind_addr.into())
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to bind TCP socket: {e}"),
- })?;
-
- // Bind WITHOUT listen: a listening reservation joins the port's
- // SO_REUSEPORT group and accepts early dials into a backlog nobody
- // drains. During cluster startup a peer's replica connector dials
- // ports whose reservations are still held (release happens per-node
- // right before spawn), so each such dial used to wedge until the
- // 2s handshake-ack timeout. A bound-only socket still holds the
- // port against ephemeral allocation but answers RST, so dialers
- // fail fast and the reconnect sweep retries cleanly.
- let addr = socket
- .local_addr()
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to get local address: {e}"),
- })?
- .as_socket()
- .ok_or_else(|| TestBinaryError::InvalidState {
- message: "Socket address is not an IP address".to_string(),
- })?;
-
- Ok(Self { socket, addr })
+ Self::reserve_unique(ip_kind, Type::STREAM, Protocol::TCP)
}
fn udp(ip_kind: IpAddrKind) -> Result<Self, TestBinaryError> {
+ Self::reserve_unique(ip_kind, Type::DGRAM, Protocol::UDP)
+ }
+
+ /// Bind an ephemeral port and claim it process-wide, rebinding until the
+ /// kernel hands back a port no other reservation holds (see
+ /// [`RESERVED_PORTS`]).
+ fn reserve_unique(
+ ip_kind: IpAddrKind,
+ sock_type: Type,
+ protocol: Protocol,
+ ) -> Result<Self, TestBinaryError> {
let domain = match ip_kind {
IpAddrKind::V4 => Domain::IPV4,
IpAddrKind::V6 => Domain::IPV6,
};
-
- let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP)).map_err(|e| {
- TestBinaryError::InvalidState {
- message: format!("Failed to create UDP socket: {e}"),
- }
- })?;
-
- socket
- .set_reuse_address(true)
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to set SO_REUSEADDR: {e}"),
- })?;
-
- #[cfg(unix)]
- socket
- .set_reuse_port(true)
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to set SO_REUSEPORT: {e}"),
- })?;
-
let bind_addr: SocketAddr = match ip_kind {
IpAddrKind::V4 => SocketAddr::new(Ipv4Addr::LOCALHOST.into(), 0),
IpAddrKind::V6 => SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 0),
};
- socket
- .bind(&bind_addr.into())
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to bind UDP socket: {e}"),
+ for _ in 0..MAX_BIND_ATTEMPTS {
+ let socket = Socket::new(domain, sock_type, Some(protocol)).map_err(|e| {
+ TestBinaryError::InvalidState {
+ message: format!("Failed to create socket: {e}"),
+ }
})?;
- let addr = socket
- .local_addr()
- .map_err(|e| TestBinaryError::InvalidState {
- message: format!("Failed to get local address: {e}"),
- })?
- .as_socket()
- .ok_or_else(|| TestBinaryError::InvalidState {
- message: "Socket address is not an IP address".to_string(),
- })?;
+ socket
+ .set_reuse_address(true)
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to set SO_REUSEADDR: {e}"),
+ })?;
+
+ #[cfg(unix)]
+ socket
+ .set_reuse_port(true)
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to set SO_REUSEPORT: {e}"),
+ })?;
+
+ socket
+ .bind(&bind_addr.into())
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to bind socket: {e}"),
+ })?;
+
+ // Bind WITHOUT listen: a listening reservation joins the port's
+ // SO_REUSEPORT group and accepts early dials into a backlog nobody
+ // drains. During cluster startup a peer's replica connector dials
+ // ports whose reservations are still held (release happens per-node
+ // right before spawn), so each such dial used to wedge until the
+ // 2s handshake-ack timeout. A bound-only socket still holds the
+ // port against ephemeral allocation but answers RST, so dialers
+ // fail fast and the reconnect sweep retries cleanly.
+ let addr = socket
+ .local_addr()
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to get local address: {e}"),
+ })?
+ .as_socket()
+ .ok_or_else(|| TestBinaryError::InvalidState {
+ message: "Socket address is not an IP address".to_string(),
+ })?;
+
+ // SO_REUSEPORT lets the kernel reuse a port another reservation
+ // already holds; rebind (dropping this socket) until the claim wins.
+ if claim_port(addr.port()) {
+ return Ok(Self { socket, addr });
+ }
+ }
- Ok(Self { socket, addr })
+ Err(TestBinaryError::InvalidState {
+ message: format!("Failed to reserve a unique port after {MAX_BIND_ATTEMPTS} attempts"),
+ })
}
fn addr(&self) -> SocketAddr {
diff --git a/core/integration/tests/data_integrity/mod.rs b/core/integration/tests/data_integrity/mod.rs
index 136a807cd..6201bc744 100644
--- a/core/integration/tests/data_integrity/mod.rs
+++ b/core/integration/tests/data_integrity/mod.rs
@@ -15,18 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-// The legacy verifiers run only without `vsr`: they rely on Http/WebSocket/Quic
-// transports, the bench producer, or consumer groups, none of which the VSR
-// server serves yet.
+// Restart-based verifiers stay `not(vsr)`: they restart cluster nodes and
+// assert on-disk state afterward, which VSR can't satisfy until replica state
+// transfer lands (a restarted replica has no way to catch up).
#[cfg(not(feature = "vsr"))]
mod verify_after_server_restart;
#[cfg(not(feature = "vsr"))]
-mod verify_consumer_group_partition_assignment;
-#[cfg(not(feature = "vsr"))]
mod verify_no_plaintext_credentials_on_disk;
#[cfg(not(feature = "vsr"))]
mod verify_user_login_after_restart;
+// The cooperative-rebalance matrix runs under vsr too: it exercises server-ng's
+// consumer-group rebalancing (a VSR feature) and never restarts the cluster, so
+// the no-state-transfer limitation above does not apply. Green at 95/95.
+mod verify_consumer_group_partition_assignment;
+
// Cross-replica on-disk data identity is VSR-only.
#[cfg(feature = "vsr")]
mod verify_cluster_replica_data_identical;
diff --git a/core/integration/tests/mod.rs b/core/integration/tests/mod.rs
index e2dfcaf9e..c20bac18c 100644
--- a/core/integration/tests/mod.rs
+++ b/core/integration/tests/mod.rs
@@ -38,10 +38,30 @@ mod cluster;
mod config_provider;
#[cfg(not(feature = "vsr"))]
mod connectors;
-// Compiled in both lanes; members are gated individually (see its `mod.rs`).
-// Under `vsr` only the cross-replica on-disk identity test is active. The four
-// legacy verifiers stay non-vsr: they use Http/WebSocket/Quic, the bench
-// producer, or consumer groups, which the VSR server does not serve yet.
+// TODO(vsr): enable the `data_integrity` suite under the `vsr` feature once
+// the full VSR path is done. Tier 1 (`verify_user_login_after_restart`,
+// `verify_no_plaintext_credentials_on_disk`) is server-side ready (create_user
+// reply body, get_users/get_user via `frontend()`, change_password, cross-shard
+// committed-reply routing) -- proven by `sdk::hello_world::replicated_*` under
+// vsr. Blocked on: (1) these tests use `[Tcp, Http, Quic, WebSocket]` but only
+// Tcp works under vsr today (Http does no VSR framing; WebSocket/Quic hit the
+// compio `buffer.rs:83` bug); (2) the module also carries
+// `verify_after_server_restart` + `verify_consumer_group_partition_assignment`,
+// which need send/poll messages + consumer groups (out of Tier 1 scope).
+//
+// `verify_consumer_group_partition_assignment` (cooperative rebalance):
+// server-ng now implements cooperative rebalance (pending revocations, eager
+// handoff of never-polled/drained partitions via the join-time in-flight
+// gather, drain/timeout completion through the reconciler) -- most of this
+// suite passes. Kept gated under vsr for two reasons: (1) the 84-case
+// multi-client+reconnect matrix is flaky on the open replica-mesh-degradation
+// front (a peer link blip loses a metadata read), and (2) two cases still need
+// follow-ups -- `should_wait_for_manual_commit_before_completing_revocation`
+// needs event-driven (on-commit) completion for sub-second latency (the
+// reconciler completes on its 1s tick), and `should_skip_revoked_partitions_in_
+// round_robin` needs the source client to re-sync to its pollable set after a
+// revocation. server-ng's rebalance is also covered by `server::cg_vsr` +
+// `stale_client_consumer_group_scenario` (both green).
mod data_integrity;
#[cfg(not(feature = "vsr"))]
mod mcp;
diff --git a/core/integration/tests/server/cg_vsr.rs b/core/integration/tests/server/cg_vsr.rs
new file mode 100644
index 000000000..2448b41e1
--- /dev/null
+++ b/core/integration/tests/server/cg_vsr.rs
@@ -0,0 +1,85 @@
+// 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.
+
+//! Consumer-group scenarios that run against server-ng (vsr): round-robin
+//! membership (join) plus group polling (single- and multi-client), where the
+//! server picks each member's partition by advancing its round-robin cursor.
+
+use crate::server::scenarios::{
+ consumer_group_auto_commit_reconnection_scenario,
+ consumer_group_duplicate_name_create_scenario, consumer_group_join_scenario,
+ consumer_group_new_messages_after_restart_scenario, consumer_group_offset_cleanup_scenario,
+ consumer_group_with_multiple_clients_polling_messages_scenario,
+ consumer_group_with_single_client_polling_messages_scenario,
+};
+use integration::iggy_harness;
+
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn duplicate_name_create_preserves_live_group(harness: &TestHarness) {
+ consumer_group_duplicate_name_create_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn join(harness: &TestHarness) {
+ consumer_group_join_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn single_client(harness: &TestHarness) {
+ consumer_group_with_single_client_polling_messages_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn multiple_clients(harness: &TestHarness) {
+ consumer_group_with_multiple_clients_polling_messages_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn offset_cleanup(harness: &TestHarness) {
+ consumer_group_offset_cleanup_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn auto_commit_reconnection(harness: &TestHarness) {
+ consumer_group_auto_commit_reconnection_scenario::run(harness).await;
+}
+
+#[iggy_harness(
+ test_client_transport = [Tcp],
+ server(tcp.socket.override_defaults = true, tcp.socket.nodelay = true)
+)]
+async fn new_messages_after_restart(harness: &TestHarness) {
+ consumer_group_new_messages_after_restart_scenario::run(harness).await;
+}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 0aa5956ff..d83bcf7de 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -18,9 +18,12 @@
// a2a_jwt is HTTP-only (JWT against the HTTP transport); vsr has no HTTP.
#[cfg(not(feature = "vsr"))]
mod a2a_jwt;
-// Consumer groups are not implemented under vsr yet.
+// Polling-based consumer-group scenarios are not implemented under vsr yet.
#[cfg(not(feature = "vsr"))]
mod cg;
+// The ported round-robin membership join scenario runs against server-ng.
+#[cfg(feature = "vsr")]
+mod cg_vsr;
// 80-case race matrix with hardcoded HTTP variants (test_matrix bypasses
// the harness transport filter); revisit under vsr once basics are green.
#[cfg(not(feature = "vsr"))]
diff --git a/core/integration/tests/server/scenarios/authentication_scenario.rs b/core/integration/tests/server/scenarios/authentication_scenario.rs
index 9861c8573..41c7c4ae6 100644
--- a/core/integration/tests/server/scenarios/authentication_scenario.rs
+++ b/core/integration/tests/server/scenarios/authentication_scenario.rs
@@ -114,7 +114,7 @@ async fn test_all_commands_require_auth(client: &IggyClient) {
let name = entry.name;
// ================================================================
- // SKIPPED COMMANDS (10 total)
+ // SKIPPED COMMANDS (11 total)
// ================================================================
// No auth required
if matches!(
@@ -127,8 +127,14 @@ async fn test_all_commands_require_auth(client: &IggyClient) {
) {
continue;
}
- // Stateful - not supported on HTTP
- if matches!(code, JOIN_CONSUMER_GROUP_CODE | LEAVE_CONSUMER_GROUP_CODE) {
+ // Stateful - not supported on HTTP. `SYNC_CONSUMER_GROUP` is
+ // SDK-internal (issued during poll partition resolution), with no
+ // top-level client method to invoke unauthenticated here; its auth
+ // gate is exercised through the server-ng dispatch allowlist instead.
+ if matches!(
+ code,
+ JOIN_CONSUMER_GROUP_CODE | LEAVE_CONSUMER_GROUP_CODE | SYNC_CONSUMER_GROUP_CODE
+ ) {
continue;
}
// Special setup required
diff --git a/core/integration/tests/server/scenarios/consumer_group_duplicate_name_create_scenario.rs b/core/integration/tests/server/scenarios/consumer_group_duplicate_name_create_scenario.rs
new file mode 100644
index 000000000..fa1b928cb
--- /dev/null
+++ b/core/integration/tests/server/scenarios/consumer_group_duplicate_name_create_scenario.rs
@@ -0,0 +1,118 @@
+// 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.
+
+//! Re-creating a consumer group under an existing name must NOT destroy the
+//! live group: the existing group, its members, and their partition
+//! assignments have to survive (an ejected member would never recover). This
+//! is the non-destructive duplicate-create path that the (HTTP-bound, hence
+//! non-vsr) `concurrent_addition` cold matrix does not cover under vsr.
+
+use crate::server::scenarios::{
+ CONSUMER_GROUP_NAME, PARTITIONS_COUNT, STREAM_NAME, TOPIC_NAME, USERNAME_1, cleanup,
+ create_client, join_consumer_group,
+};
+use iggy::prelude::CompressionAlgorithm;
+use iggy::prelude::Identifier;
+use iggy::prelude::IggyExpiry;
+use iggy::prelude::MaxTopicSize;
+use iggy::prelude::{ConsumerGroupClient, StreamClient, TopicClient};
+use integration::harness::{TestHarness, assert_clean_system, create_user, login_user};
+
+pub async fn run(harness: &TestHarness) {
+ let system_client = harness
+ .root_client()
+ .await
+ .expect("Failed to get root client");
+ let client1 = create_client(harness).await;
+
+ system_client.create_stream(STREAM_NAME).await.unwrap();
+ system_client
+ .create_topic(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ TOPIC_NAME,
+ PARTITIONS_COUNT,
+ CompressionAlgorithm::default(),
+ None,
+ IggyExpiry::NeverExpire,
+ MaxTopicSize::ServerDefault,
+ )
+ .await
+ .unwrap();
+ let original_id = system_client
+ .create_consumer_group(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ CONSUMER_GROUP_NAME,
+ )
+ .await
+ .unwrap()
+ .id;
+
+ // A member joins and gets all partitions.
+ create_user(&system_client, USERNAME_1).await;
+ login_user(&client1, USERNAME_1).await;
+ join_consumer_group(&client1).await;
+
+ let before = get_group(&system_client).await;
+ assert_eq!(
+ before.members_count, 1,
+ "the joined member must be present before the duplicate create"
+ );
+
+ // Re-create the SAME name. The reply may be ok or an already-exists-style
+ // error; what matters is that the live group is left intact -- it must not
+ // be removed and replaced by a fresh empty group with a new id.
+ let _ = system_client
+ .create_consumer_group(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ CONSUMER_GROUP_NAME,
+ )
+ .await;
+
+ let after = get_group(&system_client).await;
+ assert_eq!(
+ after.id, original_id,
+ "duplicate create must not replace the group with a new id"
+ );
+ assert_eq!(
+ after.members_count, 1,
+ "duplicate create must not eject the existing member"
+ );
+ assert_eq!(after.members.len(), 1);
+ assert_eq!(
+ after.members[0].partitions_count, PARTITIONS_COUNT,
+ "the surviving member must keep its partition assignment"
+ );
+
+ cleanup(&system_client, true).await;
+ assert_clean_system(&system_client).await;
+}
+
+async fn get_group(
+ client: &iggy::clients::client::IggyClient,
+) -> iggy::prelude::ConsumerGroupDetails {
+ client
+ .get_consumer_group(
+ &Identifier::named(STREAM_NAME).unwrap(),
+ &Identifier::named(TOPIC_NAME).unwrap(),
+ &Identifier::named(CONSUMER_GROUP_NAME).unwrap(),
+ )
+ .await
+ .unwrap()
+ .expect("consumer group must still exist")
+}
diff --git a/core/integration/tests/server/scenarios/mod.rs b/core/integration/tests/server/scenarios/mod.rs
index 2af765e8d..ed2040e53 100644
--- a/core/integration/tests/server/scenarios/mod.rs
+++ b/core/integration/tests/server/scenarios/mod.rs
@@ -25,17 +25,14 @@ pub mod bench_scenario;
pub mod concurrent_produce_consume_scenario;
#[cfg(not(feature = "vsr"))]
pub mod concurrent_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod consumer_group_auto_commit_reconnection_scenario;
-#[cfg(not(feature = "vsr"))]
+// Only consumed by `cg_vsr` (server-ng); legacy `cg.rs` has no equivalent.
+#[cfg(feature = "vsr")]
+pub mod consumer_group_duplicate_name_create_scenario;
pub mod consumer_group_join_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod consumer_group_new_messages_after_restart_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod consumer_group_offset_cleanup_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod consumer_group_with_multiple_clients_polling_messages_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod consumer_group_with_single_client_polling_messages_scenario;
pub mod consumer_timestamp_polling_scenario;
pub mod create_message_payload;
@@ -69,7 +66,6 @@ pub mod segment_rotation_race_scenario;
pub mod single_message_per_batch_scenario;
#[cfg(not(feature = "vsr"))]
pub mod snapshot_scenario;
-#[cfg(not(feature = "vsr"))]
pub mod stale_client_consumer_group_scenario;
#[cfg(not(feature = "vsr"))]
pub mod stream_size_validation_scenario;
@@ -94,7 +90,6 @@ const POLL_RETRY_INTERVAL: Duration = Duration::from_millis(100);
const STREAM_NAME: &str = "test-stream";
const TOPIC_NAME: &str = "test-topic";
const PARTITIONS_COUNT: u32 = 3;
-#[cfg(not(feature = "vsr"))]
const CONSUMER_GROUP_NAME: &str = "test-consumer-group";
const USERNAME_1: &str = "user1";
const USERNAME_2: &str = "user2";
@@ -147,7 +142,6 @@ async fn create_client(harness: &TestHarness) -> IggyClient {
.expect("Failed to create new client")
}
-#[cfg(not(feature = "vsr"))]
async fn get_consumer_group(client: &IggyClient) -> ConsumerGroupDetails {
client
.get_consumer_group(
@@ -160,7 +154,6 @@ async fn get_consumer_group(client: &IggyClient) -> ConsumerGroupDetails {
.expect("Failed to get consumer group")
}
-#[cfg(not(feature = "vsr"))]
async fn join_consumer_group(client: &IggyClient) {
client
.join_consumer_group(
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index bbac9d2d7..47633d47f 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::MuxStateMachine;
-use crate::stm::consumer_group::ConsumerGroups;
+use crate::stm::consumer_group::CompleteConsumerGroupRevocationRequest;
use crate::stm::snapshot::{FillSnapshot, MetadataSnapshot, Snapshot, SnapshotError};
use crate::stm::stream::Streams;
use crate::stm::user::{DeletePersonalAccessTokenRequest, Users};
@@ -31,6 +31,7 @@ use consensus::{
replicate_preflight, replicate_to_next_in_chain, request_preflight,
send_prepare_ok as send_prepare_ok_common,
};
+use iggy_binary_protocol::WireIdentifier;
use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment;
use iggy_binary_protocol::requests::partitions::CreatePartitionsRequest as WireCreatePartitionsRequest;
use iggy_binary_protocol::requests::partitions::CreatePartitionsWithAssignmentsRequest as PersistedCreatePartitionsRequest;
@@ -63,11 +64,9 @@ pub trait StreamsFrontend {
fn users(&self) -> &Users;
#[must_use]
fn streams(&self) -> &Streams;
- #[must_use]
- fn consumer_groups(&self) -> &ConsumerGroups;
}
-impl StreamsFrontend for MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)> {
+impl StreamsFrontend for MuxStateMachine<variadic!(Users, Streams)> {
fn users(&self) -> &Users {
&self.inner().0
}
@@ -75,10 +74,6 @@ impl StreamsFrontend for MuxStateMachine<variadic!(Users, Streams, ConsumerGroup
fn streams(&self) -> &Streams {
&self.inner().1.0
}
-
- fn consumer_groups(&self) -> &ConsumerGroups {
- &self.inner().1.1.0
- }
}
#[derive(Debug, Clone)]
@@ -783,6 +778,13 @@ where
self.client_table
.borrow_mut()
.remove_client(prepare_header.client);
+ // Drop the disconnected client from every consumer group it
+ // joined and rebalance. Deterministic side-effect of the
+ // Logout commit, applied identically on every replica.
+ self.mux_stm.streams().remove_consumer_group_member(
+ prepare_header.client,
+ iggy_common::IggyTimestamp::from(prepare_header.timestamp),
+ );
reply
} else {
// Normal op: apply SM, commit_reply.
@@ -1079,6 +1081,94 @@ where
}
}
+ /// Submit a server-originated `CompleteConsumerGroupRevocation` through the
+ /// metadata consensus group (shard 0). The partition reconciler calls this
+ /// to complete a cooperative revocation once the source has drained the
+ /// partition (or it timed out).
+ ///
+ /// Unlike a client op there is no session: a reserved internal client id
+ /// (never coordinator-minted) carries it, `request_preflight` is skipped
+ /// (server-originated), and the normal-op commit path skips reply-caching
+ /// when the client has no session. The op is internal (not client-allowed),
+ /// so it bypasses `prepare_request` and projects directly.
+ ///
+ /// # Errors
+ /// `NotPrimary` / `NotCaughtUp` when this node cannot accept the prepare,
+ /// `InProgress` / `PipelineFull` on pipeline pressure (the reconciler
+ /// retries next tick; completion is idempotent), `Canceled` if the pending
+ /// prepare was canceled before commit.
+ ///
+ /// # Panics
+ /// On a shard without consensus (only shard 0 owns the metadata consensus
+ /// group); callers must route here only on shard 0.
+ #[allow(clippy::future_not_send)]
+ pub async fn submit_complete_revocation_in_process(
+ &self,
+ stream_id: u32,
+ topic_id: u32,
+ group_id: u64,
+ source_client_id: u128,
+ partition_id: u32,
+ ) -> Result<u64, MetadataSubmitError> {
+ const INTERNAL_REQUEST_ID: u64 = u64::MAX;
+ // Reserved internal client id, distinct per (group, partition) target.
+ // The high 64 bits are all-ones -- never coordinator-minted (those carry
+ // a small home-shard number in the top bits) -- and the low 64 bits pack
+ // (group_id, partition_id). Distinct ids matter: the pipeline dedups by
+ // client id, so a shared id would cap internal completions at one
+ // in-flight cluster-wide and drain a wide rebalance one per consensus
+ // round-trip. Per-target ids let completions for different partitions
+ // pipeline concurrently while still deduping a retry of the same target.
+ let internal_client_id: u128 =
+ (u128::from(u64::MAX) << 64) | (u128::from(group_id) << 32) | u128::from(partition_id);
+
+ let consensus = self
+ .consensus
+ .as_ref()
+ .expect("submit_complete_revocation_in_process: consensus only exists on shard 0");
+
+ if !is_caught_up_primary(consensus) {
+ return Err(
+ if consensus.is_primary() && consensus.is_normal() && !consensus.is_syncing() {
+ MetadataSubmitError::NotCaughtUp
+ } else {
+ MetadataSubmitError::NotPrimary
+ },
+ );
+ }
+ if consensus
+ .pipeline()
+ .borrow()
+ .has_message_from_client(internal_client_id)
+ {
+ return Err(MetadataSubmitError::InProgress);
+ }
+ if consensus.pipeline().borrow().is_full() {
+ return Err(MetadataSubmitError::PipelineFull);
+ }
+
+ let request = CompleteConsumerGroupRevocationRequest {
+ stream_id: WireIdentifier::numeric(stream_id),
+ topic_id: WireIdentifier::numeric(topic_id),
+ group_id,
+ source_client_id,
+ partition_id,
+ };
+ let body = request.to_bytes();
+ let message = build_complete_revocation_request_message(
+ consensus,
+ internal_client_id,
+ INTERNAL_REQUEST_ID,
+ &body,
+ );
+ let prepare = message.project(consensus);
+
+ match self.dispatch_prepare_and_await(consensus, prepare).await {
+ Ok(reply) => Ok(reply.header().commit),
+ Err(Canceled) => Err(MetadataSubmitError::Canceled),
+ }
+ }
+
/// `true` when this node is the caught-up primary of the metadata
/// consensus group. Gates leader-only maintenance (the PAT cleaner)
/// off backups and lagging primaries.
@@ -1567,6 +1657,12 @@ where
.commit_register(header.client, reply, in_flight);
} else if header.operation == Operation::Logout {
self.client_table.borrow_mut().remove_client(header.client);
+ // Mirror the on_ack path: drop the disconnected client from
+ // every consumer group it joined and rebalance.
+ self.mux_stm.streams().remove_consumer_group_member(
+ header.client,
+ iggy_common::IggyTimestamp::from(header.timestamp),
+ );
} else {
// Normal op: apply SM, commit_reply.
let response = self.mux_stm.update(prepare).unwrap_or_else(|err| {
@@ -1716,6 +1812,44 @@ where
msg
}
+fn build_complete_revocation_request_message<B, P>(
+ consensus: &VsrConsensus<B, P>,
+ client_id: u128,
+ request: u64,
+ body: &[u8],
+) -> Message<RequestHeader>
+where
+ B: MessageBus,
+ P: Pipeline<Entry = PipelineEntry>,
+{
+ let header_size = size_of::<RequestHeader>();
+ let total = header_size + body.len();
+ let mut msg = Message::<RequestHeader>::new(total);
+ {
+ let slice = msg.as_mut_slice();
+ slice[header_size..total].copy_from_slice(body);
+ let header =
+ bytemuck::checked::try_from_bytes_mut::<RequestHeader>(&mut slice[..header_size])
+ .expect("zeroed bytes are a valid RequestHeader");
+ *header = RequestHeader {
+ command: Command2::Request,
+ operation: Operation::CompleteConsumerGroupRevocation,
+ size: u32::try_from(total).expect("request size fits u32"),
+ cluster: consensus.cluster(),
+ view: consensus.view(),
+ release: 0,
+ client: client_id,
+ // `validate()` requires session/request > 0 for non-register ops;
+ // there is no real session (the commit path skips reply-caching).
+ session: 1,
+ request,
+ namespace: server_common::sharding::METADATA_CONSENSUS_NAMESPACE,
+ ..RequestHeader::default()
+ };
+ }
+ msg
+}
+
fn build_prepare_message<B, P>(
consensus: &VsrConsensus<B, P>,
request: &RequestHeader,
@@ -1771,14 +1905,13 @@ where
#[cfg(test)]
mod tests {
use super::*;
- use crate::stm::consumer_group::ConsumerGroups;
use crate::stm::stream::Streams;
use crate::stm::user::Users;
use iggy_common::variadic;
use std::cell::RefCell;
use std::rc::Rc;
- type TestMux = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+ type TestMux = MuxStateMachine<variadic!(Users, Streams)>;
/// Build a peer-shard-style `IggyMetadata` with `consensus`,
/// `journal`, and `snapshot` all `None`. Enough to test the
diff --git a/core/metadata/src/stm/consumer_group.rs b/core/metadata/src/stm/consumer_group.rs
index b922fbeb5..b637a60df 100644
--- a/core/metadata/src/stm/consumer_group.rs
+++ b/core/metadata/src/stm/consumer_group.rs
@@ -15,14 +15,26 @@
// specific language governing permissions and limitations
// under the License.
+//! Consumer groups, co-located inside the topic node of the Streams STM.
+//!
+//! Groups belong to a topic, so they live on [`crate::stm::stream::Topic`]:
+//! deleting a stream/topic drops its groups for free (the topic's collections
+//! are dropped with it), and group operations are applied by the Streams STM.
+//!
+//! Group ids are **monotonic per topic and never reused** (`next_consumer_group_id`).
+//! The consumer-group offset (on the partition plane) is keyed by group id, so a
+//! never-reused id guarantees a new group can never inherit a deleted group's
+//! offset -- making a metadata->offset purge unnecessary for correctness.
+
use crate::stm::StateHandler;
-use crate::stm::snapshot::Snapshotable;
-use crate::{collect_handlers, define_state, impl_fill_restore};
+use crate::stm::stream::StreamsInner;
use bytes::Bytes;
-use ahash::AHashMap;
+use bytes::{BufMut, BytesMut};
use iggy_binary_protocol::WireIdentifier;
-use iggy_binary_protocol::codec::WireEncode;
+use iggy_binary_protocol::codec::{
+ WireDecode, WireEncode, capped_capacity, read_u32_le, read_u64_le, read_u128_le,
+};
use iggy_binary_protocol::requests::consumer_groups::{
CreateConsumerGroupRequest, DeleteConsumerGroupRequest,
};
@@ -31,63 +43,122 @@ use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::Consum
use serde::{Deserialize, Serialize};
use slab::Slab;
use std::sync::Arc;
-use std::sync::atomic::{AtomicUsize, Ordering};
+
+/// A partition being cooperatively handed off from this member to `target`.
+///
+/// The source keeps polling/committing it until the handoff completes (the
+/// consumer drains what it polled, or the revocation times out), so an
+/// in-flight batch is never lost mid-rebalance.
+#[derive(Debug, Clone)]
+pub struct PendingRevocation {
+ pub partition_id: usize,
+ /// Slab key of the member the partition moves to once completed.
+ pub target_member: usize,
+ /// Monotonic micros when the revocation was created, for timeout.
+ pub created_at: u64,
+}
#[derive(Debug, Clone)]
pub struct ConsumerGroupMember {
pub id: usize,
- pub client_id: u32,
+ pub client_id: u128,
pub partitions: Vec<usize>,
- pub partition_index: Arc<AtomicUsize>,
+ /// Partitions marked for cooperative handoff away from this member but not
+ /// yet moved. The member still owns them (poll + commit) until completion.
+ pub pending_revocations: Vec<PendingRevocation>,
}
impl ConsumerGroupMember {
#[must_use]
- pub fn new(id: usize, client_id: u32) -> Self {
+ pub const fn new(id: usize, client_id: u128) -> Self {
Self {
id,
client_id,
partitions: Vec::new(),
- partition_index: Arc::new(AtomicUsize::new(0)),
+ pending_revocations: Vec::new(),
}
}
+
+ /// Partitions this member should poll: owned minus those pending handoff
+ /// (the source stops polling a revoked partition so its consumer can drain
+ /// + commit, but it still owns it for the commit fence until completion).
+ #[must_use]
+ pub fn pollable_partitions(&self) -> Vec<usize> {
+ self.partitions
+ .iter()
+ .copied()
+ .filter(|&partition_id| self.is_pollable(partition_id))
+ .collect()
+ }
+
+ /// Whether this member may poll `partition_id`: owned and not pending
+ /// handoff. Alloc-free -- the poll fence hits this once per poll under the
+ /// metadata read lock, so it avoids the `HashSet` + `Vec` that
+ /// `pollable_partitions` builds. `pending_revocations` is tiny (bounded by
+ /// the member's partition count), so the linear scan beats a set.
+ #[must_use]
+ pub fn is_pollable(&self, partition_id: usize) -> bool {
+ self.partitions.contains(&partition_id)
+ && !self
+ .pending_revocations
+ .iter()
+ .any(|revocation| revocation.partition_id == partition_id)
+ }
}
#[derive(Debug, Clone)]
pub struct ConsumerGroup {
- pub id: usize,
+ /// Monotonic id, unique per topic and never reused. Doubles as the
+ /// consumer-group offset key on the partition plane.
+ pub id: u64,
+ /// Bumped on every `rebalance_members`. The client caches the generation it
+ /// synced at; the coordinator fences stale polls and the heartbeat re-syncs
+ /// when it advances.
+ pub generation: u64,
pub name: Arc<str>,
- pub partitions: Vec<usize>,
pub members: Slab<ConsumerGroupMember>,
}
impl ConsumerGroup {
#[must_use]
- pub const fn new(name: Arc<str>) -> Self {
+ pub const fn new(id: u64, name: Arc<str>) -> Self {
Self {
- id: 0,
+ id,
+ generation: 0,
name,
- partitions: Vec::new(),
members: Slab::new(),
}
}
- pub fn rebalance_members(&mut self) {
- let partition_count = self.partitions.len();
- let member_count = self.members.len();
-
- if member_count == 0 || partition_count == 0 {
- return;
- }
+ /// Redistribute the topic's current partitions round-robin across members.
+ /// Reads the live partition set each call, so it reflects repartitions and
+ /// bumps the generation on any membership change.
+ ///
+ /// Deliberately asymmetric with the join path (`rebalance_cooperative`):
+ /// this clears partitions and pending revocations with no cooperative
+ /// drain, so a member's in-flight (polled-but-uncommitted) partitions move
+ /// immediately and the new owner re-reads that range. That redelivery is
+ /// exactly what the join path avoids via pending revocations -- but the
+ /// callers here (leave, remove-member on disconnect, repartition) are cases
+ /// where the departing member cannot drain anyway, so an eager move is
+ /// correct and the redelivery is benign under at-least-once.
+ pub fn rebalance_members(&mut self, partition_ids: &[usize]) {
+ self.generation += 1;
+ let member_count = self.members.len();
let member_keys: Vec<usize> = self.members.iter().map(|(id, _)| id).collect();
for &member_id in &member_keys {
if let Some(member) = self.members.get_mut(member_id) {
member.partitions.clear();
+ member.pending_revocations.clear();
}
}
- for (i, &partition_id) in self.partitions.iter().enumerate() {
+ if member_count == 0 {
+ return;
+ }
+
+ for (i, &partition_id) in partition_ids.iter().enumerate() {
let target = i % member_count;
if let Some(&member_id) = member_keys.get(target)
&& let Some(member) = self.members.get_mut(member_id)
@@ -96,116 +167,358 @@ impl ConsumerGroup {
}
}
}
-}
-define_state! {
- ConsumerGroups {
- name_index: AHashMap<Arc<str>, usize>,
- topic_index: AHashMap<(usize, usize), Vec<usize>>,
- topic_name_index: AHashMap<(Arc<str>, Arc<str>), Vec<usize>>,
- items: Slab<ConsumerGroup>,
- }
-}
+ /// Cooperative rebalance (on member join): assign unassigned partitions to
+ /// idle members immediately, and mark each over-assigned member's excess as
+ /// pending revocation toward an idle member -- the partition only moves once
+ /// the source's consumer drains it (see the reconciler) or it times out, so
+ /// no in-flight batch is dropped. `now` is the replicated op timestamp
+ /// (micros), kept identical across replicas for deterministic apply.
+ #[allow(clippy::too_many_lines)]
+ pub fn rebalance_cooperative(
+ &mut self,
+ partition_ids: &[usize],
+ in_flight: &std::collections::HashSet<usize>,
+ now: u64,
+ ) {
+ self.generation += 1;
-collect_handlers! {
- ConsumerGroups {
- CreateConsumerGroup,
- DeleteConsumerGroup,
- }
-}
+ let member_count = self.members.len();
+ if member_count == 0 || partition_ids.is_empty() {
+ return;
+ }
-impl ConsumerGroupsInner {
- fn resolve_group_in_list(
- &self,
- groups_in_topic: &[usize],
- group_id: &WireIdentifier,
- ) -> Option<usize> {
- match group_id {
- WireIdentifier::Numeric(id) => {
- let g_id = *id as usize;
- groups_in_topic.contains(&g_id).then_some(g_id)
+ let mut assigned: std::collections::HashSet<usize> = std::collections::HashSet::new();
+ for (_, member) in &self.members {
+ for &partition_id in &member.partitions {
+ assigned.insert(partition_id);
}
- WireIdentifier::String(name) => groups_in_topic
+ }
+
+ // Step 1: hand unassigned partitions straight to idle members.
+ let unassigned: Vec<usize> = partition_ids
+ .iter()
+ .copied()
+ .filter(|partition_id| !assigned.contains(partition_id))
+ .collect();
+ if !unassigned.is_empty() {
+ let idle: Vec<usize> = self
+ .members
.iter()
- .find(|&&id| {
- self.items
- .get(id)
- .is_some_and(|g| g.name.as_ref() == name.as_str())
- })
- .copied(),
+ .filter(|(_, member)| member.partitions.is_empty())
+ .map(|(slab_id, _)| slab_id)
+ .collect();
+ if !idle.is_empty() {
+ for (i, partition_id) in unassigned.into_iter().enumerate() {
+ let slab_id = idle[i % idle.len()];
+ if let Some(member) = self.members.get_mut(slab_id) {
+ member.partitions.push(partition_id);
+ }
+ }
+ }
}
- }
- fn resolve_consumer_group_id_by_identifiers(
- &self,
- stream_id: &WireIdentifier,
- topic_id: &WireIdentifier,
- group_id: &WireIdentifier,
- ) -> Option<usize> {
- if let (WireIdentifier::Numeric(s), WireIdentifier::Numeric(t)) = (stream_id, topic_id) {
- let groups_in_topic = self.topic_index.get(&(*s as usize, *t as usize))?;
- return self.resolve_group_in_list(groups_in_topic, group_id);
+ // Step 2: mark excess on over-assigned members as pending revocation.
+ let partition_count = partition_ids.len();
+ let fair_share = partition_count / member_count;
+ let remainder = partition_count % member_count;
+
+ let revocation_targets: std::collections::HashSet<usize> = self
+ .members
+ .iter()
+ .flat_map(|(_, member)| {
+ member
+ .pending_revocations
+ .iter()
+ .map(|revocation| revocation.target_member)
+ })
+ .collect();
+ let idle: Vec<usize> = self
+ .members
+ .iter()
+ .filter(|(slab_id, member)| {
+ member.partitions.is_empty() && !revocation_targets.contains(slab_id)
+ })
+ .map(|(slab_id, _)| slab_id)
+ .collect();
+ if idle.is_empty() {
+ return;
}
- if let (WireIdentifier::String(s), WireIdentifier::String(t)) = (stream_id, topic_id) {
- let key = (Arc::from(s.as_str()), Arc::from(t.as_str()));
- let groups_in_topic = self.topic_name_index.get(&key)?;
- return self.resolve_group_in_list(groups_in_topic, group_id);
+ // Pass 1: collect excess (round-robin into idle members in pass 2 so a
+ // single over-assigned member's excess spreads evenly, not all to one).
+ let member_keys: Vec<usize> = self.members.iter().map(|(slab_id, _)| slab_id).collect();
+ let mut members_with_remainder = remainder;
+ let mut all_excess: Vec<(usize, usize)> = Vec::new();
+ for &slab_id in &member_keys {
+ let Some(member) = self.members.get(slab_id) else {
+ continue;
+ };
+ let pending: std::collections::HashSet<usize> = member
+ .pending_revocations
+ .iter()
+ .map(|revocation| revocation.partition_id)
+ .collect();
+ let effective: Vec<usize> = member
+ .partitions
+ .iter()
+ .copied()
+ .filter(|partition_id| !pending.contains(partition_id))
+ .collect();
+ let max_allowed = if members_with_remainder > 0 {
+ fair_share + 1
+ } else {
+ fair_share
+ };
+ if effective.len() <= max_allowed {
+ if effective.len() > fair_share {
+ members_with_remainder = members_with_remainder.saturating_sub(1);
+ }
+ continue;
+ }
+ let excess_count = effective.len() - max_allowed;
+ if members_with_remainder > 0 && effective.len() > fair_share {
+ members_with_remainder = members_with_remainder.saturating_sub(1);
+ }
+ for partition_id in effective.into_iter().rev().take(excess_count) {
+ all_excess.push((partition_id, slab_id));
+ }
}
- None
+ // Pass 2: distribute the collected excess round-robin over idle members.
+ // An in-flight partition (the source polled but hasn't committed it) is
+ // pending-revoked so the source can drain it first; a never-polled or
+ // already-drained partition has nothing in flight, so it moves now.
+ for (i, (partition_id, source_slab)) in all_excess.into_iter().enumerate() {
+ let target_slab = idle[i % idle.len()];
+ if in_flight.contains(&partition_id) {
+ if let Some(source) = self.members.get_mut(source_slab) {
+ source.pending_revocations.push(PendingRevocation {
+ partition_id,
+ target_member: target_slab,
+ created_at: now,
+ });
+ }
+ } else {
+ if let Some(source) = self.members.get_mut(source_slab) {
+ source.partitions.retain(|&p| p != partition_id);
+ }
+ if let Some(target) = self.members.get_mut(target_slab) {
+ target.partitions.push(partition_id);
+ }
+ }
+ }
}
-}
-impl ConsumerGroups {
+ /// Complete a pending revocation: move `partition_id` from the member with
+ /// `source_client_id` to its recorded target. Bumps the generation so both
+ /// clients re-sync. Returns `false` if no such pending revocation exists
+ /// (already completed / stale) or the target member is gone.
+ pub fn complete_revocation(&mut self, source_client_id: u128, partition_id: usize) -> bool {
+ let Some(source_slab) = self
+ .members
+ .iter()
+ .find(|(_, member)| member.client_id == source_client_id)
+ .map(|(slab_id, _)| slab_id)
+ else {
+ return false;
+ };
+ let Some(source) = self.members.get_mut(source_slab) else {
+ return false;
+ };
+ let Some(pos) = source
+ .pending_revocations
+ .iter()
+ .position(|revocation| revocation.partition_id == partition_id)
+ else {
+ return false;
+ };
+ let revocation = source.pending_revocations.remove(pos);
+ source.partitions.retain(|&p| p != partition_id);
+ if let Some(target) = self.members.get_mut(revocation.target_member) {
+ target.partitions.push(partition_id);
+ self.generation += 1;
+ } else {
+ // Target gone: full rebalance recovers a consistent assignment.
+ let mut all: std::collections::HashSet<usize> = std::collections::HashSet::new();
+ for (_, member) in &self.members {
+ for &p in &member.partitions {
+ all.insert(p);
+ }
+ }
+ all.insert(partition_id);
+ let mut partition_ids: Vec<usize> = all.into_iter().collect();
+ partition_ids.sort_unstable();
+ self.rebalance_members(&partition_ids);
+ }
+ true
+ }
+
+ /// `(source_client_id, partition_id, created_at)` for every pending
+ /// revocation in this group, for the reconciler's completion check.
#[must_use]
- pub fn read<F, R>(&self, f: F) -> R
- where
- F: FnOnce(&ConsumerGroupsInner) -> R,
- {
- self.inner.read(f)
+ pub fn pending_revocations(&self) -> Vec<(u128, usize, u64)> {
+ self.members
+ .iter()
+ .flat_map(|(_, member)| {
+ member.pending_revocations.iter().map(move |revocation| {
+ (
+ member.client_id,
+ revocation.partition_id,
+ revocation.created_at,
+ )
+ })
+ })
+ .collect()
}
}
-impl StateHandler for CreateConsumerGroupRequest {
- type State = ConsumerGroupsInner;
- #[allow(clippy::cast_possible_truncation)]
- fn apply(
- &self,
- state: &mut ConsumerGroupsInner,
- _timestamp: iggy_common::IggyTimestamp,
- ) -> Bytes {
- let name: Arc<str> = Arc::from(self.name.as_str());
- if state.name_index.contains_key(&name) {
- return Bytes::new();
- }
+/// Replicated `JoinConsumerGroup`, enriched by the primary's home shard.
+///
+/// The wire request carries only identifiers; the apply needs the client id
+/// (which member is joining) and can't read it from the consensus header. The
+/// home shard also gathers `in_flight` -- the group's partitions with
+/// uncommitted polled data -- so the cooperative rebalance pending-revokes only
+/// those and hands off never-polled/drained partitions immediately (the apply
+/// has no partition-plane access to classify them itself).
+#[derive(Debug, Clone)]
+pub struct JoinConsumerGroupRequest {
+ pub stream_id: WireIdentifier,
+ pub topic_id: WireIdentifier,
+ pub group_id: WireIdentifier,
+ pub client_id: u128,
+ pub in_flight: Vec<u32>,
+}
- let group = ConsumerGroup::new(name.clone());
- let id = state.items.insert(group);
- state.items[id].id = id;
+impl WireEncode for JoinConsumerGroupRequest {
+ fn encoded_size(&self) -> usize {
+ self.stream_id.encoded_size()
+ + self.topic_id.encoded_size()
+ + self.group_id.encoded_size()
+ + 16
+ + 4
+ + self.in_flight.len() * 4
+ }
- state.name_index.insert(name, id);
+ fn encode(&self, buf: &mut BytesMut) {
+ self.stream_id.encode(buf);
+ self.topic_id.encode(buf);
+ self.group_id.encode(buf);
+ buf.put_u128_le(self.client_id);
+ // `encoded_size` + the loop below use the true length, so a silent
+ // clamp here would desync the count header from the body. `in_flight`
+ // is bounded by the topic's partition count, so this never fires.
+ buf.put_u32_le(u32::try_from(self.in_flight.len()).expect("in_flight count fits u32"));
+ for partition_id in &self.in_flight {
+ buf.put_u32_le(*partition_id);
+ }
+ }
+}
- if let (WireIdentifier::Numeric(s), WireIdentifier::Numeric(t)) =
- (&self.stream_id, &self.topic_id)
- {
- state
- .topic_index
- .entry((*s as usize, *t as usize))
- .or_default()
- .push(id);
+impl WireDecode for JoinConsumerGroupRequest {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), iggy_binary_protocol::WireError> {
+ let (stream_id, mut pos) = WireIdentifier::decode(buf)?;
+ let (topic_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ let (group_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ let client_id = read_u128_le(buf, pos)?;
+ pos += 16;
+ let count = read_u32_le(buf, pos)? as usize;
+ pos += 4;
+ // Cap against the bytes actually left: `count` is read straight off a
+ // replicated prepare, so an unbounded `with_capacity` would let a
+ // corrupt/bit-rotted count (near u32::MAX) allocate gigabytes and abort
+ // every backup that applies it.
+ let mut in_flight =
+ Vec::with_capacity(capped_capacity(count, buf.len().saturating_sub(pos), 4));
+ for _ in 0..count {
+ in_flight.push(read_u32_le(buf, pos)?);
+ pos += 4;
}
+ Ok((
+ Self {
+ stream_id,
+ topic_id,
+ group_id,
+ client_id,
+ in_flight,
+ },
+ pos,
+ ))
+ }
+}
+
+/// Replicated `LeaveConsumerGroup`, enriched by the primary with the leaving
+/// client's VSR id. The apply removes the member and rebalances.
+#[derive(Debug, Clone)]
+pub struct LeaveConsumerGroupRequest {
+ pub stream_id: WireIdentifier,
+ pub topic_id: WireIdentifier,
+ pub group_id: WireIdentifier,
+ pub client_id: u128,
+}
+
+impl WireEncode for LeaveConsumerGroupRequest {
+ fn encoded_size(&self) -> usize {
+ self.stream_id.encoded_size()
+ + self.topic_id.encoded_size()
+ + self.group_id.encoded_size()
+ + 16
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ self.stream_id.encode(buf);
+ self.topic_id.encode(buf);
+ self.group_id.encode(buf);
+ buf.put_u128_le(self.client_id);
+ }
+}
+
+impl WireDecode for LeaveConsumerGroupRequest {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), iggy_binary_protocol::WireError> {
+ let (stream_id, mut pos) = WireIdentifier::decode(buf)?;
+ let (topic_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ let (group_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ let client_id = read_u128_le(buf, pos)?;
+ pos += 16;
+ Ok((
+ Self {
+ stream_id,
+ topic_id,
+ group_id,
+ client_id,
+ },
+ pos,
+ ))
+ }
+}
- if let (WireIdentifier::String(s), WireIdentifier::String(t)) =
- (&self.stream_id, &self.topic_id)
- {
- let key = (Arc::from(s.as_str()), Arc::from(t.as_str()));
- state.topic_name_index.entry(key).or_default().push(id);
+impl StateHandler for CreateConsumerGroupRequest {
+ type State = StreamsInner;
+ #[allow(clippy::cast_possible_truncation)]
+ fn apply(&self, state: &mut StreamsInner, _timestamp: iggy_common::IggyTimestamp) -> Bytes {
+ let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id) else {
+ return Bytes::new();
+ };
+ let name: Arc<str> = Arc::from(self.name.as_str());
+ // Per-(stream,topic) name uniqueness. A same-name group already exists:
+ // do NOT supersede it -- removing it would drop a live group along with
+ // its members and their assignments, and the ejected members would
+ // never recover. Leave the existing group untouched and reply empty,
+ // mirroring `CreateStream`/`CreateTopic` on a duplicate name.
+ if topic.consumer_group_index.contains_key(&name) {
+ return Bytes::new();
}
+ let id = topic.next_consumer_group_id;
+ topic.next_consumer_group_id += 1;
+ topic
+ .consumer_groups
+ .insert(id, ConsumerGroup::new(id, name.clone()));
+ topic.consumer_group_index.insert(name, id);
- // Reply body: the SDK `create_consumer_group` decodes a
- // `ConsumerGroupDetailsResponse`. A freshly created group has no
- // assigned partitions and no joined members yet.
ConsumerGroupDetailsResponse {
group: ConsumerGroupResponse {
id: id as u32,
@@ -220,177 +533,335 @@ impl StateHandler for CreateConsumerGroupRequest {
}
impl StateHandler for DeleteConsumerGroupRequest {
- type State = ConsumerGroupsInner;
- fn apply(
- &self,
- state: &mut ConsumerGroupsInner,
- _timestamp: iggy_common::IggyTimestamp,
- ) -> Bytes {
- let Some(id) = state.resolve_consumer_group_id_by_identifiers(
- &self.stream_id,
- &self.topic_id,
- &self.group_id,
- ) else {
- return Bytes::new();
+ type State = StreamsInner;
+ fn apply(&self, state: &mut StreamsInner, _timestamp: iggy_common::IggyTimestamp) -> Bytes {
+ let removed = {
+ let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id) else {
+ return Bytes::new();
+ };
+ if let Some(group_id) = topic.resolve_group_id(&self.group_id)
+ && let Some(group) = topic.consumer_groups.remove(&group_id)
+ {
+ topic.consumer_group_index.remove(&group.name);
+ true
+ } else {
+ false
+ }
};
-
- let group = state.items.remove(id);
- state.name_index.remove(&group.name);
-
- if let (WireIdentifier::Numeric(s), WireIdentifier::Numeric(t)) =
- (&self.stream_id, &self.topic_id)
- && let Some(vec) = state.topic_index.get_mut(&(*s as usize, *t as usize))
- {
- vec.retain(|&x| x != id);
+ // Bump the partition-shaping revision so the reconciler's fast-skip
+ // doesn't pass over the delete: it reclaims the group's leftover
+ // offsets on the topic's surviving partitions.
+ if removed {
+ state.revision = state.revision.wrapping_add(1);
+ // The dropped group may have held pending revocations.
+ state.recompute_pending_revocations_count();
}
+ Bytes::new()
+ }
+}
- if let (WireIdentifier::String(s), WireIdentifier::String(t)) =
- (&self.stream_id, &self.topic_id)
- {
- let key = (Arc::from(s.as_str()), Arc::from(t.as_str()));
- if let Some(vec) = state.topic_name_index.get_mut(&key) {
- vec.retain(|&x| x != id);
- }
+impl StateHandler for JoinConsumerGroupRequest {
+ type State = StreamsInner;
+ fn apply(&self, state: &mut StreamsInner, timestamp: iggy_common::IggyTimestamp) -> Bytes {
+ let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id) else {
+ return Bytes::new();
+ };
+ let Some(group_id) = topic.resolve_group_id(&self.group_id) else {
+ return Bytes::new();
+ };
+ // Snapshot the live partition ids before taking a mutable borrow of the
+ // group (both borrow the topic).
+ let partition_ids: Vec<usize> = topic.partitions.iter().map(|p| p.id).collect();
+ let Some(group) = topic.consumer_groups.get_mut(&group_id) else {
+ return Bytes::new();
+ };
+ // Idempotent: a re-join from the same client keeps its membership.
+ let already = group
+ .members
+ .iter()
+ .any(|(_, m)| m.client_id == self.client_id);
+ if already {
+ return Bytes::new();
}
+ let member_key = group
+ .members
+ .insert(ConsumerGroupMember::new(0, self.client_id));
+ group.members[member_key].id = member_key;
+ // Cooperative handoff using the home-shard-gathered `in_flight` set:
+ // never-polled/drained excess moves to the new member immediately (so a
+ // fresh group distributes synchronously at join), while partitions with
+ // uncommitted in-flight data are pending-revoked for the reconciler to
+ // complete once the source drains them (or on timeout).
+ let in_flight: std::collections::HashSet<usize> =
+ self.in_flight.iter().map(|&p| p as usize).collect();
+ group.rebalance_cooperative(&partition_ids, &in_flight, timestamp.as_micros());
+ state.recompute_pending_revocations_count();
Bytes::new()
}
}
-/// Consumer group member snapshot representation for serialization.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct ConsumerGroupMemberSnapshot {
- pub id: usize,
- pub client_id: u32,
- pub partitions: Vec<usize>,
- pub partition_index: usize,
+impl StateHandler for LeaveConsumerGroupRequest {
+ type State = StreamsInner;
+ fn apply(&self, state: &mut StreamsInner, _timestamp: iggy_common::IggyTimestamp) -> Bytes {
+ let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id) else {
+ return Bytes::new();
+ };
+ let Some(group_id) = topic.resolve_group_id(&self.group_id) else {
+ return Bytes::new();
+ };
+ let partition_ids: Vec<usize> = topic.partitions.iter().map(|p| p.id).collect();
+ let Some(group) = topic.consumer_groups.get_mut(&group_id) else {
+ return Bytes::new();
+ };
+ let member_key = group
+ .members
+ .iter()
+ .find(|(_, m)| m.client_id == self.client_id)
+ .map(|(key, _)| key);
+ if let Some(key) = member_key {
+ group.members.remove(key);
+ group.rebalance_members(&partition_ids);
+ state.recompute_pending_revocations_count();
+ }
+ Bytes::new()
+ }
}
-/// Consumer group snapshot representation for serialization.
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct ConsumerGroupSnapshot {
- pub id: usize,
- pub name: String,
- pub partitions: Vec<usize>,
- pub members: Vec<(usize, ConsumerGroupMemberSnapshot)>,
+/// Server-originated disconnect cleanup.
+///
+/// Drops a disconnected client from every consumer group it joined. Carries
+/// only the VSR client id; applied as a side-effect of the `Logout` commit (the
+/// apply can't read the client id from the consensus header).
+#[derive(Debug, Clone)]
+pub struct RemoveConsumerGroupMemberRequest {
+ pub client_id: u128,
}
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct ConsumerGroupsSnapshot {
- pub items: Vec<(usize, ConsumerGroupSnapshot)>,
- pub topic_index: Vec<((usize, usize), Vec<usize>)>,
- pub topic_name_index: Vec<((String, String), Vec<usize>)>,
+impl WireEncode for RemoveConsumerGroupMemberRequest {
+ fn encoded_size(&self) -> usize {
+ 16
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ buf.put_u128_le(self.client_id);
+ }
}
-impl Snapshotable for ConsumerGroups {
- type Snapshot = ConsumerGroupsSnapshot;
+impl WireDecode for RemoveConsumerGroupMemberRequest {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), iggy_binary_protocol::WireError> {
+ let client_id = read_u128_le(buf, 0)?;
+ Ok((Self { client_id }, 16))
+ }
+}
- fn to_snapshot(&self) -> Self::Snapshot {
- self.inner.read(|inner| {
- let items: Vec<(usize, ConsumerGroupSnapshot)> = inner
- .items
- .iter()
- .map(|(group_id, group)| {
- let members: Vec<(usize, ConsumerGroupMemberSnapshot)> = group
+impl StateHandler for RemoveConsumerGroupMemberRequest {
+ type State = StreamsInner;
+ fn apply(&self, state: &mut StreamsInner, _timestamp: iggy_common::IggyTimestamp) -> Bytes {
+ let mut any_removed = false;
+ for (_, stream) in &mut state.items {
+ for (_, topic) in &mut stream.topics {
+ for group in topic.consumer_groups.values_mut() {
+ let member_key = group
.members
.iter()
- .map(|(member_id, member)| {
- (
- member_id,
- ConsumerGroupMemberSnapshot {
- id: member.id,
- client_id: member.client_id,
- partitions: member.partitions.clone(),
- partition_index: member.partition_index.load(Ordering::Relaxed),
- },
- )
- })
- .collect();
+ .find(|(_, m)| m.client_id == self.client_id)
+ .map(|(key, _)| key);
+ // Collect the partition set only when this group actually
+ // holds the member. This runs on every logout inside the
+ // no-await commit critical section on every replica, so the
+ // alloc stays out of the common (no-match) path.
+ if let Some(key) = member_key {
+ let partition_ids: Vec<usize> =
+ topic.partitions.iter().map(|p| p.id).collect();
+ group.members.remove(key);
+ group.rebalance_members(&partition_ids);
+ any_removed = true;
+ }
+ }
+ }
+ }
+ // Only perturb the reconciler when a membership actually changed.
+ if any_removed {
+ state.revision = state.revision.wrapping_add(1);
+ state.recompute_pending_revocations_count();
+ }
+ Bytes::new()
+ }
+}
- (
- group_id,
- ConsumerGroupSnapshot {
- id: group.id,
- name: group.name.to_string(),
- partitions: group.partitions.clone(),
- members,
- },
- )
- })
- .collect();
+/// Reconciler-originated completion of a pending cooperative revocation.
+///
+/// Moves `partition_id` from the member `source_client_id` to its recorded
+/// target, once the source consumer has drained the partition (committed up to
+/// what it polled) or the revocation timed out.
+#[derive(Debug, Clone)]
+pub struct CompleteConsumerGroupRevocationRequest {
+ pub stream_id: WireIdentifier,
+ pub topic_id: WireIdentifier,
+ pub group_id: u64,
+ pub source_client_id: u128,
+ pub partition_id: u32,
+}
- let topic_index: Vec<((usize, usize), Vec<usize>)> = inner
- .topic_index
- .iter()
- .map(|(&k, v)| (k, v.clone()))
- .collect();
+impl WireEncode for CompleteConsumerGroupRevocationRequest {
+ fn encoded_size(&self) -> usize {
+ self.stream_id.encoded_size() + self.topic_id.encoded_size() + 8 + 16 + 4
+ }
- let topic_name_index: Vec<((String, String), Vec<usize>)> = inner
- .topic_name_index
- .iter()
- .map(|((s, t), v)| ((s.to_string(), t.to_string()), v.clone()))
- .collect();
+ fn encode(&self, buf: &mut BytesMut) {
+ self.stream_id.encode(buf);
+ self.topic_id.encode(buf);
+ buf.put_u64_le(self.group_id);
+ buf.put_u128_le(self.source_client_id);
+ buf.put_u32_le(self.partition_id);
+ }
+}
- ConsumerGroupsSnapshot {
- items,
- topic_index,
- topic_name_index,
- }
- })
+impl WireDecode for CompleteConsumerGroupRevocationRequest {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), iggy_binary_protocol::WireError> {
+ let (stream_id, mut pos) = WireIdentifier::decode(buf)?;
+ let (topic_id, n) = WireIdentifier::decode(&buf[pos..])?;
+ pos += n;
+ let group_id = read_u64_le(buf, pos)?;
+ pos += 8;
+ let source_client_id = read_u128_le(buf, pos)?;
+ pos += 16;
+ let partition_id = read_u32_le(buf, pos)?;
+ pos += 4;
+ Ok((
+ Self {
+ stream_id,
+ topic_id,
+ group_id,
+ source_client_id,
+ partition_id,
+ },
+ pos,
+ ))
}
+}
- fn from_snapshot(
- snapshot: Self::Snapshot,
- ) -> Result<Self, crate::stm::snapshot::SnapshotError> {
- let mut name_index: AHashMap<Arc<str>, usize> = AHashMap::new();
- let mut group_entries: Vec<(usize, ConsumerGroup)> = Vec::new();
+impl StateHandler for CompleteConsumerGroupRevocationRequest {
+ type State = StreamsInner;
+ fn apply(&self, state: &mut StreamsInner, _timestamp: iggy_common::IggyTimestamp) -> Bytes {
+ let Some(topic) = state.topic_mut(&self.stream_id, &self.topic_id) else {
+ return Bytes::new();
+ };
+ let completed = topic
+ .consumer_groups
+ .get_mut(&self.group_id)
+ .is_some_and(|group| {
+ group.complete_revocation(self.source_client_id, self.partition_id as usize)
+ });
+ if completed {
+ state.revision = state.revision.wrapping_add(1);
+ state.recompute_pending_revocations_count();
+ }
+ Bytes::new()
+ }
+}
- for (slab_key, group_snap) in snapshot.items {
- let members: Slab<ConsumerGroupMember> = group_snap
- .members
- .into_iter()
- .map(|(member_key, member_snap)| {
- let member = ConsumerGroupMember {
- id: member_snap.id,
- client_id: member_snap.client_id,
- partitions: member_snap.partitions,
- partition_index: Arc::new(AtomicUsize::new(member_snap.partition_index)),
- };
- (member_key, member)
- })
- .collect();
+/// Consumer group member snapshot representation for serialization.
+///
+/// Snapshots are encoded by `rmp_serde::to_vec` as **positional arrays** (no
+/// field-name map), so `#[serde(default)]` only fills a missing **trailing**
+/// element -- a new field is back/forward-compatible only if appended last. A
+/// field inserted mid-struct shifts every later position and corrupts decode.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ConsumerGroupMemberSnapshot {
+ pub id: usize,
+ pub client_id: u128,
+ pub partitions: Vec<usize>,
+ /// `(partition_id, target_member_slab, created_at_micros)` per pending
+ /// cooperative revocation. Trailing + `serde(default)` so a snapshot that
+ /// predates this field still decodes (the default fills the absent
+ /// trailing element).
+ #[serde(default)]
+ pub pending_revocations: Vec<(usize, usize, u64)>,
+}
- let group_name: Arc<str> = Arc::from(group_snap.name.as_str());
- let group = ConsumerGroup {
- id: group_snap.id,
- name: group_name.clone(),
- partitions: group_snap.partitions,
- members,
- };
+/// Consumer group snapshot representation for serialization (nested under the
+/// topic snapshot). Positional array, same trailing-only rule as
+/// [`ConsumerGroupMemberSnapshot`].
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ConsumerGroupSnapshot {
+ pub id: u64,
+ // No `serde(default)`: `generation` is not the trailing field, so a default
+ // could not fill it positionally anyway, and every consumer-group snapshot
+ // (new with co-located groups) always writes it.
+ pub generation: u64,
+ pub name: String,
+ pub members: Vec<(usize, ConsumerGroupMemberSnapshot)>,
+}
- name_index.insert(group_name, slab_key);
- group_entries.push((slab_key, group));
+impl ConsumerGroupSnapshot {
+ #[must_use]
+ pub fn from_group(group: &ConsumerGroup) -> Self {
+ let members = group
+ .members
+ .iter()
+ .map(|(member_id, member)| {
+ (
+ member_id,
+ ConsumerGroupMemberSnapshot {
+ id: member.id,
+ client_id: member.client_id,
+ partitions: member.partitions.clone(),
+ pending_revocations: member
+ .pending_revocations
+ .iter()
+ .map(|revocation| {
+ (
+ revocation.partition_id,
+ revocation.target_member,
+ revocation.created_at,
+ )
+ })
+ .collect(),
+ },
+ )
+ })
+ .collect();
+ Self {
+ id: group.id,
+ generation: group.generation,
+ name: group.name.to_string(),
+ members,
}
+ }
- let items = group_entries.into_iter().collect();
-
- let topic_index: AHashMap<(usize, usize), Vec<usize>> =
- snapshot.topic_index.into_iter().collect();
-
- let topic_name_index: AHashMap<(Arc<str>, Arc<str>), Vec<usize>> = snapshot
- .topic_name_index
+ #[must_use]
+ pub fn into_group(self) -> ConsumerGroup {
+ let members: Slab<ConsumerGroupMember> = self
+ .members
.into_iter()
- .map(|((s, t), v)| ((Arc::from(s.as_str()), Arc::from(t.as_str())), v))
+ .map(|(member_key, member_snap)| {
+ (
+ member_key,
+ ConsumerGroupMember {
+ id: member_snap.id,
+ client_id: member_snap.client_id,
+ partitions: member_snap.partitions,
+ pending_revocations: member_snap
+ .pending_revocations
+ .into_iter()
+ .map(
+ |(partition_id, target_member, created_at)| PendingRevocation {
+ partition_id,
+ target_member,
+ created_at,
+ },
+ )
+ .collect(),
+ },
+ )
+ })
.collect();
-
- let inner = ConsumerGroupsInner {
- name_index,
- topic_index,
- topic_name_index,
- items,
- last_result: None,
- };
- Ok(inner.into())
+ ConsumerGroup {
+ id: self.id,
+ generation: self.generation,
+ name: Arc::from(self.name.as_str()),
+ members,
+ }
}
}
-
-impl_fill_restore!(ConsumerGroups, consumer_groups);
diff --git a/core/metadata/src/stm/mux.rs b/core/metadata/src/stm/mux.rs
index aa72f13d0..ade55ca03 100644
--- a/core/metadata/src/stm/mux.rs
+++ b/core/metadata/src/stm/mux.rs
@@ -33,8 +33,8 @@ use crate::stm::{State, StateMachine};
/// state.
///
/// The variadic tuple impl below recurses over the state list, so a
-/// `MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>` ends up
-/// with `Bundle = (Users::Bundle, (Streams::Bundle, (ConsumerGroups::Bundle, ())))`.
+/// `MuxStateMachine<variadic!(Users, Streams)>` ends up
+/// with `Bundle = (Users::Bundle, (Streams::Bundle, ()))`.
pub trait WithFactory: Sized {
type Bundle: Clone + Send + Sync;
fn factory_bundle(&self) -> Self::Bundle;
@@ -210,7 +210,6 @@ where
#[allow(unused_imports)]
mod tests {
use super::*;
- use crate::stm::consumer_group::{ConsumerGroups, ConsumerGroupsInner};
use crate::stm::snapshot::{FillSnapshot, MetadataSnapshot, RestoreSnapshot, Snapshotable};
use crate::stm::stream::{Streams, StreamsInner};
use crate::stm::user::{Users, UsersInner};
@@ -235,13 +234,12 @@ mod tests {
#[test]
fn mux_state_machine_snapshot_roundtrip() {
- type MuxTuple = (Users, (Streams, (ConsumerGroups, ())));
+ type MuxTuple = (Users, (Streams, ()));
let users: Users = UsersInner::new().into();
let streams: Streams = StreamsInner::new().into();
- let consumer_groups: ConsumerGroups = ConsumerGroupsInner::new().into();
- let mux = MuxStateMachine::new(variadic!(users, streams, consumer_groups));
+ let mux = MuxStateMachine::new(variadic!(users, streams));
// Fill the typed snapshot
let mut snapshot = MetadataSnapshot::new(12345);
@@ -250,7 +248,6 @@ mod tests {
// Verify all fields are filled
assert!(snapshot.users.is_some());
assert!(snapshot.streams.is_some());
- assert!(snapshot.consumer_groups.is_some());
let restored: MuxStateMachine<MuxTuple> =
MuxStateMachine::restore_snapshot(&snapshot).unwrap();
@@ -260,7 +257,6 @@ mod tests {
restored.fill_snapshot(&mut verify_snapshot).unwrap();
assert!(verify_snapshot.users.is_some());
assert!(verify_snapshot.streams.is_some());
- assert!(verify_snapshot.consumer_groups.is_some());
}
#[test]
@@ -268,14 +264,12 @@ mod tests {
use crate::impls::metadata::IggySnapshot;
use crate::stm::snapshot::Snapshot;
- type MuxTuple = (Users, (Streams, (ConsumerGroups, ())));
+ type MuxTuple = (Users, (Streams, ()));
let users: Users = UsersInner::new().into();
let streams: Streams = StreamsInner::new().into();
- let consumer_groups: ConsumerGroups = ConsumerGroupsInner::new().into();
- let mux: MuxStateMachine<MuxTuple> =
- MuxStateMachine::new(variadic!(users, streams, consumer_groups));
+ let mux: MuxStateMachine<MuxTuple> = MuxStateMachine::new(variadic!(users, streams));
let sequence_number = 12345u64;
let snapshot = IggySnapshot::create(&mux, sequence_number).unwrap();
@@ -294,7 +288,6 @@ mod tests {
// Verify snapshot fields are present
assert!(decoded.snapshot().users.is_some());
assert!(decoded.snapshot().streams.is_some());
- assert!(decoded.snapshot().consumer_groups.is_some());
// Restore MuxStateMachine from the state side (symmetric with fill_snapshot)
let restored: MuxStateMachine<MuxTuple> =
@@ -305,12 +298,11 @@ mod tests {
restored.fill_snapshot(&mut verify_snapshot).unwrap();
assert!(verify_snapshot.users.is_some());
assert!(verify_snapshot.streams.is_some());
- assert!(verify_snapshot.consumer_groups.is_some());
}
#[test]
fn mux_state_machine_default_initializes_empty_state() {
- type MuxTuple = (Users, (Streams, (ConsumerGroups, ())));
+ type MuxTuple = (Users, (Streams, ()));
let mux = MuxStateMachine::<MuxTuple>::default();
@@ -318,6 +310,5 @@ mod tests {
mux.fill_snapshot(&mut verify_snapshot).unwrap();
assert!(verify_snapshot.users.is_some());
assert!(verify_snapshot.streams.is_some());
- assert!(verify_snapshot.consumer_groups.is_some());
}
}
diff --git a/core/metadata/src/stm/snapshot.rs b/core/metadata/src/stm/snapshot.rs
index ea439bffe..5f704ae46 100644
--- a/core/metadata/src/stm/snapshot.rs
+++ b/core/metadata/src/stm/snapshot.rs
@@ -18,7 +18,6 @@
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use std::fmt;
-use crate::stm::consumer_group::ConsumerGroupsSnapshot;
use crate::stm::stream::StreamsSnapshot;
use crate::stm::user::UsersSnapshot;
@@ -115,10 +114,9 @@ pub struct MetadataSnapshot {
pub sequence_number: u64,
/// Users state machine snapshot data.
pub users: Option<UsersSnapshot>,
- /// Streams state machine snapshot data.
+ /// Streams state machine snapshot data (consumer groups are nested inside
+ /// each topic).
pub streams: Option<StreamsSnapshot>,
- /// Consumer groups state machine snapshot data.
- pub consumer_groups: Option<ConsumerGroupsSnapshot>,
}
impl Default for MetadataSnapshot {
@@ -137,7 +135,6 @@ impl MetadataSnapshot {
sequence_number,
users: None,
streams: None,
- consumer_groups: None,
}
}
@@ -309,7 +306,6 @@ mod tests {
assert_eq!(decoded.sequence_number, 42);
assert!(decoded.users.is_none());
assert!(decoded.streams.is_none());
- assert!(decoded.consumer_groups.is_none());
}
#[test]
@@ -351,7 +347,8 @@ mod tests {
created_at: ts,
created_revision: 0,
}],
- round_robin_counter: 0,
+ consumer_groups: Vec::new(),
+ next_consumer_group_id: 1,
},
)],
},
@@ -363,7 +360,6 @@ mod tests {
assert_eq!(decoded.sequence_number, 100);
assert!(decoded.users.is_none());
- assert!(decoded.consumer_groups.is_none());
let streams = decoded.streams.as_ref().unwrap();
assert_eq!(streams.items.len(), 1);
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index c910c4299..bdd34bca8 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -16,12 +16,18 @@
// under the License.
use crate::stm::StateHandler;
+use crate::stm::consumer_group::{
+ CompleteConsumerGroupRevocationRequest, ConsumerGroup, ConsumerGroupSnapshot,
+ JoinConsumerGroupRequest, LeaveConsumerGroupRequest, RemoveConsumerGroupMemberRequest,
+};
use crate::stm::snapshot::Snapshotable;
use crate::{collect_handlers, define_state, impl_fill_restore};
use ahash::AHashMap;
use bytes::{Bytes, BytesMut};
-use iggy_binary_protocol::WireIdentifier;
use iggy_binary_protocol::codec::WireEncode;
+use iggy_binary_protocol::requests::consumer_groups::{
+ CreateConsumerGroupRequest, DeleteConsumerGroupRequest,
+};
use iggy_binary_protocol::requests::partitions::{
CreatePartitionsWithAssignmentsRequest, DeletePartitionsRequest,
};
@@ -32,9 +38,14 @@ use iggy_binary_protocol::requests::topics::{
CreateTopicRequest, CreateTopicWithAssignmentsRequest, DeleteTopicRequest, PurgeTopicRequest,
UpdateTopicRequest,
};
+use iggy_binary_protocol::responses::consumer_groups::consumer_group_response::ConsumerGroupResponse;
+use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::{
+ ConsumerGroupDetailsResponse, ConsumerGroupMemberResponse,
+};
use iggy_binary_protocol::responses::streams::StreamResponse;
use iggy_binary_protocol::responses::streams::get_stream::{GetStreamResponse, TopicHeader};
use iggy_binary_protocol::responses::topics::get_topic::PartitionResponse;
+use iggy_binary_protocol::{WireIdentifier, WireName};
use iggy_common::{
CompressionAlgorithm, IggyExpiry, IggyTimestamp, MaxTopicSize, StreamStats, TopicStats,
};
@@ -94,6 +105,12 @@ pub struct StatsSnapshot {
}
/// Topic snapshot representation for serialization.
+///
+/// Encoded by `rmp_serde::to_vec` as a **positional array**, so `serde(default)`
+/// only fills **trailing** elements absent from an older snapshot. The two
+/// consumer-group fields below are therefore deliberately last: a topic
+/// snapshot written before co-located consumer groups has a shorter array, and
+/// the defaults fill the missing tail. Any future field must also be appended.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TopicSnapshot {
pub id: usize,
@@ -105,7 +122,17 @@ pub struct TopicSnapshot {
pub max_topic_size: MaxTopicSize,
pub stats: StatsSnapshot,
pub partitions: Vec<PartitionSnapshot>,
- pub round_robin_counter: usize,
+ // `round_robin_counter` is intentionally NOT snapshotted. It is a local
+ // load-balancing hint advanced on the `Balanced`-send read path (outside
+ // the replicated apply), so each replica's value drifts independently;
+ // persisting it would make the snapshot diverge per replica. Restored to 0.
+ //
+ // The two consumer-group fields are trailing so the `serde(default)` above
+ // actually works for snapshots predating co-located consumer groups.
+ #[serde(default)]
+ pub consumer_groups: Vec<(u64, ConsumerGroupSnapshot)>,
+ #[serde(default)]
+ pub next_consumer_group_id: u64,
}
#[derive(Debug, Clone)]
@@ -121,6 +148,21 @@ pub struct Topic {
pub stats: Arc<TopicStats>,
pub partitions: Vec<Partition>,
pub round_robin_counter: Arc<AtomicUsize>,
+
+ /// Consumer groups belonging to this topic, keyed by monotonic group id.
+ /// Co-located so a stream/topic delete drops them automatically.
+ pub consumer_groups: AHashMap<u64, ConsumerGroup>,
+ /// Group name -> id, for per-topic name uniqueness + name resolution.
+ pub consumer_group_index: AHashMap<Arc<str>, u64>,
+ /// Monotonic group-id counter; never reused so the partition-plane offset
+ /// key (keyed by group id) can't be inherited by a recreated group.
+ ///
+ /// Ceiling: the partition-plane offset key is `u32`, so a group id must stay
+ /// within `u32::MAX` (the wire rewrite in `server-ng` truncates to u32 and
+ /// `expect`s this). ~4 billion group creates on a single topic is
+ /// unreachable in practice, but the cap is real -- past it the wire id would
+ /// wrap and could collide with a live group's offset key.
+ pub next_consumer_group_id: u64,
}
impl Default for Topic {
@@ -136,6 +178,9 @@ impl Default for Topic {
stats: Arc::new(TopicStats::default()),
partitions: Vec::new(),
round_robin_counter: Arc::new(AtomicUsize::new(0)),
+ consumer_groups: AHashMap::default(),
+ consumer_group_index: AHashMap::default(),
+ next_consumer_group_id: 1,
}
}
}
@@ -161,6 +206,37 @@ impl Topic {
stats: Arc::new(TopicStats::new(stream_stats)),
partitions: Vec::new(),
round_robin_counter: Arc::new(AtomicUsize::new(0)),
+ consumer_groups: AHashMap::default(),
+ consumer_group_index: AHashMap::default(),
+ next_consumer_group_id: 1,
+ }
+ }
+
+ /// Re-run round-robin assignment for every consumer group under this topic
+ /// against the current partition set. Called after a partition-count change
+ /// (`CreatePartitions`/`DeletePartitions`) so groups pick up added
+ /// partitions and drop removed ones; each `rebalance_members` bumps the
+ /// group generation so stale clients re-sync.
+ pub fn rebalance_consumer_groups(&mut self) {
+ if self.consumer_groups.is_empty() {
+ return;
+ }
+ let partition_ids: Vec<usize> = self.partitions.iter().map(|p| p.id).collect();
+ for group in self.consumer_groups.values_mut() {
+ group.rebalance_members(&partition_ids);
+ }
+ }
+
+ /// Resolve a consumer-group identifier to its monotonic id within this
+ /// topic. Numeric resolves directly; string via the name index.
+ #[must_use]
+ pub fn resolve_group_id(&self, group_id: &WireIdentifier) -> Option<u64> {
+ match group_id {
+ WireIdentifier::Numeric(id) => {
+ let id = u64::from(*id);
+ self.consumer_groups.contains_key(&id).then_some(id)
+ }
+ WireIdentifier::String(name) => self.consumer_group_index.get(name.as_str()).copied(),
}
}
}
@@ -248,6 +324,12 @@ define_state! {
// it onto each new Partition::created_revision. Deterministic across
// replicas: same ops, same order.
revision: u64,
+ // Total pending cooperative revocations across all groups, recomputed
+ // once per commit by `post_apply`. The consensus tick reads it O(1)
+ // every 10ms instead of walking every stream/topic/group/member to
+ // decide whether to wake the reconciler. Deterministic (same ops, same
+ // recompute on every replica).
+ pending_revocations_count: u64,
}
}
@@ -263,10 +345,40 @@ collect_handlers! {
PurgeTopic,
CreatePartitionsWithAssignments,
DeletePartitions,
+ // Consumer groups are co-located under the topic, so the Streams STM
+ // applies these too. `Join`/`Leave` use the enriched request types from
+ // `crate::stm::consumer_group` (imported above) which carry the VSR
+ // client id.
+ CreateConsumerGroup,
+ DeleteConsumerGroup,
+ JoinConsumerGroup,
+ LeaveConsumerGroup,
+ RemoveConsumerGroupMember,
+ CompleteConsumerGroupRevocation,
}
}
impl StreamsInner {
+ /// Recompute `pending_revocations_count` so the consensus tick's
+ /// `has_pending_revocations` read (and the reconciler's fast-skip) is O(1)
+ /// instead of walking every group each 10ms. Called only by the apply
+ /// handlers that can change pending revocations (join, leave, remove,
+ /// complete, and group-dropping deletes), so non-consumer-group commits pay
+ /// nothing. Recompute (not a delta) keeps the count drift-proof.
+ pub(crate) fn recompute_pending_revocations_count(&mut self) {
+ let mut count: u64 = 0;
+ for (_, stream) in &self.items {
+ for (_, topic) in &stream.topics {
+ for group in topic.consumer_groups.values() {
+ for (_, member) in &group.members {
+ count += member.pending_revocations.len() as u64;
+ }
+ }
+ }
+ }
+ self.pending_revocations_count = count;
+ }
+
fn resolve_stream_id(&self, identifier: &WireIdentifier) -> Option<usize> {
match identifier {
WireIdentifier::Numeric(id) => {
@@ -295,6 +407,19 @@ impl StreamsInner {
WireIdentifier::String(name) => stream.topic_index.get(name.as_str()).copied(),
}
}
+
+ /// Mutable topic resolved from (stream, topic) identifiers -- the
+ /// consumer-group `StateHandler`s in [`crate::stm::consumer_group`] operate
+ /// through this.
+ pub(crate) fn topic_mut(
+ &mut self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ ) -> Option<&mut Topic> {
+ let stream_id = self.resolve_stream_id(stream_id)?;
+ let topic_id = self.resolve_topic_id(stream_id, topic_id)?;
+ self.items.get_mut(stream_id)?.topics.get_mut(topic_id)
+ }
}
impl Streams {
@@ -306,6 +431,312 @@ impl Streams {
self.inner.read(f)
}
+ /// Build the `ConsumerGroupDetailsResponse` for a group (members + their
+ /// round-robin partition assignment). `partitions_count` is the topic's
+ /// total partition count. `None` if the stream/topic/group is unknown.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation, clippy::missing_panics_doc)]
+ pub fn consumer_group_details(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ group_id: &WireIdentifier,
+ ) -> Option<ConsumerGroupDetailsResponse> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ let group = topic
+ .consumer_groups
+ .get(&topic.resolve_group_id(group_id)?)?;
+ let members = group
+ .members
+ .iter()
+ .map(|(_, member)| ConsumerGroupMemberResponse {
+ id: member.id as u32,
+ partitions_count: member.partitions.len() as u32,
+ partitions: member.partitions.iter().map(|&p| p as u32).collect(),
+ })
+ .collect();
+ Some(ConsumerGroupDetailsResponse {
+ group: ConsumerGroupResponse {
+ id: group.id as u32,
+ partitions_count: topic.partitions.len() as u32,
+ members_count: group.members.len() as u32,
+ // The name was validated at create, so the fallback is
+ // unreachable.
+ name: WireName::new(group.name.as_ref())
+ .unwrap_or_else(|_| WireName::new("unknown").expect("valid")),
+ },
+ members,
+ })
+ })
+ }
+
+ /// All consumer groups of a topic (for `GetConsumerGroups`). `None` if the
+ /// stream/topic is unknown.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation, clippy::missing_panics_doc)]
+ pub fn consumer_group_list(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ ) -> Option<Vec<ConsumerGroupResponse>> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ let partitions_count = topic.partitions.len() as u32;
+ Some(
+ topic
+ .consumer_groups
+ .values()
+ .map(|group| ConsumerGroupResponse {
+ id: group.id as u32,
+ partitions_count,
+ members_count: group.members.len() as u32,
+ name: WireName::new(group.name.as_ref())
+ .unwrap_or_else(|_| WireName::new("unknown").expect("valid")),
+ })
+ .collect(),
+ )
+ })
+ }
+
+ /// The requesting member's `(generation, partitions)` -- served by the
+ /// `SyncConsumerGroup` endpoint for client-side partition selection.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn consumer_group_member_assignment(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ group_id: &WireIdentifier,
+ client_id: u128,
+ ) -> Option<(u64, Vec<u32>)> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ let group = topic
+ .consumer_groups
+ .get(&topic.resolve_group_id(group_id)?)?;
+ let (_, member) = group
+ .members
+ .iter()
+ .find(|(_, m)| m.client_id == client_id)?;
+ // The client polls only its non-revoked partitions; a partition
+ // pending handoff stays owned (commit fence) but is no longer polled
+ // so its consumer can drain + commit it, completing the revocation.
+ let partitions = member
+ .pollable_partitions()
+ .iter()
+ .map(|&p| p as u32)
+ .collect();
+ Some((group.generation, partitions))
+ })
+ }
+
+ /// Whether any consumer group has a pending cooperative revocation. O(1):
+ /// reads the `pending_revocations_count` that `post_apply` maintains per
+ /// commit. The consensus tick polls this every 10ms to wake the reconciler
+ /// promptly when a source drains a revoked partition, so it must not walk.
+ #[must_use]
+ pub fn has_pending_revocations(&self) -> bool {
+ self.inner.read(|inner| inner.pending_revocations_count > 0)
+ }
+
+ /// The topic's current partition ids, for the join-time in-flight gather
+ /// (the home shard reads each partition's poll/commit state to classify the
+ /// cooperative handoff). `None` if the stream/topic does not resolve.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn topic_partition_ids(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ ) -> Option<Vec<u32>> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ Some(topic.partitions.iter().map(|p| p.id as u32).collect())
+ })
+ }
+
+ /// Partitions currently owned by some live member of the group (union over
+ /// members, pending-revoked included since the source still owns them until
+ /// completion). The join-time in-flight gather uses this to tell a genuine
+ /// in-flight hold (a live member polled past its commit) from a stale
+ /// `last_polled` left by a since-removed member: only an owned partition can
+ /// be in flight, so an unowned one with uncommitted data is the dead-member
+ /// residue of a reconnect and must be reassigned, not protected.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn consumer_group_assigned_partitions(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ group_id: &WireIdentifier,
+ ) -> Option<std::collections::HashSet<u32>> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ let group = topic
+ .consumer_groups
+ .get(&topic.resolve_group_id(group_id)?)?;
+ Some(
+ group
+ .members
+ .iter()
+ .flat_map(|(_, member)| member.partitions.iter().map(|&p| p as u32))
+ .collect(),
+ )
+ })
+ }
+
+ /// Every pending cooperative revocation across all groups, as
+ /// `(stream_id, topic_id, group_id, source_client_id, partition_id,
+ /// created_at_micros)`. The reconciler reads this each pass to decide which
+ /// revocations to complete (source drained, or timed out).
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation, clippy::type_complexity)]
+ pub fn consumer_group_pending_revocations(&self) -> Vec<(u32, u32, u64, u128, u32, u64)> {
+ self.inner.read(|inner| {
+ let mut out = Vec::new();
+ for (stream_id, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ for group in topic.consumer_groups.values() {
+ for (source_client_id, partition_id, created_at) in
+ group.pending_revocations()
+ {
+ out.push((
+ stream_id as u32,
+ topic_id as u32,
+ group.id,
+ source_client_id,
+ partition_id as u32,
+ created_at,
+ ));
+ }
+ }
+ }
+ }
+ out
+ })
+ }
+
+ /// The group's id (the consumer-group offset key) if `client_id` currently
+ /// owns `partition_id` in it -- the poll/commit fence. `None` for a stale
+ /// client whose partition was reassigned, prompting a re-sync.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn consumer_group_fence(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ group_id: &WireIdentifier,
+ client_id: u128,
+ partition_id: u32,
+ require_pollable: bool,
+ ) -> Option<u64> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ let group = topic
+ .consumer_groups
+ .get(&topic.resolve_group_id(group_id)?)?;
+ let (_, member) = group
+ .members
+ .iter()
+ .find(|(_, m)| m.client_id == client_id)?;
+ // Poll fence (`require_pollable`) rejects a pending-revoked partition
+ // so the source stops polling it (re-sync drops it from its set);
+ // commit fence keeps the full owned set so the source can still
+ // commit it and drain the handoff.
+ let owns = if require_pollable {
+ member.is_pollable(partition_id as usize)
+ } else {
+ member.partitions.iter().any(|&p| p as u32 == partition_id)
+ };
+ owns.then_some(group.id)
+ })
+ }
+
+ /// The group's monotonic id (the consumer-group offset key) regardless of
+ /// membership. `None` if the stream/topic/group no longer resolves, so a
+ /// consumer-offset read of a deleted group reports "no offset" and a write
+ /// rewrite can substitute the numeric id the partition plane keys under.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn resolve_consumer_group_id(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ group_id: &WireIdentifier,
+ ) -> Option<u64> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ topic.resolve_group_id(group_id)
+ })
+ }
+
+ /// `(stream_id, topic_id, group_id)` of every group the client belongs to,
+ /// for `get_me` membership reporting.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn consumer_group_memberships(&self, client_id: u128) -> Vec<(u32, u32, u32)> {
+ self.inner.read(|inner| {
+ let mut out = Vec::new();
+ for (stream_id, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ for group in topic.consumer_groups.values() {
+ if group.members.iter().any(|(_, m)| m.client_id == client_id) {
+ out.push((stream_id as u32, topic_id as u32, group.id as u32));
+ }
+ }
+ }
+ }
+ out
+ })
+ }
+
+ /// Drop a disconnected client from every consumer group it joined and
+ /// rebalance. Applied through the left-right writer as a deterministic
+ /// side-effect of the `Logout` commit on each replica (not a separate
+ /// replicated op). A no-op on the reader-mode peers, where commits aren't
+ /// applied.
+ pub fn remove_consumer_group_member(&self, client_id: u128, timestamp: IggyTimestamp) {
+ let cmd = StreamsCommand::RemoveConsumerGroupMember(
+ RemoveConsumerGroupMemberRequest { client_id },
+ timestamp,
+ );
+ if let Err(error) = self.inner.try_apply(cmd) {
+ tracing::error!(
+ client_id,
+ %error,
+ "remove_consumer_group_member dispatched to reader-only Streams STM"
+ );
+ }
+ }
+
+ /// Total consumer-group count across all topics (for stats).
+ #[must_use]
+ pub fn consumer_group_count(&self) -> usize {
+ self.inner.read(|inner| {
+ inner
+ .items
+ .iter()
+ .flat_map(|(_, stream)| stream.topics.iter())
+ .map(|(_, topic)| topic.consumer_groups.len())
+ .sum()
+ })
+ }
+
#[must_use]
pub fn partition_count_context(
&self,
@@ -339,6 +770,56 @@ impl Streams {
.map(|(_, next_partition_id)| next_partition_id)
}
+ /// Pick the next partition for a `Balanced` send, advancing the topic's
+ /// round-robin counter. `None` if the topic has no partitions.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn next_balanced_partition(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ ) -> Option<u32> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ let count = topic.partitions.len();
+ if count == 0 {
+ return None;
+ }
+ let current = topic
+ .round_robin_counter
+ .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |c| {
+ Some((c + 1) % count)
+ })
+ .unwrap_or(0);
+ Some(topic.partitions[current % count].id as u32)
+ })
+ }
+
+ /// Pick the partition for a `MessagesKey` send by hashing the key modulo
+ /// the partition count. `None` if the topic has no partitions.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn partition_by_messages_key(
+ &self,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ key: &[u8],
+ ) -> Option<u32> {
+ self.inner.read(|inner| {
+ let stream_id = inner.resolve_stream_id(stream_id)?;
+ let topic_id = inner.resolve_topic_id(stream_id, topic_id)?;
+ let topic = inner.items.get(stream_id)?.topics.get(topic_id)?;
+ let count = topic.partitions.len();
+ if count == 0 {
+ return None;
+ }
+ let index = iggy_common::calculate_32(key) as usize % count;
+ Some(topic.partitions[index % count].id as u32)
+ })
+ }
+
#[must_use]
pub fn namespace_from_partition(
&self,
@@ -455,6 +936,8 @@ impl StateHandler for DeleteStreamRequest {
state.items.remove(stream_id);
state.index.remove(&name);
state.revision = state.revision.wrapping_add(1);
+ // The dropped stream may have held groups with pending revocations.
+ state.recompute_pending_revocations_count();
Bytes::new()
}
}
@@ -516,6 +999,9 @@ impl StateHandler for CreateTopicWithAssignmentsRequest {
stats: Arc::new(TopicStats::new(stream.stats.clone())),
partitions: Vec::new(),
round_robin_counter: Arc::new(AtomicUsize::new(0)),
+ consumer_groups: AHashMap::default(),
+ consumer_group_index: AHashMap::default(),
+ next_consumer_group_id: 1,
};
let topic_id = stream.topics.insert(topic);
@@ -660,6 +1146,8 @@ impl StateHandler for DeleteTopicRequest {
stream.topics.remove(topic_id);
stream.topic_index.remove(&name);
state.revision = state.revision.wrapping_add(1);
+ // The dropped topic may have held groups with pending revocations.
+ state.recompute_pending_revocations_count();
Bytes::new()
}
}
@@ -736,6 +1224,8 @@ impl StateHandler for CreatePartitionsWithAssignmentsRequest {
created_revision: new_revision,
});
}
+ // Added partitions are unassigned until the groups rebalance.
+ topic.rebalance_consumer_groups();
// Matches legacy CreatePartitions wire contract: empty-ok body on
// success. SDK discards the reply payload (resolved ids are derivable
@@ -767,6 +1257,8 @@ impl StateHandler for DeletePartitionsRequest {
topic
.partitions
.truncate(topic.partitions.len() - count_to_delete);
+ // Members assigned the removed partitions must give them up.
+ topic.rebalance_consumer_groups();
}
if did_delete {
state.revision = state.revision.wrapping_add(1);
@@ -825,9 +1317,14 @@ impl Snapshotable for Streams {
created_revision: p.created_revision,
})
.collect(),
- round_robin_counter: topic
- .round_robin_counter
- .load(Ordering::Relaxed),
+ consumer_groups: topic
+ .consumer_groups
+ .iter()
+ .map(|(&id, group)| {
+ (id, ConsumerGroupSnapshot::from_group(group))
+ })
+ .collect(),
+ next_consumer_group_id: topic.next_consumer_group_id,
},
)
})
@@ -899,7 +1396,29 @@ impl Snapshotable for Streams {
created_revision: p.created_revision,
})
.collect(),
- round_robin_counter: Arc::new(AtomicUsize::new(topic_snap.round_robin_counter)),
+ // Not snapshotted (see `TopicSnapshot`): start fresh.
+ round_robin_counter: Arc::new(AtomicUsize::new(0)),
+ consumer_group_index: topic_snap
+ .consumer_groups
+ .iter()
+ .map(|(_, group_snap)| (Arc::from(group_snap.name.as_str()), group_snap.id))
+ .collect(),
+ next_consumer_group_id: topic_snap
+ .next_consumer_group_id
+ .max(
+ topic_snap
+ .consumer_groups
+ .iter()
+ .map(|(id, _)| id + 1)
+ .max()
+ .unwrap_or(1),
+ )
+ .max(1),
+ consumer_groups: topic_snap
+ .consumer_groups
+ .into_iter()
+ .map(|(id, group_snap)| (id, group_snap.into_group()))
+ .collect(),
};
topic_index.insert(topic_name, topic_slab_key);
topic_entries.push((topic_slab_key, topic));
@@ -922,12 +1441,15 @@ impl Snapshotable for Streams {
}
let items: Slab<Stream> = stream_entries.into_iter().collect();
- let inner = StreamsInner {
+ let mut inner = StreamsInner {
index,
items,
revision: snapshot.revision,
+ // Recomputed from the restored groups just below.
+ pending_revocations_count: 0,
last_result: None,
};
+ inner.recompute_pending_revocations_count();
Ok(inner.into())
}
}
diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs
index 4a1e6965b..3250f8311 100644
--- a/core/partitions/src/iggy_partition.rs
+++ b/core/partitions/src/iggy_partition.rs
@@ -86,6 +86,12 @@ where
pub dirty_offset: AtomicU64,
pub consumer_offsets: Arc<ConsumerOffsets>,
pub consumer_group_offsets: Arc<ConsumerGroupOffsets>,
+ /// Highest offset this partition has served (polled) to each consumer group.
+ /// The cooperative-rebalance reconciler completes a pending revocation once
+ /// the source group has committed up to what it was polled
+ /// (`committed >= last_polled`), i.e. nothing is in flight. Ephemeral (not
+ /// persisted): a fresh server treats a group as never-polled.
+ pub last_polled_offsets: Arc<ConsumerGroupOffsets>,
pub stats: Arc<PartitionStats>,
pub created_at: IggyTimestamp,
pub revision_id: u64,
@@ -178,6 +184,7 @@ where
dirty_offset: AtomicU64::new(0),
consumer_offsets: Arc::new(ConsumerOffsets::with_capacity(1)),
consumer_group_offsets: Arc::new(ConsumerGroupOffsets::with_capacity(1)),
+ last_polled_offsets: Arc::new(ConsumerGroupOffsets::with_capacity(1)),
stats,
created_at: IggyTimestamp::now(),
revision_id: 0,
@@ -390,6 +397,35 @@ where
}
}
+ /// Group ids that currently have a stored offset on this partition. Used by
+ /// the reconciler to find offsets belonging to deleted consumer groups.
+ #[must_use]
+ #[allow(clippy::cast_possible_truncation)]
+ pub fn consumer_group_offset_ids(&self) -> Vec<u64> {
+ self.consumer_group_offsets
+ .pin()
+ .keys()
+ .map(|key| key.0 as u64)
+ .collect()
+ }
+
+ /// Reclaim a deleted consumer group's offset on this partition (in-memory
+ /// entry + persisted file). A no-op if the group has no stored offset here.
+ ///
+ /// # Errors
+ /// Returns an I/O error if deleting the persisted offset file fails.
+ #[allow(clippy::cast_possible_truncation)]
+ pub async fn delete_consumer_group_offset(&self, group_id: u64) -> Result<(), IggyError> {
+ self.consumer_group_offsets
+ .pin()
+ .remove(&ConsumerGroupId(group_id as usize));
+ if let Some(path) = self.persisted_offset_path(ConsumerKind::ConsumerGroup, group_id as u32)
+ {
+ delete_persisted_offset(&path).await?;
+ }
+ Ok(())
+ }
+
async fn store_consumer_offset_and_persist(
&self,
consumer: PollingConsumer,
@@ -626,6 +662,27 @@ where
let (fragments, last_matching_offset) =
result.unwrap_or_else(|| (PollFragments::new(), None));
+ // Record the highest offset served to a consumer group, so the
+ // cooperative-rebalance reconciler knows when a pending-revoked
+ // partition has been fully drained (committed >= last polled).
+ if let (PollingConsumer::ConsumerGroup(group_id, _), Some(last_offset)) =
+ (consumer, last_matching_offset)
+ {
+ let guard = self.last_polled_offsets.pin();
+ let key = ConsumerGroupId(group_id);
+ if let Some(existing) = guard.get(&key) {
+ existing.offset.fetch_max(last_offset, Ordering::Relaxed);
+ } else {
+ let created = ConsumerOffset::new(
+ ConsumerKind::ConsumerGroup,
+ u32::try_from(group_id).unwrap_or(u32::MAX),
+ last_offset,
+ String::new(),
+ );
+ guard.insert(key, created);
+ }
+ }
+
if args.auto_commit && !fragments.is_empty() {
let last_offset =
last_matching_offset.expect("non-empty poll result must have a last offset");
diff --git a/core/sdk/src/client_wrappers/binary_system_client.rs b/core/sdk/src/client_wrappers/binary_system_client.rs
index 80db85278..e4e34d9ad 100644
--- a/core/sdk/src/client_wrappers/binary_system_client.rs
+++ b/core/sdk/src/client_wrappers/binary_system_client.rs
@@ -85,6 +85,18 @@ impl SystemClient for ClientWrapper {
}
}
+ async fn refresh_consumer_group_assignments(&self) {
+ match self {
+ ClientWrapper::Iggy(client) => client.refresh_consumer_group_assignments().await,
+ ClientWrapper::Http(client) => client.refresh_consumer_group_assignments().await,
+ ClientWrapper::Tcp(client) => client.refresh_consumer_group_assignments().await,
+ ClientWrapper::Quic(client) => client.refresh_consumer_group_assignments().await,
+ ClientWrapper::WebSocket(client) => {
+ client.refresh_consumer_group_assignments().await;
+ }
+ }
+ }
+
async fn snapshot(
&self,
compression: SnapshotCompression,
diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index 16de4715d..98beb9830 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs
@@ -259,6 +259,13 @@ impl Client for IggyClient {
}
} else {
debug!("Heartbeat was sent successfully.");
+ // Picks up a widened assignment (e.g. partition-count
+ // change) without waiting for an ownership-fence rejection.
+ client
+ .read()
+ .await
+ .refresh_consumer_group_assignments()
+ .await;
}
sleep(heartbeat_interval.get_duration()).await
}
diff --git a/core/sdk/src/quic/quic_client.rs b/core/sdk/src/quic/quic_client.rs
index 2aacab718..d59317ee7 100644
--- a/core/sdk/src/quic/quic_client.rs
+++ b/core/sdk/src/quic/quic_client.rs
@@ -71,6 +71,8 @@ pub struct QuicClient {
consensus_session: Arc<StdMutex<ConsensusSession>>,
#[cfg(feature = "vsr")]
skip_auto_login_once: Mutex<bool>,
+ #[cfg(feature = "vsr")]
+ consumer_group_state: Arc<iggy_common::ConsumerGroupClientState>,
}
unsafe impl Send for QuicClient {}
@@ -170,6 +172,11 @@ impl BinaryTransport for QuicClient {
fn get_heartbeat_interval(&self) -> IggyDuration {
self.config.heartbeat_interval
}
+
+ #[cfg(feature = "vsr")]
+ fn consumer_group_state(&self) -> Arc<iggy_common::ConsumerGroupClientState> {
+ Arc::clone(&self.consumer_group_state)
+ }
}
#[cfg(feature = "vsr")]
@@ -272,6 +279,8 @@ impl QuicClient {
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
#[cfg(feature = "vsr")]
skip_auto_login_once: Mutex::new(false),
+ #[cfg(feature = "vsr")]
+ consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
diff --git a/core/sdk/src/tcp/tcp_client.rs b/core/sdk/src/tcp/tcp_client.rs
index f6e6f4980..6a5f2c684 100644
--- a/core/sdk/src/tcp/tcp_client.rs
+++ b/core/sdk/src/tcp/tcp_client.rs
@@ -84,6 +84,8 @@ pub struct TcpClient {
consensus_session: Arc<StdMutex<ConsensusSession>>,
#[cfg(feature = "vsr")]
skip_auto_login_once: Mutex<bool>,
+ #[cfg(feature = "vsr")]
+ consumer_group_state: Arc<iggy_common::ConsumerGroupClientState>,
}
impl Default for TcpClient {
@@ -199,6 +201,11 @@ impl BinaryTransport for TcpClient {
fn get_heartbeat_interval(&self) -> IggyDuration {
self.config.heartbeat_interval
}
+
+ #[cfg(feature = "vsr")]
+ fn consumer_group_state(&self) -> Arc<iggy_common::ConsumerGroupClientState> {
+ Arc::clone(&self.consumer_group_state)
+ }
}
#[cfg(feature = "vsr")]
@@ -294,6 +301,8 @@ impl TcpClient {
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
#[cfg(feature = "vsr")]
skip_auto_login_once: Mutex::new(false),
+ #[cfg(feature = "vsr")]
+ consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 0dbbf8182..2b74d0318 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -254,6 +254,7 @@ fn map_eviction_reason(reason: EvictionReason) -> IggyError {
| EvictionReason::NoSession
| EvictionReason::SessionTooLow
| EvictionReason::SessionReleaseMismatch => IggyError::Unauthenticated,
+ EvictionReason::StaleClient => IggyError::StaleClient,
_ => IggyError::InvalidCommand,
}
}
diff --git a/core/sdk/src/websocket/websocket_client.rs b/core/sdk/src/websocket/websocket_client.rs
index e1a983839..aa34977b5 100644
--- a/core/sdk/src/websocket/websocket_client.rs
+++ b/core/sdk/src/websocket/websocket_client.rs
@@ -82,6 +82,8 @@ pub struct WebSocketClient {
consensus_session: Arc<StdMutex<ConsensusSession>>,
#[cfg(feature = "vsr")]
skip_auto_login_once: Mutex<bool>,
+ #[cfg(feature = "vsr")]
+ consumer_group_state: Arc<iggy_common::ConsumerGroupClientState>,
}
impl Default for WebSocketClient {
@@ -186,6 +188,11 @@ impl BinaryTransport for WebSocketClient {
fn get_heartbeat_interval(&self) -> IggyDuration {
self.config.heartbeat_interval
}
+
+ #[cfg(feature = "vsr")]
+ fn consumer_group_state(&self) -> Arc<iggy_common::ConsumerGroupClientState> {
+ Arc::clone(&self.consumer_group_state)
+ }
}
#[cfg(feature = "vsr")]
@@ -240,6 +247,8 @@ impl WebSocketClient {
consensus_session: Arc::new(StdMutex::new(ConsensusSession::new())),
#[cfg(feature = "vsr")]
skip_auto_login_once: Mutex::new(false),
+ #[cfg(feature = "vsr")]
+ consumer_group_state: Arc::new(iggy_common::ConsumerGroupClientState::new()),
})
}
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 6465525c5..1e93ce804 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -60,7 +60,6 @@ use metadata::IggyMetadata;
use metadata::MuxStateMachine;
use metadata::impls::metadata::{IggySnapshot, StreamsFrontend};
use metadata::impls::recovery::recover;
-use metadata::stm::consumer_group::ConsumerGroups;
use metadata::stm::mux::WithFactory;
use metadata::stm::snapshot::Snapshot;
use metadata::stm::stream::{Partition, Streams};
@@ -98,13 +97,13 @@ use tracing::{error, info, warn};
const SHARD_REPLICA_ID: u8 = 0;
-type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams)>;
/// Cross-thread bundle carrying one `ReadHandleFactory` per metadata
/// state. Shard 0 mints one after `recover()` and broadcasts a clone to
/// every peer shard; each peer rebuilds a reader-mode
/// [`ServerNgMuxStateMachine`] on its own runtime, skipping the WAL.
-type ServerNgMetadataBundle = <variadic!(Users, Streams, ConsumerGroups) as WithFactory>::Bundle;
+type ServerNgMetadataBundle = <variadic!(Users, Streams) as WithFactory>::Bundle;
pub(crate) type ServerNgMetadata = IggyMetadata<
VsrConsensus<Rc<IggyMessageBus>>,
@@ -895,11 +894,41 @@ async fn shard_main(
}
tick_shard.tick_metadata().await;
tick_shard.tick_partitions().await;
+ // While a cooperative revocation is pending, wake the reconciler each
+ // tick so it completes the handoff within ~one tick of the source
+ // draining the partition, instead of waiting for the 1s periodic pass.
+ if tick_shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .has_pending_revocations()
+ {
+ tick_shard.dispatch_metadata_commit_tick();
+ }
compio::time::sleep(CONSENSUS_TICK_INTERVAL).await;
}
});
bus.track_background(consensus_tick_handle);
+ // Per-shard heartbeat verifier: evicts connections that stop pinging,
+ // releasing their consumer-group membership. Gated on config so a
+ // deployment without heartbeats never reaps live sessions.
+ let heartbeat_stop_tx = if config.heartbeat.enabled {
+ let (hb_stop_tx, hb_stop_rx) = channel::<()>(1);
+ let hb_shard = Rc::clone(&shard);
+ let hb_sessions = Rc::clone(&sessions);
+ let hb_interval = config.heartbeat.interval.get_duration();
+ let hb_handle = compio::runtime::spawn(async move {
+ crate::dispatch::run_heartbeat_verifier(hb_shard, hb_sessions, hb_interval, hb_stop_rx)
+ .await;
+ });
+ bus.track_background(hb_handle);
+ Some(hb_stop_tx)
+ } else {
+ None
+ };
+
// Expired-PAT cleaner: shard 0 only (it owns the metadata consensus
// group) and only when enabled. Each pass no-ops unless this node is
// the caught-up metadata primary, so the delete is proposed once and
@@ -955,6 +984,9 @@ async fn shard_main(
let _ = stop_tx.try_send(());
let _ = reconcile_stop_tx.try_send(());
let _ = consensus_tick_stop_tx.try_send(());
+ if let Some(tx) = &heartbeat_stop_tx {
+ let _ = tx.try_send(());
+ }
if let Some(cleaner_stop_tx) = &pat_cleaner_stop {
let _ = cleaner_stop_tx.try_send(());
}
@@ -966,6 +998,9 @@ async fn shard_main(
let _ = stop_tx.try_send(());
let _ = reconcile_stop_tx.try_send(());
let _ = consensus_tick_stop_tx.try_send(());
+ if let Some(tx) = &heartbeat_stop_tx {
+ let _ = tx.try_send(());
+ }
if let Some(cleaner_stop_tx) = &pat_cleaner_stop {
let _ = cleaner_stop_tx.try_send(());
}
diff --git a/core/server-ng/src/consumer_group.rs b/core/server-ng/src/consumer_group.rs
new file mode 100644
index 000000000..7509363cb
--- /dev/null
+++ b/core/server-ng/src/consumer_group.rs
@@ -0,0 +1,257 @@
+// 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.
+
+//! Consumer-group Join/Leave request enrichment.
+//!
+//! The wire `Join`/`Leave` requests carry only the stream/topic/group
+//! identifiers. The replicated metadata `apply` additionally needs the
+//! joining client's VSR id (which member) and, for Join, the topic's
+//! partition count (to seed the group's partition list) -- and it cannot
+//! read the Streams STM from inside the consumer-group apply. So the
+//! primary enriches the op here before replication, mirroring the PAT mint
+//! in [`crate::pat`] and the password hash in [`crate::users`].
+
+use crate::bootstrap::ServerNgShard;
+use crate::responses::resolve_partition_namespace;
+use crate::wire::{request_body, rewrite_request_body};
+use consensus::MetadataHandle;
+use iggy_binary_protocol::codec::{WireDecode, WireEncode};
+use iggy_binary_protocol::primitives::consumer::WireConsumer;
+use iggy_binary_protocol::requests::consumer_groups::{
+ JoinConsumerGroupRequest as WireJoinConsumerGroupRequest,
+ LeaveConsumerGroupRequest as WireLeaveConsumerGroupRequest,
+};
+use iggy_binary_protocol::requests::consumer_offsets::{
+ DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request,
+ StoreConsumerOffsetRequest,
+};
+use iggy_binary_protocol::{KIND_CONSUMER_GROUP, Operation, RequestHeader, WireIdentifier};
+use iggy_common::IggyError;
+use metadata::impls::metadata::StreamsFrontend;
+use metadata::stm::consumer_group::{
+ JoinConsumerGroupRequest as ReplicatedJoinConsumerGroupRequest,
+ LeaveConsumerGroupRequest as ReplicatedLeaveConsumerGroupRequest,
+};
+use server_common::Message;
+use shard::{PartitionRead, PartitionReadReply};
+use std::rc::Rc;
+
+/// Rewrite a `Join`/`Leave` request body into the replicated form carrying the
+/// client's VSR id (which the apply can't read from the consensus header).
+///
+/// For `Join` the home shard also gathers `in_flight` -- the group's partitions
+/// with uncommitted polled data -- by reading each partition's poll/commit state
+/// (via the partition-read mesh), so the cooperative rebalance pending-revokes
+/// only those and hands off never-polled/drained partitions synchronously at
+/// join. Every other operation passes through.
+pub(crate) async fn maybe_rewrite_consumer_group_request(
+ shard: &Rc<ServerNgShard>,
+ request: Message<RequestHeader>,
+) -> Result<Message<RequestHeader>, IggyError> {
+ let operation = request.header().operation;
+ let client_id = request.header().client;
+ let body = request_body(&request);
+ let rewritten = match operation {
+ Operation::JoinConsumerGroup => {
+ let wire = WireJoinConsumerGroupRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let in_flight =
+ gather_in_flight(shard, &wire.stream_id, &wire.topic_id, &wire.group_id).await;
+ ReplicatedJoinConsumerGroupRequest {
+ stream_id: wire.stream_id,
+ topic_id: wire.topic_id,
+ group_id: wire.group_id,
+ client_id,
+ in_flight,
+ }
+ .to_bytes()
+ }
+ Operation::LeaveConsumerGroup => {
+ let wire = WireLeaveConsumerGroupRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ ReplicatedLeaveConsumerGroupRequest {
+ stream_id: wire.stream_id,
+ topic_id: wire.topic_id,
+ group_id: wire.group_id,
+ client_id,
+ }
+ .to_bytes()
+ }
+ _ => return Ok(request),
+ };
+
+ rewrite_request_body(&request, &rewritten)
+}
+
+/// Gather the group's in-flight partitions (`last_polled` present and
+/// `committed < last_polled`) for the cooperative-rebalance classification. A
+/// not-yet-created group, an unresolved topic, or a partition that does not
+/// answer is treated as not-in-flight (eager handoff). An eager handoff leaves
+/// no `PendingRevocation` record, so the reconciler never revisits it -- a
+/// misclassification here just redelivers the uncommitted range to the new
+/// owner, which is correct under at-least-once.
+async fn gather_in_flight(
+ shard: &Rc<ServerNgShard>,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ group_id: &WireIdentifier,
+) -> Vec<u32> {
+ let streams = shard.plane.metadata().mux_stm.streams();
+ let Some(monotonic_group_id) = streams.resolve_consumer_group_id(stream_id, topic_id, group_id)
+ else {
+ // Fresh group (e.g. create-if-not-exists): nothing polled yet.
+ return Vec::new();
+ };
+ let Some(partition_ids) = streams.topic_partition_ids(stream_id, topic_id) else {
+ return Vec::new();
+ };
+ // Partitions a live member currently owns. A `last_polled` past the commit
+ // only means in-flight work when a live member still holds the partition;
+ // for an unowned one it is the residue of a member removed on disconnect
+ // (the reconnect case), which must be reassigned and re-read, not protected.
+ let assigned = streams
+ .consumer_group_assigned_partitions(stream_id, topic_id, group_id)
+ .unwrap_or_default();
+ // Resolve namespaces up front (sync), then fire every partition's
+ // `GroupOffsetState` read concurrently. The reads are independent, so a
+ // wide-topic join must not serialize N cross-shard round-trips before the
+ // join op can even be proposed.
+ let targets: Vec<(u32, _)> = partition_ids
+ .into_iter()
+ .filter_map(|partition_id| {
+ resolve_partition_namespace(shard, stream_id, topic_id, Some(partition_id))
+ .ok()
+ .map(|ns| (partition_id, ns))
+ })
+ .collect();
+ let results = futures::future::join_all(targets.iter().map(|&(partition_id, ns)| async move {
+ let reply = shard
+ .partition_read(
+ ns,
+ PartitionRead::GroupOffsetState {
+ group_id: monotonic_group_id,
+ },
+ )
+ .await;
+ (partition_id, ns, reply)
+ }))
+ .await;
+
+ let mut in_flight = Vec::new();
+ let mut stale_clears = Vec::new();
+ for (partition_id, ns, reply) in results {
+ let Some(PartitionReadReply::GroupOffsetState {
+ last_polled: Some(polled),
+ committed,
+ }) = reply
+ else {
+ continue;
+ };
+ if committed.is_some_and(|c| c >= polled) {
+ continue;
+ }
+ if assigned.contains(&partition_id) {
+ in_flight.push(partition_id);
+ } else {
+ // Stale mark from a removed member: drop it so a later join in this
+ // same restart does not misread it once the partition is reassigned.
+ stale_clears.push(shard.partition_read(
+ ns,
+ PartitionRead::ClearGroupLastPolled {
+ group_id: monotonic_group_id,
+ },
+ ));
+ }
+ }
+ // Fire the stale-mark clears concurrently too; the result is unused.
+ futures::future::join_all(stale_clears).await;
+ in_flight
+}
+
+/// Rewrite a group consumer-offset op so its consumer id is the group's
+/// monotonic id rather than the wire name. The partition plane keys group
+/// offsets by that numeric id (decoded from `WireIdentifier::Numeric`), so the
+/// read path -- which resolves the same id from metadata -- and the reconciler
+/// purge agree, and a re-created group (new id) never inherits a stale offset.
+/// Individual-consumer ops and every other operation pass through untouched.
+#[allow(clippy::cast_possible_truncation)]
+pub(crate) fn maybe_rewrite_consumer_offset_request(
+ shard: &Rc<ServerNgShard>,
+ request: Message<RequestHeader>,
+) -> Result<Message<RequestHeader>, IggyError> {
+ let operation = request.header().operation;
+ if !matches!(
+ operation,
+ Operation::StoreConsumerOffset
+ | Operation::StoreConsumerOffset2
+ | Operation::DeleteConsumerOffset
+ | Operation::DeleteConsumerOffset2
+ ) {
+ return Ok(request);
+ }
+ let body = request_body(&request);
+ // The 4 store/delete (v1 + v2) ops differ only in the decode type; this
+ // collapses their identical decode -> resolve group id -> rewrite consumer
+ // id -> re-encode bodies. A non-group consumer or unresolved group returns
+ // the request untouched (the apply/read path handles the miss).
+ macro_rules! rewrite_group_offset {
+ ($ty:ty) => {{
+ let mut wire = <$ty>::decode_from(body).map_err(|_| IggyError::InvalidCommand)?;
+ let Some(group_id) =
+ resolve_group_offset_id(shard, &wire.consumer, (&wire.stream_id, &wire.topic_id))
+ else {
+ return Ok(request);
+ };
+ // The partition-plane group-offset key is u32 (see the documented
+ // ceiling on `Topic::next_consumer_group_id`). Clamp on the
+ // ~4-billion-creates overflow rather than panic this live
+ // client-driven path, matching `iggy_partition.rs`'s identical cast.
+ wire.consumer.id = WireIdentifier::Numeric(u32::try_from(group_id).unwrap_or(u32::MAX));
+ wire.to_bytes()
+ }};
+ }
+ let rewritten = match operation {
+ Operation::StoreConsumerOffset => rewrite_group_offset!(StoreConsumerOffsetRequest),
+ Operation::StoreConsumerOffset2 => rewrite_group_offset!(StoreConsumerOffset2Request),
+ Operation::DeleteConsumerOffset => rewrite_group_offset!(DeleteConsumerOffsetRequest),
+ Operation::DeleteConsumerOffset2 => rewrite_group_offset!(DeleteConsumerOffset2Request),
+ // The outer `matches!` already filtered to the 4 ops above, but the
+ // match is over the 37-variant `Operation`, so a catch-all is required.
+ _ => return Ok(request),
+ };
+
+ rewrite_request_body(&request, &rewritten)
+}
+
+/// Resolve the monotonic group id for a group consumer-offset op, or `None` for
+/// an individual consumer (kind != 2) / unresolved group (leave the body as-is;
+/// the apply / read path handle the miss).
+fn resolve_group_offset_id(
+ shard: &Rc<ServerNgShard>,
+ consumer: &WireConsumer,
+ namespace: (&WireIdentifier, &WireIdentifier),
+) -> Option<u64> {
+ if consumer.kind != KIND_CONSUMER_GROUP {
+ return None;
+ }
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .resolve_consumer_group_id(namespace.0, namespace.1, &consumer.id)
+}
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index e8b3e1099..24edd63bc 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -27,6 +27,9 @@ use crate::auth::{
verify_login_credentials, verify_pat_credentials,
};
use crate::bootstrap::{ServerNgShard, ServerNgShardHandle};
+use crate::consumer_group::{
+ maybe_rewrite_consumer_group_request, maybe_rewrite_consumer_offset_request,
+};
use crate::login_register::LoginRegisterError;
use crate::pat::maybe_rewrite_pat_request;
use crate::responses::{
@@ -42,22 +45,27 @@ use bytes::Bytes;
use consensus::{MetadataHandle, PartitionsHandle};
use iggy_binary_protocol::codes::{
GET_CLIENT_CODE, GET_CLIENTS_CODE, GET_CLUSTER_METADATA_CODE, GET_CONSUMER_OFFSET_CODE,
- GET_ME_CODE, PING_CODE, POLL_MESSAGES_CODE,
+ GET_ME_CODE, PING_CODE, POLL_MESSAGES_CODE, SYNC_CONSUMER_GROUP_CODE,
};
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy;
+use iggy_binary_protocol::requests::consumer_groups::SyncConsumerGroupRequest;
use iggy_binary_protocol::requests::consumer_offsets::GetConsumerOffsetRequest;
use iggy_binary_protocol::requests::messages::PollMessagesRequest;
use iggy_binary_protocol::requests::system::get_client::GetClientRequest;
use iggy_binary_protocol::requests::users::{LoginRegisterRequest, LoginRegisterWithPatRequest};
+use iggy_binary_protocol::responses::clients::client_response::ConsumerGroupInfoResponse;
use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse;
use iggy_binary_protocol::responses::clients::get_clients::GetClientsResponse;
-use iggy_binary_protocol::{GenericHeader, Operation, RequestHeader, WireDecode, WireEncode};
+use iggy_binary_protocol::responses::consumer_groups::SyncConsumerGroupResponse;
+use iggy_binary_protocol::{
+ GenericHeader, KIND_CONSUMER_GROUP, Operation, RequestHeader, WireDecode, WireEncode,
+};
use iggy_common::{IggyError, PollingStrategy};
use message_bus::client_listener::RequestHandler;
use message_bus::replica::listener::MessageHandler;
use message_bus::{IggyMessageBus, MessageBus};
-use metadata::impls::metadata::MetadataSubmitError;
+use metadata::impls::metadata::{MetadataSubmitError, StreamsFrontend};
use partitions::{Partition, PollingArgs, PollingConsumer};
use secrecy::ExposeSecret;
use server_common::Message;
@@ -184,6 +192,25 @@ pub(crate) fn make_partition_read_handler(
stored: partition.get_consumer_offset(consumer),
current_offset: partition.offsets().commit_offset,
},
+ PartitionRead::GroupOffsetState { group_id } => {
+ #[allow(clippy::cast_possible_truncation)]
+ let key = iggy_common::ConsumerGroupId(group_id as usize);
+ let load = |offset: &iggy_common::ConsumerOffset| {
+ offset.offset.load(std::sync::atomic::Ordering::Relaxed)
+ };
+ let committed = partition.consumer_group_offsets.pin().get(&key).map(load);
+ let last_polled = partition.last_polled_offsets.pin().get(&key).map(load);
+ PartitionReadReply::GroupOffsetState {
+ last_polled,
+ committed,
+ }
+ }
+ PartitionRead::ClearGroupLastPolled { group_id } => {
+ #[allow(clippy::cast_possible_truncation)]
+ let key = iggy_common::ConsumerGroupId(group_id as usize);
+ partition.last_polled_offsets.pin().remove(&key);
+ PartitionReadReply::Ack
+ }
};
let _ = reply.try_send(result);
})
@@ -304,6 +331,28 @@ pub(crate) fn make_metadata_submit_handler(
};
let _ = reply.try_send(committed);
}
+ shard::MetadataSubmit::CompleteRevocation {
+ stream_id,
+ topic_id,
+ group_id,
+ source_client_id,
+ partition_id,
+ reply,
+ } => {
+ let commit = shard
+ .plane
+ .metadata()
+ .submit_complete_revocation_in_process(
+ stream_id,
+ topic_id,
+ group_id,
+ source_client_id,
+ partition_id,
+ )
+ .await
+ .ok();
+ let _ = reply.try_send(commit);
+ }
}
})
.detach();
@@ -390,6 +439,12 @@ async fn handle_client_request(
ensure_transport_connection(shard, sessions, transport_client_id);
+ // Any request is liveness proof, not just PING: an idle-but-active client
+ // (e.g. an admin issuing reads between long sleeps) must not be evicted by
+ // the heartbeat verifier. A genuinely dead connection sends nothing, so the
+ // intended stale-client eviction still fires. No-ops for an unbound client.
+ sessions.borrow_mut().record_heartbeat(transport_client_id);
+
let header = *request.header();
if header.operation == Operation::NonReplicated {
// Auth bypass guard: only `PING` and `GET_CLUSTER_METADATA` are
@@ -465,10 +520,15 @@ async fn handle_client_request(
// TRANSPORT id (home-shard routing bits), not the VSR session id --
// partition ops are sessionless ("session lifecycle is metadata-only").
if header.operation.is_partition() {
+ // The consumer-group offset fence keys on the bound VSR client id (the
+ // member id), not the transport id stamped into the partition-op
+ // header. `bound` is Some here (unbound transports returned above).
+ let vsr_client_id = bound.map_or(0, |(client_id, _)| client_id);
let namespace = match resolve_partition_request_namespace(
shard,
header.operation,
request_body(&request),
+ vsr_client_id,
) {
Ok(namespace) => namespace,
Err(error) => {
@@ -503,6 +563,24 @@ async fn handle_client_request(
send_empty_partition_reply(shard, transport_client_id, &header).await;
return;
}
+ // A group consumer-offset op carries the group NAME on the wire; the
+ // partition plane keys the offset by the group's monotonic id (the same
+ // key the poll path auto-commits under and the read path resolves), so
+ // rewrite the consumer id before replication -- the apply layer has no
+ // metadata access to resolve it.
+ let request = match maybe_rewrite_consumer_offset_request(shard, request) {
+ Ok(rewritten) => rewritten,
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ operation = ?header.operation,
+ "failed to rewrite consumer-offset request; replying empty"
+ );
+ send_empty_partition_reply(shard, transport_client_id, &header).await;
+ return;
+ }
+ };
let request = request.transmute_header(|header, new_header: &mut RequestHeader| {
*new_header = header;
new_header.namespace = namespace;
@@ -557,6 +635,20 @@ async fn handle_client_request(
return;
}
};
+ // Enrich consumer-group Join/Leave with the client's VSR id (+ topic
+ // partition count for Join) before replication; see `crate::consumer_group`.
+ let request = match maybe_rewrite_consumer_group_request(shard, request).await {
+ Ok(rewritten) => rewritten,
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ operation = ?header.operation,
+ "dropping consumer-group request with invalid payload"
+ );
+ return;
+ }
+ };
let request_header = *request.header();
// Replicated request: run consensus on the metadata owner (shard 0) and
// bring the committed reply back here. This shard owns the connection,
@@ -615,6 +707,9 @@ async fn handle_non_replicated_request(
let code = u32::from_le_bytes(request.header().reserved[CODE_RANGE].try_into().unwrap());
match code {
PING_CODE => {
+ // A ping is the client's liveness proof; reset its staleness clock
+ // so the heartbeat verifier doesn't evict an active connection.
+ sessions.borrow_mut().record_heartbeat(transport_client_id);
let commit = current_metadata_commit(shard);
let reply = build_empty_reply(
request.header(),
@@ -639,7 +734,7 @@ async fn handle_non_replicated_request(
// sourced from this shard's `SessionManager` (not `IggyMetadata`),
// so it is built here rather than in
// `build_non_replicated_response`.
- let response = build_get_me_response(sessions, transport_client_id);
+ let response = build_get_me_response(shard, sessions, transport_client_id);
send_non_replicated_bytes(
shard,
&request,
@@ -654,7 +749,10 @@ async fn handle_non_replicated_request(
// gather across all shards (scatter-gather over the mesh).
let infos = shard.list_all_clients().await;
let response = GetClientsResponse {
- clients: infos.iter().map(connected_client_to_response).collect(),
+ clients: infos
+ .iter()
+ .map(|info| connected_client_to_response(shard, info))
+ .collect(),
};
send_non_replicated_bytes(
shard,
@@ -677,9 +775,26 @@ async fn handle_non_replicated_request(
let found = target.and_then(|id| infos.iter().find(|info| info.client_id as u32 == id));
// The SDK decodes an empty body as `None` (client not found).
let bytes = found.map_or_else(Bytes::new, |info| {
+ let consumer_groups = info.vsr_client_id.map_or_else(Vec::new, |vsr_client_id| {
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(vsr_client_id)
+ .into_iter()
+ .map(
+ |(stream_id, topic_id, group_id)| ConsumerGroupInfoResponse {
+ stream_id,
+ topic_id,
+ group_id,
+ },
+ )
+ .collect()
+ });
ClientDetailsResponse {
- client: connected_client_to_response(info),
- consumer_groups: Vec::new(),
+ client: connected_client_to_response(shard, info),
+ consumer_groups,
}
.to_bytes()
});
@@ -692,6 +807,9 @@ async fn handle_non_replicated_request(
GET_CONSUMER_OFFSET_CODE => {
handle_get_consumer_offset(shard, transport_client_id, &request).await;
}
+ SYNC_CONSUMER_GROUP_CODE => {
+ handle_sync_consumer_group(shard, transport_client_id, &request).await;
+ }
_ => handle_default_non_replicated(shard, transport_client_id, code, &request).await,
}
}
@@ -798,6 +916,100 @@ async fn send_unauthenticated_eviction(shard: &Rc<ServerNgShard>, transport_clie
}
}
+/// Per-shard heartbeat verifier: evict connections that have not pinged within
+/// `1.2 x interval`. Mirrors the legacy `verify_heartbeats` periodic task.
+/// Eviction reuses the disconnect path (drops the client from its consumer
+/// groups + rebalances via the replicated `Logout`) and sends a session-
+/// terminal `Eviction(StaleClient)` so the client fails fast and can reconnect.
+#[allow(clippy::future_not_send)]
+pub(crate) async fn run_heartbeat_verifier(
+ shard: Rc<ServerNgShard>,
+ sessions: Rc<RefCell<SessionManager>>,
+ interval: std::time::Duration,
+ stop_rx: shard::Receiver<()>,
+) {
+ // Legacy `MAX_THRESHOLD`: a client is stale once it misses ~1.2 intervals.
+ let max_age = interval.mul_f64(1.2);
+ loop {
+ if stop_rx.try_recv().is_ok() {
+ break;
+ }
+ let stale = sessions
+ .borrow()
+ .collect_stale(max_age, std::time::Instant::now());
+ for transport_client_id in stale {
+ // The heartbeat verifier exists to release a dead client's
+ // consumer-group membership (so the group rebalances off it). A
+ // connection that holds no membership has nothing for the eviction
+ // to clean up; reaping it would only drop a still-usable session
+ // (e.g. an idle admin connection that polls between long gaps),
+ // which the legacy server tolerates. The real transport-disconnect
+ // path still reaps it on socket close. So only evict a stale
+ // connection that is actually a group member.
+ let is_group_member = sessions
+ .borrow()
+ .bound_client_id(transport_client_id)
+ .is_some_and(|vsr_client_id| {
+ !shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(vsr_client_id)
+ .is_empty()
+ });
+ if is_group_member {
+ evict_stale_client(&shard, &sessions, transport_client_id).await;
+ }
+ }
+ compio::time::sleep(interval).await;
+ }
+}
+
+/// Evict one stale connection: drop its session (releasing consumer-group
+/// membership through a replicated `Logout`) and notify the client with a
+/// session-terminal `Eviction(StaleClient)`.
+#[allow(clippy::future_not_send)]
+async fn evict_stale_client(
+ shard: &Rc<ServerNgShard>,
+ sessions: &Rc<RefCell<SessionManager>>,
+ transport_client_id: u128,
+) {
+ let bound = sessions.borrow_mut().remove_connection(transport_client_id);
+ if let Some((vsr_client_id, session)) = bound {
+ submit_disconnect_logout(Rc::clone(shard), vsr_client_id, session);
+ }
+ let ctx = shard.plane.metadata().consensus.as_ref().map_or(
+ consensus::EvictionContext {
+ cluster: 0,
+ view: 0,
+ replica: 0,
+ },
+ consensus::EvictionContext::from_consensus,
+ );
+ let eviction = consensus::build_eviction_message(
+ ctx,
+ transport_client_id,
+ iggy_binary_protocol::EvictionReason::StaleClient,
+ );
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id, eviction.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to send stale-client eviction"
+ );
+ } else {
+ warn!(
+ transport_client_id,
+ "evicted stale client (missed heartbeat)"
+ );
+ }
+}
+
/// Serve `poll_messages`: resolve the partition namespace, run the read on
/// the owning shard ([`shard::IggyShard::partition_read`]), and re-encode
/// the stored batches into the legacy wire `PolledMessages` body.
@@ -841,13 +1053,21 @@ async fn handle_poll_messages(
}
Err(error) => {
// A zero-byte body would panic the SDK's `PolledMessages`
- // decoder; reply the 16-byte empty-poll shape instead.
+ // decoder; reply the 16-byte empty-poll shape instead. A generation
+ // fence (the client's cached assignment is stale after a rebalance)
+ // carries the re-sync sentinel so the SDK re-syncs and retries
+ // rather than treating the empty poll as end-of-partition.
warn!(
transport_client_id,
error = %error,
"poll_messages request rejected; replying empty poll"
);
- empty_polled_messages_body(0)
+ let partition_id = if matches!(error, IggyError::ConsumerGroupPartitionNotOwned(..)) {
+ iggy_common::RESYNC_REQUIRED_PARTITION_SENTINEL
+ } else {
+ 0
+ };
+ empty_polled_messages_body(partition_id)
}
};
send_non_replicated_bytes(shard, request, transport_client_id, body, "poll_messages").await;
@@ -893,6 +1113,54 @@ async fn handle_get_consumer_offset(
.await;
}
+/// Serve `SyncConsumerGroup`: return the requesting member's current partition
+/// assignment + group generation so the client can select partitions locally.
+/// The member is keyed by the connection's bound VSR client id
+/// (`header().client`). An empty body decodes as "no assignment" on the SDK.
+#[allow(clippy::future_not_send)]
+async fn handle_sync_consumer_group(
+ shard: &Rc<ServerNgShard>,
+ transport_client_id: u128,
+ request: &Message<RequestHeader>,
+) {
+ let body = match SyncConsumerGroupRequest::decode_from(request_body(request)) {
+ Ok(wire) => shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_member_assignment(
+ &wire.stream_id,
+ &wire.topic_id,
+ &wire.group_id,
+ request.header().client,
+ )
+ .map_or_else(Bytes::new, |(generation, partitions)| {
+ SyncConsumerGroupResponse {
+ generation,
+ partitions,
+ }
+ .to_bytes()
+ }),
+ Err(error) => {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "sync_consumer_group request rejected; replying empty"
+ );
+ Bytes::new()
+ }
+ };
+ send_non_replicated_bytes(
+ shard,
+ request,
+ transport_client_id,
+ body,
+ "sync_consumer_group",
+ )
+ .await;
+}
+
/// Ack a partition op that cannot be routed (unresolved or never-
/// materialised namespace) with an empty Reply. The SDK connection
/// processes replies in lockstep, so a silent drop wedges every
@@ -980,23 +1248,60 @@ fn empty_polled_messages_body(partition_id: u32) -> Bytes {
type DecodedPollRequest = (IggyNamespace, u32, PollingConsumer, PollingArgs);
+#[allow(clippy::cast_possible_truncation)]
fn decode_poll_request(
shard: &Rc<ServerNgShard>,
request: &Message<RequestHeader>,
) -> Result<DecodedPollRequest, IggyError> {
let wire = PollMessagesRequest::decode_from(request_body(request))
.map_err(|_| IggyError::InvalidCommand)?;
+ let strategy = polling_strategy_from_wire(&wire.strategy)?;
+ let args = PollingArgs::new(strategy, wire.count, wire.auto_commit);
+
+ // Consumer-group poll: the client selects which of its assigned partitions
+ // to read and sends it explicitly. The coordinator FENCES ownership (a stale
+ // client whose partition was reassigned is rejected with
+ // `ConsumerGroupPartitionNotOwned`, prompting a re-sync) and resolves the
+ // group's monotonic id -- the offset key the store rewrite and read path
+ // both use, so `next()` reads back the offset it just committed.
+ if wire.consumer.kind == KIND_CONSUMER_GROUP {
+ let partition_id = wire.partition_id.ok_or(IggyError::InvalidIdentifier)?;
+ let client_id = request.header().client;
+ let group_id = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_fence(
+ &wire.stream_id,
+ &wire.topic_id,
+ &wire.consumer.id,
+ client_id,
+ partition_id,
+ // Poll fence: reject a pending-revoked partition so the source
+ // re-syncs and skips it (it still commits it via the offset fence).
+ true,
+ )
+ .ok_or(IggyError::ConsumerGroupPartitionNotOwned(
+ client_id as u32,
+ partition_id,
+ ))?;
+ let namespace = resolve_partition_namespace(
+ shard,
+ &wire.stream_id,
+ &wire.topic_id,
+ Some(partition_id),
+ )?;
+ #[allow(clippy::cast_possible_truncation)]
+ let consumer = PollingConsumer::ConsumerGroup(group_id as usize, partition_id as usize);
+ return Ok((namespace, partition_id, consumer, args));
+ }
+
let partition_id = wire.partition_id.ok_or(IggyError::InvalidIdentifier)?;
let namespace =
resolve_partition_namespace(shard, &wire.stream_id, &wire.topic_id, Some(partition_id))?;
let consumer = polling_consumer_from_wire(&wire.consumer, partition_id)?;
- let strategy = polling_strategy_from_wire(&wire.strategy)?;
- Ok((
- namespace,
- partition_id,
- consumer,
- PollingArgs::new(strategy, wire.count, wire.auto_commit),
- ))
+ Ok((namespace, partition_id, consumer, args))
}
fn decode_consumer_offset_request(
@@ -1008,7 +1313,22 @@ fn decode_consumer_offset_request(
let partition_id = wire.partition_id.ok_or(IggyError::InvalidIdentifier)?;
let namespace =
resolve_partition_namespace(shard, &wire.stream_id, &wire.topic_id, Some(partition_id))?;
- let consumer = polling_consumer_from_wire(&wire.consumer, partition_id)?;
+ // A group offset is keyed by the group's monotonic id (any client may read
+ // it, member or not), the same key the write path is rewritten to. An
+ // unresolved group (e.g. deleted) has no offset, so the read reports None.
+ let consumer = if wire.consumer.kind == KIND_CONSUMER_GROUP {
+ let group_id = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .resolve_consumer_group_id(&wire.stream_id, &wire.topic_id, &wire.consumer.id)
+ .ok_or(IggyError::InvalidIdentifier)?;
+ #[allow(clippy::cast_possible_truncation)]
+ PollingConsumer::ConsumerGroup(group_id as usize, partition_id as usize)
+ } else {
+ polling_consumer_from_wire(&wire.consumer, partition_id)?
+ };
Ok((namespace, partition_id, consumer))
}
@@ -1030,7 +1350,7 @@ fn polling_consumer_from_wire(
consumer_id,
partition_id as usize,
)),
- 2 => Ok(PollingConsumer::ConsumerGroup(
+ KIND_CONSUMER_GROUP => Ok(PollingConsumer::ConsumerGroup(
consumer_id,
partition_id as usize,
)),
diff --git a/core/server-ng/src/lib.rs b/core/server-ng/src/lib.rs
index bea6b7bc2..090b8d8ad 100644
--- a/core/server-ng/src/lib.rs
+++ b/core/server-ng/src/lib.rs
@@ -20,6 +20,7 @@
pub mod auth;
pub mod bootstrap;
pub mod config_writer;
+pub mod consumer_group;
pub mod dispatch;
pub mod login_register;
pub mod partition_helpers;
diff --git a/core/server-ng/src/partition_reconciler.rs b/core/server-ng/src/partition_reconciler.rs
index 5bd707c5a..10152bb96 100644
--- a/core/server-ng/src/partition_reconciler.rs
+++ b/core/server-ng/src/partition_reconciler.rs
@@ -56,8 +56,10 @@ use ahash::{AHashMap, AHashSet};
use configs::server_ng::ServerNgConfig;
use consensus::{MetadataHandle, PartitionsHandle};
use futures::FutureExt;
+use iggy_common::{ConsumerGroupId, IggyTimestamp};
use metadata::impls::metadata::StreamsFrontend;
use server_common::sharding::{IggyNamespace, ShardId};
+use shard::MetadataSubmit;
use shard::ReconcileOp;
use shard::shards_table::{ShardsTable, calculate_shard_assignment};
use shard::{Receiver, Sender};
@@ -65,7 +67,7 @@ use std::cell::{Cell, RefCell};
use std::rc::Rc;
use std::sync::Arc;
use std::time::{Duration, Instant};
-use tracing::{debug, error, trace};
+use tracing::{debug, error, trace, warn};
const BACKOFF_BASE: Duration = Duration::from_secs(1);
const BACKOFF_MAX: Duration = Duration::from_mins(1);
@@ -237,6 +239,9 @@ struct PassCounters {
backoff_skipped: usize,
/// Stale incarnations (slab-key reuse) torn down for rebuild.
stale: usize,
+ /// Consumer-group offsets reclaimed for groups deleted while their topic
+ /// survived (a bare `DeleteConsumerGroup`, not a topic/stream delete).
+ cg_offsets_purged: usize,
}
impl PassCounters {
@@ -247,6 +252,7 @@ impl PassCounters {
+ self.removed_routed
+ self.backoff_skipped
+ self.stale
+ + self.cg_offsets_purged
}
}
@@ -258,6 +264,12 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
let shard_id = ctx.shard.id;
let revision = current_revision(ctx);
+ // Cooperative-revocation completion runs every tick, before the fast-skip:
+ // a timeout fires on wall-clock and a drain on partition-offset state, and
+ // neither bumps `Streams::revision`, so the skip would otherwise starve an
+ // idle group's pending revocations forever. Cheap no-op when none pending.
+ reconcile_pending_revocations(ctx);
+
// Fast-skip: committed partition set unchanged since the last
// fully-converged pass and no backoff retry due, so the O(N) diff is
// pure waste. Safe because reconcile is level-triggered: the next
@@ -280,6 +292,7 @@ async fn reconcile_once(ctx: &ReconcilerCtx) -> bool {
reconcile_additions(ctx, target, &mut counters).await;
reconcile_removals(ctx, &target_set, &mut counters).await;
+ reconcile_consumer_group_offsets(ctx, &mut counters).await;
let local_set: AHashSet<IggyNamespace> =
ctx.shard.plane.partitions().namespaces().copied().collect();
@@ -563,6 +576,136 @@ async fn tear_down_owned_partition(
counters.removed_local += 1;
}
+/// Reclaim consumer-group offsets left behind by a `DeleteConsumerGroup` whose
+/// topic still exists (a topic/stream delete already drops the whole partition
+/// directory, offsets included). For each owned partition, any stored
+/// consumer-group offset whose group id is no longer present in the topic's
+/// committed metadata is removed (in-memory entry + persisted file). Monotonic,
+/// never-reused group ids make this purely reclamation -- a recreated group
+/// gets a fresh id and never reads a dead group's offset -- so it is safe to do
+/// lazily on the reconcile pass rather than synchronously on delete.
+async fn reconcile_consumer_group_offsets(ctx: &ReconcilerCtx, counters: &mut PassCounters) {
+ let live_groups = snapshot_topic_live_groups(ctx);
+ let partitions = ctx.shard.plane.partitions();
+ let owned: Vec<IggyNamespace> = partitions.namespaces().copied().collect();
+ for ns in owned {
+ let Some(partition) = partitions.get_by_ns(&ns) else {
+ continue;
+ };
+ let stored = partition.consumer_group_offset_ids();
+ if stored.is_empty() {
+ continue;
+ }
+ let live = live_groups.get(&(ns.stream_id(), ns.topic_id()));
+ for group_id in stored {
+ let still_live = live.is_some_and(|set| set.contains(&group_id));
+ if still_live {
+ continue;
+ }
+ if let Err(err) = partition.delete_consumer_group_offset(group_id).await {
+ warn!(
+ shard = ctx.shard.id,
+ ns_raw = ns.inner(),
+ group_id,
+ error = %err,
+ "reconciler failed to reclaim deleted consumer-group offset"
+ );
+ continue;
+ }
+ counters.cg_offsets_purged += 1;
+ }
+ }
+}
+
+/// Complete cooperative consumer-group revocations whose source member has
+/// drained the partition (`committed >= last_polled`), was never polled, or
+/// timed out. Reads pending revocations from metadata + local partition offset
+/// state, then submits a `CompleteRevocation` op to shard 0 (the metadata
+/// consensus owner). Idempotent + fire-and-forget: a not-yet-completable or
+/// transiently-failed revocation is retried next pass.
+#[allow(clippy::cast_possible_truncation)]
+fn reconcile_pending_revocations(ctx: &ReconcilerCtx) {
+ let streams = ctx.shard.plane.metadata().mux_stm.streams();
+ // O(1) fast-skip before the walk: `consumer_group_pending_revocations`
+ // allocates a vec and walks every stream/topic/group/member, and the
+ // reconciler hits this every tick. `has_pending_revocations` reads the
+ // maintained counter, so the common (nothing-pending) case pays nothing.
+ if !streams.has_pending_revocations() {
+ return;
+ }
+ let pending = streams.consumer_group_pending_revocations();
+ if pending.is_empty() {
+ return;
+ }
+ let partitions = ctx.shard.plane.partitions();
+ let now = IggyTimestamp::now().as_micros();
+ let timeout = ctx.config.consumer_group.rebalancing_timeout.as_micros();
+ for (stream_id, topic_id, group_id, source_client_id, partition_id, created_at) in pending {
+ let ns = IggyNamespace::new(stream_id as usize, topic_id as usize, partition_id as usize);
+ // The partition lives on its owner shard; only that shard's reconciler
+ // can read its offsets. Other shards skip (the owner completes it).
+ let Some(partition) = partitions.get_by_ns(&ns) else {
+ continue;
+ };
+ let key = ConsumerGroupId(group_id as usize);
+ let last_polled = partition
+ .last_polled_offsets
+ .pin()
+ .get(&key)
+ .map(|offset| offset.offset.load(std::sync::atomic::Ordering::Relaxed));
+ let committed = partition
+ .consumer_group_offsets
+ .pin()
+ .get(&key)
+ .map(|offset| offset.offset.load(std::sync::atomic::Ordering::Relaxed));
+ let timed_out = now.saturating_sub(created_at) >= timeout;
+ // None: never polled -> nothing in flight, hand off now. Some(polled):
+ // only safe once the source committed what it was served (or timeout).
+ let completable =
+ last_polled.is_none_or(|polled| committed.is_some_and(|c| c >= polled) || timed_out);
+ if !completable {
+ continue;
+ }
+ let (reply, _rx) = shard::channel::<Option<u64>>(1);
+ ctx.shard
+ .forward_metadata_submit(MetadataSubmit::CompleteRevocation {
+ stream_id,
+ topic_id,
+ group_id,
+ source_client_id,
+ partition_id,
+ reply,
+ });
+ }
+}
+
+/// `(stream_id, topic_id) -> live consumer-group offset keys` from committed
+/// metadata. The partition plane keys a group's offset by the monotonic group
+/// id (the store path is rewritten to it; the read path resolves it), so the
+/// live-set carries those ids too -- otherwise the reconciler would treat every
+/// live offset as orphaned and purge it.
+fn snapshot_topic_live_groups(ctx: &ReconcilerCtx) -> AHashMap<(usize, usize), AHashSet<u64>> {
+ ctx.shard.plane.metadata().mux_stm.streams().read(|inner| {
+ let mut map: AHashMap<(usize, usize), AHashSet<u64>> = AHashMap::new();
+ for (_, stream) in &inner.items {
+ for (topic_id, topic) in &stream.topics {
+ if topic.consumer_groups.is_empty() {
+ continue;
+ }
+ map.insert(
+ (stream.id, topic_id),
+ topic
+ .consumer_groups
+ .values()
+ .map(|group| group.id)
+ .collect(),
+ );
+ }
+ }
+ map
+ })
+}
+
/// Committed `(namespace, created_revision)` pairs. The epoch lets the
/// additions pass detect a stale local incarnation after slab-key reuse
/// without an `Arc<TopicStats>` clone per partition; stats are fetched
@@ -645,7 +788,6 @@ mod tests {
use metadata::MuxStateMachine;
use metadata::impls::metadata::IggySnapshot;
use metadata::stm::StateMachine;
- use metadata::stm::consumer_group::ConsumerGroups;
use metadata::stm::stream::Streams;
use metadata::stm::user::Users;
use partitions::{IggyPartitions, PartitionsConfig};
@@ -659,7 +801,7 @@ mod tests {
use std::time::Instant;
use tempfile::TempDir;
- type TestMux = MuxStateMachine<iggy_common::variadic!(Users, Streams, ConsumerGroups)>;
+ type TestMux = MuxStateMachine<iggy_common::variadic!(Users, Streams)>;
type TestShard = IggyShard<
Rc<IggyMessageBus>,
journal::prepare_journal::PrepareJournal,
@@ -774,6 +916,60 @@ mod tests {
.expect("DeleteStream apply succeeds");
}
+ fn seed_create_consumer_group(
+ mux: &TestMux,
+ op: u64,
+ stream_id: u32,
+ topic_id: u32,
+ name: &str,
+ ) {
+ use iggy_binary_protocol::requests::consumer_groups::CreateConsumerGroupRequest;
+ let req = CreateConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(stream_id),
+ topic_id: WireIdentifier::numeric(topic_id),
+ name: WireName::new(name).expect("test group name fits WireName"),
+ };
+ mux.update(build_prepare(op, Operation::CreateConsumerGroup, &req))
+ .expect("CreateConsumerGroup apply succeeds");
+ }
+
+ fn seed_delete_consumer_group(
+ mux: &TestMux,
+ op: u64,
+ stream_id: u32,
+ topic_id: u32,
+ group_id: u32,
+ ) {
+ use iggy_binary_protocol::requests::consumer_groups::DeleteConsumerGroupRequest;
+ let req = DeleteConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(stream_id),
+ topic_id: WireIdentifier::numeric(topic_id),
+ group_id: WireIdentifier::numeric(group_id),
+ };
+ mux.update(build_prepare(op, Operation::DeleteConsumerGroup, &req))
+ .expect("DeleteConsumerGroup apply succeeds");
+ }
+
+ fn seed_join_consumer_group(
+ mux: &TestMux,
+ op: u64,
+ stream_id: u32,
+ topic_id: u32,
+ group_id: u32,
+ client_id: u128,
+ ) {
+ use metadata::stm::consumer_group::JoinConsumerGroupRequest;
+ let req = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(stream_id),
+ topic_id: WireIdentifier::numeric(topic_id),
+ group_id: WireIdentifier::numeric(group_id),
+ client_id,
+ in_flight: Vec::new(),
+ };
+ mux.update(build_prepare(op, Operation::JoinConsumerGroup, &req))
+ .expect("JoinConsumerGroup apply succeeds");
+ }
+
fn test_config(tmp: &TempDir) -> ServerNgConfig {
let mut cfg = ServerNgConfig::default();
// `SystemConfig` is not `Clone`, so `Arc::make_mut` is out; build a
@@ -1438,4 +1634,188 @@ mod tests {
"defer must not re-drive teardown: the directory must remain"
);
}
+
+ /// A bare `DeleteConsumerGroup` (topic survives) leaves the group's offsets
+ /// on the partition. The reconciler must reclaim a deleted group's offset
+ /// while leaving a still-live group's offset untouched.
+ #[compio::test]
+ async fn reconcile_reclaims_offsets_of_deleted_consumer_group() {
+ use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset};
+
+ let tmp = TempDir::new().expect("tempdir for system path");
+ let config = test_config(&tmp);
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-cg");
+ seed_topic(&mux, 2, 0, "topic-cg", vec![assignment(0, 1)]);
+
+ let shard = build_test_shard(0, &config, mux);
+ let ctx = make_ctx(Rc::clone(&shard), 1, Rc::new(config));
+ reconcile_pass(&ctx).await;
+
+ let ns = IggyNamespace::new(0, 0, 0);
+ assert!(shard.plane.partitions().contains(&ns));
+
+ // Two groups: "dead" gets id 1, "live" gets id 2 (per-topic monotonic).
+ let stm = &shard.plane.metadata().mux_stm;
+ seed_create_consumer_group(stm, 3, 0, 0, "dead");
+ seed_create_consumer_group(stm, 4, 0, 0, "live");
+
+ // Offsets are keyed by the monotonic group id (the id the store path is
+ // rewritten to and the read path / live-set resolve), not the name hash.
+ let dead_key: u32 = 1;
+ let live_key: u32 = 2;
+ {
+ let partitions = shard.plane.partitions();
+ let partition = partitions.get_by_ns(&ns).expect("partition materialised");
+ partition.consumer_group_offsets.pin().insert(
+ ConsumerGroupId(dead_key as usize),
+ ConsumerOffset::new(ConsumerKind::ConsumerGroup, dead_key, 7, String::new()),
+ );
+ partition.consumer_group_offsets.pin().insert(
+ ConsumerGroupId(live_key as usize),
+ ConsumerOffset::new(ConsumerKind::ConsumerGroup, live_key, 9, String::new()),
+ );
+ }
+
+ // Delete the "dead" group (id 1); "live" (id 2) stays.
+ seed_delete_consumer_group(stm, 5, 0, 0, 1);
+ reconcile_pass(&ctx).await;
+
+ let partitions = shard.plane.partitions();
+ let partition = partitions
+ .get_by_ns(&ns)
+ .expect("partition still materialised");
+ let mut ids = partition.consumer_group_offset_ids();
+ ids.sort_unstable();
+ assert_eq!(
+ ids,
+ vec![u64::from(live_key)],
+ "deleted group's offset reclaimed; live group's offset retained"
+ );
+ }
+
+ /// A partition-count change must re-run consumer-group assignment: a new
+ /// partition gets assigned, a removed one is dropped. Pure metadata test --
+ /// the assignment lives in the Streams STM.
+ #[compio::test]
+ async fn create_delete_partitions_reassigns_consumer_group() {
+ use metadata::impls::metadata::StreamsFrontend;
+
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-rp");
+ seed_topic(
+ &mux,
+ 2,
+ 0,
+ "topic-rp",
+ vec![assignment(0, 1), assignment(1, 2)],
+ );
+ seed_create_consumer_group(&mux, 3, 0, 0, "cg");
+ // Single member owns every partition (group id 1, the first in topic).
+ seed_join_consumer_group(&mux, 4, 0, 0, 1, 100);
+
+ let group = WireIdentifier::numeric(1);
+ let stream = WireIdentifier::numeric(0);
+ let topic = WireIdentifier::numeric(0);
+ let assigned = |mux: &TestMux| -> Vec<u32> {
+ let (_, mut partitions) = mux
+ .streams()
+ .consumer_group_member_assignment(&stream, &topic, &group, 100)
+ .expect("member assignment present");
+ partitions.sort_unstable();
+ partitions
+ };
+ assert_eq!(
+ assigned(&mux),
+ vec![0, 1],
+ "joined member owns both partitions"
+ );
+
+ // Add one partition (request-relative id 0 rebases to absolute id 2).
+ mux.update(build_prepare(
+ 5,
+ Operation::CreatePartitionsWithAssignments,
+ &CreatePartitionsWithAssignmentsRequest {
+ request: CreatePartitionsRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ },
+ partitions: vec![assignment(0, 3)],
+ },
+ ))
+ .expect("CreatePartitions apply succeeds");
+ assert_eq!(
+ assigned(&mux),
+ vec![0, 1, 2],
+ "added partition must be reassigned to the member"
+ );
+
+ // Remove one partition; the member drops the highest id.
+ mux.update(build_prepare(
+ 6,
+ Operation::DeletePartitions,
+ &iggy_binary_protocol::requests::partitions::DeletePartitionsRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ },
+ ))
+ .expect("DeletePartitions apply succeeds");
+ assert_eq!(
+ assigned(&mux),
+ vec![0, 1],
+ "removed partition must be dropped from the assignment"
+ );
+ }
+
+ /// A disconnect (`remove_consumer_group_member`) drops the client from
+ /// every group it joined and rebalances its partitions onto the survivors.
+ #[compio::test]
+ async fn disconnect_removes_member_from_groups_and_rebalances() {
+ use metadata::impls::metadata::StreamsFrontend;
+
+ let mux = TestMux::default();
+ seed_stream(&mux, 1, "stream-dc");
+ seed_topic(
+ &mux,
+ 2,
+ 0,
+ "topic-dc",
+ vec![assignment(0, 1), assignment(1, 2)],
+ );
+ seed_create_consumer_group(&mux, 3, 0, 0, "cg"); // group id 1
+ seed_join_consumer_group(&mux, 4, 0, 0, 1, 100);
+ seed_join_consumer_group(&mux, 5, 0, 0, 1, 200);
+
+ let stream = WireIdentifier::numeric(0);
+ let topic = WireIdentifier::numeric(0);
+ let group = WireIdentifier::numeric(1);
+ let assigned = |client: u128| -> Option<Vec<u32>> {
+ mux.streams()
+ .consumer_group_member_assignment(&stream, &topic, &group, client)
+ .map(|(_, mut partitions)| {
+ partitions.sort_unstable();
+ partitions
+ })
+ };
+ // Two members, two partitions: each owns one.
+ assert_eq!(assigned(100).map(|p| p.len()), Some(1));
+ assert_eq!(assigned(200).map(|p| p.len()), Some(1));
+
+ // Client 100 disconnects.
+ mux.streams()
+ .remove_consumer_group_member(100, iggy_common::IggyTimestamp::default());
+
+ assert_eq!(
+ assigned(100),
+ None,
+ "disconnected client must leave the group"
+ );
+ assert_eq!(
+ assigned(200),
+ Some(vec![0, 1]),
+ "survivor must take over the disconnected member's partitions"
+ );
+ }
}
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index e1102ad49..7ca8cfdce 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -29,8 +29,13 @@ use crate::wire::{transport_kind_to_wire, usize_to_u32};
use bytes::Bytes;
use consensus::{MetadataHandle, VsrConsensus};
use iggy_binary_protocol::codes::{
- GET_CLUSTER_METADATA_CODE, GET_STATS_CODE, GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE,
- GET_TOPICS_CODE, GET_USER_CODE, GET_USERS_CODE,
+ GET_CLUSTER_METADATA_CODE, GET_CONSUMER_GROUP_CODE, GET_CONSUMER_GROUPS_CODE, GET_STATS_CODE,
+ GET_STREAM_CODE, GET_STREAMS_CODE, GET_TOPIC_CODE, GET_TOPICS_CODE, GET_USER_CODE,
+ GET_USERS_CODE,
+};
+use iggy_binary_protocol::primitives::consumer::WireConsumer;
+use iggy_binary_protocol::requests::consumer_groups::{
+ GetConsumerGroupRequest, GetConsumerGroupsRequest,
};
use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffset2Request, DeleteConsumerOffsetRequest, StoreConsumerOffset2Request,
@@ -42,7 +47,9 @@ use iggy_binary_protocol::requests::streams::{GetStreamRequest, GetStreamsReques
use iggy_binary_protocol::requests::topics::{GetTopicRequest, GetTopicsRequest};
use iggy_binary_protocol::requests::users::GetUserRequest;
use iggy_binary_protocol::responses::clients::client_response::ClientResponse;
+use iggy_binary_protocol::responses::clients::client_response::ConsumerGroupInfoResponse;
use iggy_binary_protocol::responses::clients::get_client::ClientDetailsResponse;
+use iggy_binary_protocol::responses::consumer_groups::GetConsumerGroupsResponse;
use iggy_binary_protocol::responses::personal_access_tokens::RawPersonalAccessTokenResponse;
use iggy_binary_protocol::responses::streams::StreamResponse;
use iggy_binary_protocol::responses::streams::get_stream::{
@@ -59,8 +66,8 @@ use iggy_binary_protocol::responses::users::get_user::UserDetailsResponse;
use iggy_binary_protocol::responses::users::get_users::GetUsersResponse;
use iggy_binary_protocol::responses::users::user_response::UserResponse;
use iggy_binary_protocol::{
- Command2, GenericHeader, Operation, ReplyHeader, RequestHeader, WireDecode, WireEncode,
- WireIdentifier, WireName, WirePartitioning,
+ Command2, GenericHeader, KIND_CONSUMER_GROUP, Operation, ReplyHeader, RequestHeader,
+ WireDecode, WireEncode, WireIdentifier, WireName, WirePartitioning,
};
use iggy_common::IggyError;
use metadata::impls::metadata::StreamsFrontend;
@@ -72,15 +79,16 @@ use shard::ConnectedClientInfo;
use std::cell::RefCell;
use std::rc::Rc;
-/// Build the `get_me` reply for the requesting connection, sourced
-/// entirely from the per-shard [`SessionManager`] (`user_id`, transport
-/// kind, peer address) -- no message-bus lookup. `consumer_groups` is
-/// empty (see [`client_record_to_response`]).
+/// Build the `get_me` reply for the requesting connection. Identity
+/// (`user_id`, transport kind, peer address) comes from the per-shard
+/// [`SessionManager`]; the `consumer_groups` list is read from the
+/// (replicated) consumer-group STM by the connection's bound VSR client id.
pub(crate) fn build_get_me_response(
+ shard: &Rc<ServerNgShard>,
sessions: &Rc<RefCell<SessionManager>>,
transport_client_id: u128,
) -> ClientDetailsResponse {
- let client = sessions
+ let mut client = sessions
.borrow()
.client_record(transport_client_id)
.map_or_else(
@@ -98,11 +106,41 @@ pub(crate) fn build_get_me_response(
consumer_groups_count: 0,
}
},
- |record| connected_client_to_response(&record),
+ |record| connected_client_to_response(shard, &record),
);
+
+ // The wire `consumer_groups` list keys off the connection's bound VSR
+ // client id (the same id recorded as a group member by the replicated
+ // Join op), not the transport id.
+ let consumer_groups = sessions
+ .borrow()
+ .get_session(transport_client_id)
+ .map(|(vsr_client_id, _)| {
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(vsr_client_id)
+ })
+ .unwrap_or_default()
+ .into_iter()
+ .map(
+ |(stream_id, topic_id, group_id)| ConsumerGroupInfoResponse {
+ stream_id,
+ topic_id,
+ group_id,
+ },
+ )
+ .collect::<Vec<_>>();
+
+ #[allow(clippy::cast_possible_truncation)]
+ {
+ client.consumer_groups_count = consumer_groups.len() as u32;
+ }
ClientDetailsResponse {
client,
- consumer_groups: Vec::new(),
+ consumer_groups,
}
}
@@ -110,12 +148,24 @@ pub(crate) fn build_get_me_response(
/// `SessionManager` or a `get_clients` gather) into the wire
/// [`ClientResponse`]. Shared by `get_me`, `get_clients`, and `get_client`.
///
-/// TODO(consumer-group-membership): `consumer_groups_count` is always 0
-/// -- server-ng does not yet track which consumer groups a connection
-/// joined. Populate once the partition-reconciliation / consumer-group
-/// rebalancing work lands; until then `get_me` / `get_clients` report no
-/// memberships.
-pub(crate) fn connected_client_to_response(info: &ConnectedClientInfo) -> ClientResponse {
+/// `consumer_groups_count` is resolved from the connection's bound VSR client
+/// id against the replicated `Streams` STM (memberships are keyed by VSR id, not
+/// transport id). Connections that never bound (pre-register) count 0.
+pub(crate) fn connected_client_to_response(
+ shard: &Rc<ServerNgShard>,
+ info: &ConnectedClientInfo,
+) -> ClientResponse {
+ let consumer_groups_count = info.vsr_client_id.map_or(0, |vsr_client_id| {
+ #[allow(clippy::cast_possible_truncation)]
+ let count = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(vsr_client_id)
+ .len() as u32;
+ count
+ });
// The transport client id is a u128 `(shard << 112) | seq`; the wire
// `client_id` is the u32 seq tail.
#[allow(clippy::cast_possible_truncation)]
@@ -124,14 +174,75 @@ pub(crate) fn connected_client_to_response(info: &ConnectedClientInfo) -> Client
user_id: info.user_id.unwrap_or(u32::MAX),
transport: transport_kind_to_wire(info.transport),
address: info.address.to_string(),
- consumer_groups_count: 0,
+ consumer_groups_count,
+ }
+}
+
+/// Fence a consumer-group offset commit/delete: a group consumer may only
+/// touch the offset of a partition it currently owns. `Ok` for individual
+/// consumers (no fence) and for owned group partitions; `Err` otherwise so a
+/// stale client re-syncs instead of corrupting the shared group offset.
+fn fence_group_offset(
+ shard: &Rc<ServerNgShard>,
+ consumer: &WireConsumer,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ partition_id: Option<u32>,
+ client_id: u128,
+) -> Result<(), IggyError> {
+ if consumer.kind != KIND_CONSUMER_GROUP {
+ return Ok(());
}
+ let partition_id = partition_id.ok_or(IggyError::InvalidIdentifier)?;
+ #[allow(clippy::cast_possible_truncation)]
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ // Commit fence: allow a pending-revoked partition (the source commits it
+ // to drain the cooperative handoff), so `require_pollable = false`.
+ .consumer_group_fence(
+ stream_id,
+ topic_id,
+ &consumer.id,
+ client_id,
+ partition_id,
+ false,
+ )
+ .map(|_| ())
+ .ok_or(IggyError::ConsumerGroupPartitionNotOwned(
+ client_id as u32,
+ partition_id,
+ ))
+}
+
+/// Fence a consumer-group offset op then resolve its target partition
+/// namespace. Shared by the four `Store`/`Delete` consumer-offset arms.
+fn fence_and_resolve_offset_namespace(
+ shard: &Rc<ServerNgShard>,
+ consumer: &WireConsumer,
+ stream_id: &WireIdentifier,
+ topic_id: &WireIdentifier,
+ partition_id: Option<u32>,
+ client_id: u128,
+) -> Result<IggyNamespace, IggyError> {
+ fence_group_offset(
+ shard,
+ consumer,
+ stream_id,
+ topic_id,
+ partition_id,
+ client_id,
+ )?;
+ resolve_partition_namespace(shard, stream_id, topic_id, partition_id)
}
pub(crate) fn resolve_partition_request_namespace(
shard: &Rc<ServerNgShard>,
operation: Operation,
body: &[u8],
+ client_id: u128,
) -> Result<u64, IggyError> {
let namespace = match operation {
Operation::SendMessages => {
@@ -153,41 +264,49 @@ pub(crate) fn resolve_partition_request_namespace(
Operation::StoreConsumerOffset => {
let request = StoreConsumerOffsetRequest::decode_from(body)
.map_err(|_| IggyError::InvalidCommand)?;
- resolve_partition_namespace(
+ fence_and_resolve_offset_namespace(
shard,
+ &request.consumer,
&request.stream_id,
&request.topic_id,
request.partition_id,
+ client_id,
)?
}
Operation::DeleteConsumerOffset => {
let request = DeleteConsumerOffsetRequest::decode_from(body)
.map_err(|_| IggyError::InvalidCommand)?;
- resolve_partition_namespace(
+ fence_and_resolve_offset_namespace(
shard,
+ &request.consumer,
&request.stream_id,
&request.topic_id,
request.partition_id,
+ client_id,
)?
}
Operation::StoreConsumerOffset2 => {
let request = StoreConsumerOffset2Request::decode_from(body)
.map_err(|_| IggyError::InvalidCommand)?;
- resolve_partition_namespace(
+ fence_and_resolve_offset_namespace(
shard,
+ &request.consumer,
&request.stream_id,
&request.topic_id,
request.partition_id,
+ client_id,
)?
}
Operation::DeleteConsumerOffset2 => {
let request = DeleteConsumerOffset2Request::decode_from(body)
.map_err(|_| IggyError::InvalidCommand)?;
- resolve_partition_namespace(
+ fence_and_resolve_offset_namespace(
shard,
+ &request.consumer,
&request.stream_id,
&request.topic_id,
request.partition_id,
+ client_id,
)?
}
Operation::DeleteSegments => {
@@ -209,8 +328,22 @@ fn resolve_send_messages_namespace(
shard: &Rc<ServerNgShard>,
header: &SendMessagesHeader,
) -> Result<IggyNamespace, IggyError> {
- let WirePartitioning::PartitionId(partition_id) = header.partitioning else {
- return Err(IggyError::FeatureUnavailable);
+ let partition_id = match &header.partitioning {
+ WirePartitioning::PartitionId(partition_id) => *partition_id,
+ WirePartitioning::Balanced => shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .next_balanced_partition(&header.stream_id, &header.topic_id)
+ .ok_or(IggyError::InvalidIdentifier)?,
+ WirePartitioning::MessagesKey(key) => shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .partition_by_messages_key(&header.stream_id, &header.topic_id, key)
+ .ok_or(IggyError::InvalidIdentifier)?,
};
resolve_partition_namespace(
shard,
@@ -291,6 +424,32 @@ pub(crate) fn build_non_replicated_response(
})
})
}
+ GET_CONSUMER_GROUP_CODE => {
+ let request = GetConsumerGroupRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let response = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_details(&request.stream_id, &request.topic_id, &request.group_id);
+ Ok(response.map_or(NonReplicatedResponse::Empty, |response| {
+ NonReplicatedResponse::Bytes(response.to_bytes())
+ }))
+ }
+ GET_CONSUMER_GROUPS_CODE => {
+ let request = GetConsumerGroupsRequest::decode_from(body)
+ .map_err(|_| IggyError::InvalidCommand)?;
+ let groups = shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_list(&request.stream_id, &request.topic_id);
+ Ok(groups.map_or(NonReplicatedResponse::Empty, |groups| {
+ NonReplicatedResponse::Bytes(GetConsumerGroupsResponse { groups }.to_bytes())
+ }))
+ }
_ => match iggy_binary_protocol::dispatch::lookup_command(code) {
Some(meta) if !meta.is_replicated() => Ok(NonReplicatedResponse::Empty),
Some(_) => Err(IggyError::FeatureUnavailable),
@@ -346,12 +505,14 @@ fn build_stats_response(shard: &Rc<ServerNgShard>) -> Result<StatsResponse, Iggy
messages_count,
))
})?;
- let consumer_groups_count = shard
- .plane
- .metadata()
- .mux_stm
- .consumer_groups()
- .read(|groups| usize_to_u32(groups.items.len()))?;
+ let consumer_groups_count = usize_to_u32(
+ shard
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_count(),
+ )?;
Ok(StatsResponse {
process_id: std::process::id(),
diff --git a/core/server-ng/src/session_manager.rs b/core/server-ng/src/session_manager.rs
index e7fceb96f..560ec8cd3 100644
--- a/core/server-ng/src/session_manager.rs
+++ b/core/server-ng/src/session_manager.rs
@@ -29,6 +29,7 @@ use message_bus::installer::conn_info::ClientTransportKind;
use shard::ConnectedClientInfo;
use std::collections::HashMap;
use std::net::SocketAddr;
+use std::time::{Duration, Instant};
/// Connection lifecycle states.
///
@@ -62,6 +63,9 @@ pub struct Connection {
pub address: SocketAddr,
pub transport: ClientTransportKind,
pub state: ConnectionState,
+ /// Last time the client proved liveness (a `ping`). The heartbeat
+ /// verifier evicts connections stale past the configured threshold.
+ pub last_heartbeat: Instant,
}
/// Bridges transport connections to consensus sessions.
@@ -99,11 +103,47 @@ impl SessionManager {
address: SocketAddr,
transport: ClientTransportKind,
) {
- self.connections.entry(connection_id).or_insert(Connection {
- address,
- transport,
- state: ConnectionState::Connected,
- });
+ self.connections
+ .entry(connection_id)
+ .or_insert_with(|| Connection {
+ address,
+ transport,
+ state: ConnectionState::Connected,
+ last_heartbeat: Instant::now(),
+ });
+ }
+
+ /// Record a liveness heartbeat (`ping`) for a connection, resetting its
+ /// staleness clock. No-op for an unknown connection.
+ pub fn record_heartbeat(&mut self, connection_id: u128) {
+ if let Some(conn) = self.connections.get_mut(&connection_id) {
+ conn.last_heartbeat = Instant::now();
+ }
+ }
+
+ /// Connection ids whose last heartbeat is older than `max_age` -- the
+ /// stale set the heartbeat verifier evicts. Only `Bound`/`Authenticated`
+ /// connections are considered (a freshly-`Connected` socket mid-handshake
+ /// is left alone until it authenticates).
+ #[must_use]
+ pub fn collect_stale(&self, max_age: Duration, now: Instant) -> Vec<u128> {
+ self.connections
+ .iter()
+ .filter(|(_, conn)| !matches!(conn.state, ConnectionState::Connected))
+ .filter(|(_, conn)| now.duration_since(conn.last_heartbeat) > max_age)
+ .map(|(&id, _)| id)
+ .collect()
+ }
+
+ /// The consensus client id a connection is bound to, if any. The heartbeat
+ /// verifier reads it to look up consumer-group membership before deciding
+ /// whether an eviction would actually release anything.
+ #[must_use]
+ pub fn bound_client_id(&self, connection_id: u128) -> Option<u128> {
+ match self.connections.get(&connection_id)?.state {
+ ConnectionState::Bound { client_id, .. } => Some(client_id),
+ ConnectionState::Authenticated { .. } | ConnectionState::Connected => None,
+ }
}
/// Remove a connection (disconnect). Cleans up the reverse index if bound.
@@ -287,8 +327,13 @@ const fn record_from(connection_id: u128, conn: &Connection) -> ConnectedClientI
}
ConnectionState::Connected => None,
};
+ let vsr_client_id = match conn.state {
+ ConnectionState::Bound { client_id, .. } => Some(client_id),
+ ConnectionState::Authenticated { .. } | ConnectionState::Connected => None,
+ };
ConnectedClientInfo {
client_id: connection_id,
+ vsr_client_id,
user_id,
transport: conn.transport,
address: conn.address,
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index 0ecd7b76b..7c89aeda1 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -160,6 +160,19 @@ pub enum MetadataSubmit {
request: Message<GenericHeader>,
reply: Sender<Option<Message<GenericHeader>>>,
},
+ /// A shard's partition reconciler asks shard 0 to complete a cooperative
+ /// consumer-group revocation (the source drained the partition or it timed
+ /// out). Server-originated: shard 0 proposes it through metadata consensus
+ /// with no client session. Fire-and-forget + idempotent -- `reply` carries
+ /// the commit op (or `None` on a transient submit failure) for logging only.
+ CompleteRevocation {
+ stream_id: u32,
+ topic_id: u32,
+ group_id: u64,
+ source_client_id: u128,
+ partition_id: u32,
+ reply: Sender<Option<u64>>,
+ },
}
/// Handler shard 0 runs for an inbound [`MetadataSubmit`].
@@ -180,6 +193,10 @@ pub struct ConnectedClientInfo {
/// Transport (coordinator-minted) client id; top 16 bits are the home
/// shard. The wire `client_id` is the `u32` seq tail.
pub client_id: u128,
+ /// Bound VSR client id, if the connection completed register. Keys the
+ /// connection to its consumer-group memberships (stored by VSR id, not
+ /// transport id).
+ pub vsr_client_id: Option<u128>,
pub user_id: Option<u32>,
pub transport: ClientTransportKind,
pub address: std::net::SocketAddr,
@@ -207,6 +224,22 @@ pub enum PartitionRead {
ConsumerOffset {
consumer: PollingConsumer,
},
+ /// Cooperative-rebalance classification: the group's last-polled and
+ /// committed offsets on this partition, so the join enrichment can tell an
+ /// in-flight partition (committed < last-polled) from a never-polled/drained
+ /// one. `group_id` is the monotonic consumer-group id (offset key).
+ GroupOffsetState {
+ group_id: u64,
+ },
+ /// Drop the group's ephemeral `last_polled` mark on this partition. The
+ /// join-time gather issues this when it finds an uncommitted `last_polled`
+ /// for a partition no live member owns: the residue of a since-removed
+ /// member (reconnect). Clearing it stops a later join in the same restart
+ /// from misreading the dead mark as a live in-flight hold. `group_id` is the
+ /// monotonic consumer-group id (offset key).
+ ClearGroupLastPolled {
+ group_id: u64,
+ },
}
/// Reply to a [`PartitionRead`].
@@ -220,6 +253,14 @@ pub enum PartitionReadReply {
stored: Option<u64>,
current_offset: u64,
},
+ /// Reply to [`PartitionRead::GroupOffsetState`]: the group's last-polled and
+ /// committed offsets on this partition (each `None` if absent).
+ GroupOffsetState {
+ last_polled: Option<u64>,
+ committed: Option<u64>,
+ },
+ /// Acknowledges a [`PartitionRead::ClearGroupLastPolled`].
+ Ack,
/// The owning shard has no materialised partition for the namespace
/// (unknown, tombstoned, or mid-reconcile). Callers surface an error
/// instead of an empty result.
diff --git a/core/simulator/src/deps.rs b/core/simulator/src/deps.rs
index 4a998f731..63087cb10 100644
--- a/core/simulator/src/deps.rs
+++ b/core/simulator/src/deps.rs
@@ -19,7 +19,6 @@ use iggy_binary_protocol::PrepareHeader;
use iggy_common::variadic;
use journal::{Journal, JournalHandle, Storage};
use metadata::MuxStateMachine;
-use metadata::stm::consumer_group::ConsumerGroups;
use metadata::stm::stream::Streams;
use metadata::stm::user::Users;
use server_common::{Message, iobuf::Owned};
@@ -163,4 +162,4 @@ impl JournalHandle for SimJournal<MemStorage> {
pub struct SimSnapshot {}
/// Type alias for simulator state machine
-pub type SimMuxStateMachine = MuxStateMachine<variadic!(Users, Streams, ConsumerGroups)>;
+pub type SimMuxStateMachine = MuxStateMachine<variadic!(Users, Streams)>;
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 2dc02436b..7e44fefee 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -21,7 +21,6 @@ use consensus::{LocalPipeline, VsrConsensus};
use iggy_common::IggyByteSize;
use iggy_common::variadic;
use metadata::IggyMetadata;
-use metadata::stm::consumer_group::{ConsumerGroups, ConsumerGroupsInner};
use metadata::stm::stream::{Streams, StreamsInner};
use metadata::stm::user::{Users, UsersInner};
use partitions::{IggyPartitions, PartitionsConfig};
@@ -49,8 +48,7 @@ pub type Replica = shard::IggyShard<
pub fn new_replica(id: u8, name: String, bus: &Arc<SimOutbox>, replica_count: u8) -> Replica {
let users: Users = UsersInner::new().into();
let streams: Streams = StreamsInner::new().into();
- let consumer_groups: ConsumerGroups = ConsumerGroupsInner::new().into();
- let mux = SimMuxStateMachine::new(variadic!(users, streams, consumer_groups));
+ let mux = SimMuxStateMachine::new(variadic!(users, streams));
// Metadata uses namespace=0 (not partition-scoped)
let metadata_consensus = VsrConsensus::new(