You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/09/24 10:08:40 UTC
(iggy) branch fix/4273-stale-consumer-group-members updated (86c5b9bf2 -> ab4b46694)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch fix/4273-stale-consumer-group-members
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit 86c5b9bf2 Merge branch 'master' into fix/4273-stale-consumer-group-members
omit 23492f133 fix(server): recover stalled consumer groups after restarts
omit 5d8129e95 chore(deps): Bump the github-actions group across 1 directory with 5 updates (#4268)
omit 85a397d0e chore(deps): Bump the python group across 3 directories with 2 updates (#4280)
omit a131a2d32 chore(deps): Bump the java group across 1 directory with 5 updates (#4267)
new ab4b46694 fix(server): recover groups after missed disconnect cleanup
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (86c5b9bf2)
\
N -- N -- N refs/heads/fix/4273-stale-consumer-group-members (ab4b46694)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/_common.yml | 6 +-
.github/workflows/_test.yml | 16 +--
.github/workflows/_test_examples.yml | 2 +-
.github/workflows/coverage-baseline.yml | 20 ++--
.github/workflows/post-merge.yml | 2 +-
.github/workflows/publish.yml | 2 +-
bdd/docker-compose.cluster.yml | 1 +
bdd/python/uv.lock | 64 +++++-----
.../tests/cluster/client_table_restart.rs | 131 +++++++++++++++++++--
core/metadata/src/impls/metadata.rs | 5 +-
core/metadata/src/stm/stream.rs | 24 ++--
core/server/src/consumer_group/liveness.rs | 31 +++--
core/server/src/session_manager.rs | 13 +-
core/shard/src/lib.rs | 2 -
examples/python/pyproject.toml | 4 +-
examples/python/uv.lock | 68 +++++------
foreign/java/gradle/libs.versions.toml | 2 +-
foreign/python/uv.lock | 64 +++++-----
18 files changed, 288 insertions(+), 169 deletions(-)
(iggy) 01/01: fix(server): recover groups after missed disconnect cleanup
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch fix/4273-stale-consumer-group-members
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit ab4b46694cbfd4d19c19bd4ded056d47f62660d8
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Wed Sep 23 21:10:43 2026 +0200
fix(server): recover groups after missed disconnect cleanup
Disconnected consumers could retain partition assignments when Logout
failed or skipped a missing client-table entry. Replacement consumers
then polled empty, even across server restarts.
Reconcile persisted memberships with bound sessions on every replica.
Commit epoch-fenced Logout operations to release orphaned assignments
while preserving group IDs, saved offsets and live consumers. Persist
member epochs through Join records and snapshots, and reuse a derived
membership index for session refresh and expiry checks.
Bound heartbeat reports and retry backpressure. Defer expiry while
reports are incomplete, with recurring warnings and a resumption log.
Cover failed disconnect cleanup, restart recovery and session rebinds.
Verify automatic reassignment and saved-offset continuation after
client-table eviction. Document missing periodic simulator tasks.
Remove the ignored -g/--group-id CLI creation option. Set the shared
BDD consumer timeout above the cluster timeout for valid startup.
New servers still read the previous snapshot and WAL formats.
The replica protocol change requires coordinated cluster upgrades.
Fixes #4273
---
bdd/docker-compose.cluster.yml | 1 +
core/binary_protocol/src/consensus/command.rs | 11 +-
.../src/consensus/consumer_session.rs | 218 ++++
core/binary_protocol/src/consensus/header.rs | 6 +
core/binary_protocol/src/consensus/mod.rs | 4 +
core/binary_protocol/src/lib.rs | 22 +-
core/cli/src/args/consumer_group.rs | 6 +-
core/configs/src/common/defaults.rs | 10 +
core/configs/src/common/displays.rs | 6 +-
core/configs/src/common/server.rs | 7 +
core/configs/src/server_config/validators.rs | 98 +-
.../test_consumer_group_create_command.rs | 34 +-
.../tests/cluster/client_table_restart.rs | 399 ++++++-
.../stale_client_consumer_group_scenario.rs | 262 ++++-
core/metadata/src/impls/metadata.rs | 427 ++++++-
core/metadata/src/impls/recovery.rs | 53 +-
core/metadata/src/stm/consumer_group.rs | 242 +++-
core/metadata/src/stm/mod.rs | 10 +
core/metadata/src/stm/stream.rs | 81 +-
core/server/config.toml | 19 +-
core/server/src/boot/mod.rs | 47 +-
core/server/src/boot/recovery.rs | 4 +
core/server/src/boot/threads.rs | 12 +-
core/server/src/consumer_group.rs | 4 +
core/server/src/consumer_group/liveness.rs | 1162 ++++++++++++++++++++
core/server/src/dispatch/mod.rs | 14 +-
core/server/src/dispatch/session_ops.rs | 2 +-
core/server/src/dispatch/submit.rs | 24 +-
core/server/src/dispatch/test_support.rs | 8 +
core/server/src/partition_reconciler.rs | 1 +
core/server/src/session_manager.rs | 70 +-
core/server/src/shell.rs | 4 +
core/server_common/src/consensus_message.rs | 16 +-
core/server_common/src/sharding/mod.rs | 10 +
core/shard/src/lib.rs | 75 +-
core/shard/src/poll/completion_tests.rs | 1 +
core/simulator/README.md | 1 +
core/simulator/src/client.rs | 18 +-
core/simulator/src/lib.rs | 274 ++++-
core/simulator/src/packet.rs | 7 +-
core/simulator/src/replica.rs | 12 +-
41 files changed, 3501 insertions(+), 181 deletions(-)
diff --git a/bdd/docker-compose.cluster.yml b/bdd/docker-compose.cluster.yml
index d84da3475..fb9b02954 100644
--- a/bdd/docker-compose.cluster.yml
+++ b/bdd/docker-compose.cluster.yml
@@ -59,6 +59,7 @@ x-cluster-topology: &cluster-topology
# Keep role-specific scenarios on view 0 when a loaded CI runner delays
# one replica during startup.
IGGY_CLUSTER_HEARTBEAT_TIMEOUT: "30s"
+ IGGY_CONSUMER_GROUP_SESSION_TIMEOUT: "60s"
IGGY_CLUSTER_NODES_0_NAME: leader-node
IGGY_CLUSTER_NODES_0_IP: 172.28.0.101
IGGY_CLUSTER_NODES_0_REPLICA_ID: "0"
diff --git a/core/binary_protocol/src/consensus/command.rs b/core/binary_protocol/src/consensus/command.rs
index 81e80336d..46103bd02 100644
--- a/core/binary_protocol/src/consensus/command.rs
+++ b/core/binary_protocol/src/consensus/command.rs
@@ -83,6 +83,9 @@ pub enum Command {
// the connection it owns.
ForwardLogout = 28,
ForwardLogoutResult = 29,
+
+ // Backups report transport-bound sessions to the metadata primary. No reply.
+ ConsumerSessionHeartbeat = 30,
}
// SAFETY: Command is #[repr(u8)] with no padding bytes.
@@ -93,7 +96,7 @@ unsafe impl CheckedBitPattern for Command {
type Bits = u8;
fn is_valid_bit_pattern(bits: &u8) -> bool {
- *bits <= Self::ForwardLogoutResult as u8
+ *bits <= Self::ConsumerSessionHeartbeat as u8
}
}
@@ -114,8 +117,8 @@ mod tests {
#[test]
fn replica_auth_commands_are_valid_bit_patterns() {
- // Locks the is_valid_bit_pattern bump: 14..=29 parse, 30 still rejects.
- for command in 14u8..=29 {
+ // Locks the is_valid_bit_pattern bump: 14..=30 parse, 31 still rejects.
+ for command in 14u8..=30 {
let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
buf.resize(256, 0);
buf[60] = command;
@@ -123,7 +126,7 @@ mod tests {
}
let mut buf: AVec<u8, ConstAlign<16>> = AVec::new(16);
buf.resize(256, 0);
- buf[60] = 30;
+ buf[60] = 31;
assert!(bytemuck::checked::try_from_bytes::<GenericHeader>(&buf).is_err());
}
}
diff --git a/core/binary_protocol/src/consensus/consumer_session.rs b/core/binary_protocol/src/consensus/consumer_session.rs
new file mode 100644
index 000000000..f3a56867c
--- /dev/null
+++ b/core/binary_protocol/src/consensus/consumer_session.rs
@@ -0,0 +1,218 @@
+// 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 super::{Command, ConsensusError, ConsensusHeader, HEADER_SIZE, Operation};
+use crate::codec::{read_u64_le, read_u128_le};
+use crate::{WireDecode, WireEncode, WireError};
+use bytemuck::{CheckedBitPattern, NoUninit};
+use bytes::{BufMut, BytesMut};
+
+pub const MAX_CONSUMER_SESSIONS_PER_HEARTBEAT: usize = 1024;
+
+/// A transport-bound session observed by a replica, independent of client pings.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ConsumerSession {
+ pub client_id: u128,
+ pub session: u64,
+}
+
+impl ConsumerSession {
+ pub const ENCODED_SIZE: usize = size_of::<u128>() + size_of::<u64>();
+}
+
+impl WireEncode for ConsumerSession {
+ fn encoded_size(&self) -> usize {
+ Self::ENCODED_SIZE
+ }
+
+ fn encode(&self, buf: &mut BytesMut) {
+ buf.put_u128_le(self.client_id);
+ buf.put_u64_le(self.session);
+ }
+}
+
+impl WireDecode for ConsumerSession {
+ fn decode(buf: &[u8]) -> Result<(Self, usize), WireError> {
+ let client_id = read_u128_le(buf, 0)?;
+ let session = read_u64_le(buf, size_of::<u128>())?;
+ if client_id == 0 || session == 0 {
+ return Err(WireError::Validation(
+ "consumer session identity must be nonzero".into(),
+ ));
+ }
+ Ok((Self { client_id, session }, Self::ENCODED_SIZE))
+ }
+}
+
+/// Same-release replica control frame with bounded, packed [`ConsumerSession`] entries.
+///
+/// A heartbeat renews only the named epochs;
+/// absence from a batch does not authorize eviction. An incomplete gather
+/// defers expiry, including when no sessions could be collected.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, CheckedBitPattern, NoUninit)]
+#[repr(C)]
+pub struct ConsumerSessionHeartbeatHeader {
+ pub checksum: u128,
+ pub checksum_body: u128,
+ pub cluster: u128,
+ pub size: u32,
+ pub view: u32,
+ pub release: u32,
+ pub command: Command,
+ pub replica: u8,
+ pub reserved_frame: [u8; 66],
+ /// 1 if any local shard failed to report its clients, otherwise 0.
+ pub incomplete: u8,
+ pub reserved: [u8; 127],
+}
+
+const _: () = assert!(size_of::<ConsumerSessionHeartbeatHeader>() == HEADER_SIZE);
+
+impl ConsensusHeader for ConsumerSessionHeartbeatHeader {
+ const FRAME_SEALED: bool = true;
+ const COMMAND: Command = Command::ConsumerSessionHeartbeat;
+
+ fn checksum(&self) -> u128 {
+ self.checksum
+ }
+
+ fn set_checksum(&mut self, checksum: u128) {
+ self.checksum = checksum;
+ }
+
+ fn operation(&self) -> Operation {
+ Operation::Reserved
+ }
+
+ fn command(&self) -> Command {
+ self.command
+ }
+
+ fn size(&self) -> u32 {
+ self.size
+ }
+
+ fn validate(&self) -> Result<(), ConsensusError> {
+ if self.command != Self::COMMAND {
+ return Err(ConsensusError::InvalidCommand {
+ expected: Self::COMMAND,
+ found: self.command,
+ });
+ }
+ let body_size = (self.size as usize).saturating_sub(HEADER_SIZE);
+ if (self.size as usize) < HEADER_SIZE {
+ return Err(ConsensusError::InvalidField(
+ "consumer session heartbeat size is smaller than its header".into(),
+ ));
+ }
+ if body_size > MAX_CONSUMER_SESSIONS_PER_HEARTBEAT * ConsumerSession::ENCODED_SIZE {
+ return Err(ConsensusError::InvalidField(
+ "consumer session heartbeat exceeds the batch limit".into(),
+ ));
+ }
+ if !body_size.is_multiple_of(ConsumerSession::ENCODED_SIZE) {
+ return Err(ConsensusError::InvalidField(
+ "consumer session heartbeat contains a truncated session".into(),
+ ));
+ }
+ if self.incomplete > 1 {
+ return Err(ConsensusError::InvalidField(
+ "consumer session heartbeat incomplete flag must be 0 or 1".into(),
+ ));
+ }
+ if self.release != 0 {
+ return Err(ConsensusError::InvalidField(
+ "consumer session heartbeat release must be zero".into(),
+ ));
+ }
+ if self.reserved_frame != [0; 66] || self.reserved != [0; 127] {
+ return Err(ConsensusError::InvalidField(
+ "consumer session heartbeat reserved bytes must be zero".into(),
+ ));
+ }
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn session_round_trip_and_truncation() {
+ let session = ConsumerSession {
+ client_id: u128::MAX,
+ session: u64::MAX,
+ };
+ let encoded = session.to_bytes();
+ assert_eq!(encoded.len(), ConsumerSession::ENCODED_SIZE);
+ assert_eq!(
+ ConsumerSession::decode(&encoded).unwrap(),
+ (session, encoded.len())
+ );
+ for length in 0..encoded.len() {
+ assert!(ConsumerSession::decode(&encoded[..length]).is_err());
+ }
+ assert!(ConsumerSession::decode(&[0; ConsumerSession::ENCODED_SIZE]).is_err());
+ }
+
+ #[test]
+ fn heartbeat_rejects_unbounded_or_partial_batches() {
+ let mut header = ConsumerSessionHeartbeatHeader {
+ checksum: 0,
+ checksum_body: 0,
+ cluster: 1,
+ size: 0,
+ view: 0,
+ release: 0,
+ command: Command::ConsumerSessionHeartbeat,
+ replica: 0,
+ reserved_frame: [0; 66],
+ incomplete: 0,
+ reserved: [0; 127],
+ };
+ for count in [0, 1, MAX_CONSUMER_SESSIONS_PER_HEARTBEAT] {
+ header.size =
+ u32::try_from(HEADER_SIZE + count * ConsumerSession::ENCODED_SIZE).unwrap();
+ assert!(header.validate().is_ok());
+ header.seal();
+ assert!(header.verify_frame().is_ok());
+ }
+ for size in [
+ 0,
+ HEADER_SIZE + 1,
+ HEADER_SIZE + (MAX_CONSUMER_SESSIONS_PER_HEARTBEAT + 1) * ConsumerSession::ENCODED_SIZE,
+ ] {
+ header.size = u32::try_from(size).unwrap();
+ assert!(header.validate().is_err());
+ }
+ header.size = u32::try_from(HEADER_SIZE).unwrap();
+ header.incomplete = 1;
+ assert!(header.validate().is_ok());
+ header.incomplete = 2;
+ assert!(header.validate().is_err());
+ header.incomplete = 0;
+ header.release = 1;
+ assert!(header.validate().is_err());
+ header.release = 0;
+ header.reserved[0] = 1;
+ assert!(header.validate().is_err());
+ header.reserved[0] = 0;
+ header.reserved_frame[0] = 1;
+ assert!(header.validate().is_err());
+ }
+}
diff --git a/core/binary_protocol/src/consensus/header.rs b/core/binary_protocol/src/consensus/header.rs
index eaf8126db..bb996be64 100644
--- a/core/binary_protocol/src/consensus/header.rs
+++ b/core/binary_protocol/src/consensus/header.rs
@@ -38,6 +38,12 @@
//! operations wait for their forward timeout in a mixed fleet. These commands are
//! another reason the release cannot be rolled node by node.
//!
+//! Consumer-session heartbeats (command 30) also require a coordinated upgrade.
+//! Older replicas reject this command and close the replica connection each time
+//! a new backup reports its sessions.
+//! Older replicas cannot report live sessions to a new primary, which would
+//! expire their consumer-group memberships after the configured timeout.
+//!
//! `Prepare`, `Request`, `Reply`, and `Eviction` are unaffected. Prepares keep
//! `checksum` as their view-independent identity, and the three client-facing
//! headers are sealed on neither side, so SDKs are untouched.
diff --git a/core/binary_protocol/src/consensus/mod.rs b/core/binary_protocol/src/consensus/mod.rs
index 24a3768af..67525a14e 100644
--- a/core/binary_protocol/src/consensus/mod.rs
+++ b/core/binary_protocol/src/consensus/mod.rs
@@ -37,12 +37,16 @@
//! - [`StartViewHeader`] - new primary -> all replicas (header-only)
mod command;
+mod consumer_session;
mod error;
mod header;
mod operation;
mod reply_result;
pub use command::Command;
+pub use consumer_session::{
+ ConsumerSession, ConsumerSessionHeartbeatHeader, MAX_CONSUMER_SESSIONS_PER_HEARTBEAT,
+};
pub use error::ConsensusError;
pub use header::{
CHECKSUM_UNSEALED, CommitHeader, ConsensusHeader, DVC_HEADERS_MAX, DoViewChangeHeader,
diff --git a/core/binary_protocol/src/lib.rs b/core/binary_protocol/src/lib.rs
index fb3b64d63..1735e8b29 100644
--- a/core/binary_protocol/src/lib.rs
+++ b/core/binary_protocol/src/lib.rs
@@ -70,17 +70,17 @@ pub mod version;
pub use codec::{WireDecode, WireEncode};
pub use consensus::{
- CHECKSUM_UNSEALED, Command, CommitHeader, ConsensusError, ConsensusHeader, DVC_HEADERS_MAX,
- DoViewChangeHeader, EvictionHeader, EvictionReason, ForwardLogoutHeader, ForwardLogoutOutcome,
- ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterOutcome,
- ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, Operation, PrepareHeader,
- PrepareOkHeader, REJECTION_SECTION_LEN, RESERVED_COMMAND_LEN, RepairPrepareHeader,
- RepairRangeReplyHeader, ReplyHeader, RequestHeader, RequestPreparesHeader,
- RequestStartViewHeader, RequestStateChunkHeader, RequestStateTransferHeader,
- RoutedRequestHeader, SIZE_FIELD_OFFSET, StartViewChangeHeader, StartViewHeader,
- StateChunkHeader, StateTransferTargetHeader, frame_body, frame_checksum_bytes,
- prepare_identity_checksum_bytes, read_size_field, result_code, result_section_len,
- write_rejection_section,
+ CHECKSUM_UNSEALED, Command, CommitHeader, ConsensusError, ConsensusHeader, ConsumerSession,
+ ConsumerSessionHeartbeatHeader, DVC_HEADERS_MAX, DoViewChangeHeader, EvictionHeader,
+ EvictionReason, ForwardLogoutHeader, ForwardLogoutOutcome, ForwardLogoutResultHeader,
+ ForwardRegisterHeader, ForwardRegisterOutcome, ForwardRegisterResultHeader, GenericHeader,
+ HEADER_SIZE, MAX_CONSUMER_SESSIONS_PER_HEARTBEAT, Operation, PrepareHeader, PrepareOkHeader,
+ REJECTION_SECTION_LEN, RESERVED_COMMAND_LEN, RepairPrepareHeader, RepairRangeReplyHeader,
+ ReplyHeader, RequestHeader, RequestPreparesHeader, RequestStartViewHeader,
+ RequestStateChunkHeader, RequestStateTransferHeader, RoutedRequestHeader, SIZE_FIELD_OFFSET,
+ StartViewChangeHeader, StartViewHeader, StateChunkHeader, StateTransferTargetHeader,
+ frame_body, frame_checksum_bytes, prepare_identity_checksum_bytes, read_size_field,
+ result_code, result_section_len, write_rejection_section,
};
pub use dispatch::{COMMAND_TABLE, CommandMeta, lookup_by_operation, lookup_command};
pub use error::WireError;
diff --git a/core/cli/src/args/consumer_group.rs b/core/cli/src/args/consumer_group.rs
index 488511f62..efdb502c4 100644
--- a/core/cli/src/args/consumer_group.rs
+++ b/core/cli/src/args/consumer_group.rs
@@ -25,13 +25,12 @@ pub(crate) enum ConsumerGroupAction {
///
/// Stream ID can be specified as a stream name or ID
/// Topic ID can be specified as a topic name or ID
- /// The server assigns the group ID. The legacy --group-id flag is ignored.
+ /// The server assigns the group ID.
///
/// Examples:
/// iggy consumer-group create 1 1 prod
/// iggy consumer-group create stream 2 test
/// iggy consumer-group create 2 topic receiver
- /// iggy consumer-group create -g 4 stream topic group
#[clap(verbatim_doc_comment, visible_alias = "c")]
Create(ConsumerGroupCreateArgs),
/// Delete consumer group with given ID for given stream ID and topic ID
@@ -94,9 +93,6 @@ pub(crate) struct ConsumerGroupCreateArgs {
/// Topic ID can be specified as a topic name or ID
#[arg(value_parser = clap::value_parser!(Identifier))]
pub(crate) topic_id: Identifier,
- /// Legacy consumer group ID flag (ignored)
- #[clap(short, long)]
- pub(crate) group_id: Option<u32>,
/// Consumer group name to create
pub(crate) name: String,
}
diff --git a/core/configs/src/common/defaults.rs b/core/configs/src/common/defaults.rs
index 738121928..18e588c54 100644
--- a/core/configs/src/common/defaults.rs
+++ b/core/configs/src/common/defaults.rs
@@ -180,6 +180,16 @@ impl Default for HeartbeatConfig {
impl Default for ConsumerGroupConfig {
fn default() -> ConsumerGroupConfig {
ConsumerGroupConfig {
+ session_timeout: SERVER_CONFIG
+ .consumer_group
+ .session_timeout
+ .parse()
+ .unwrap(),
+ heartbeat_interval: SERVER_CONFIG
+ .consumer_group
+ .heartbeat_interval
+ .parse()
+ .unwrap(),
rebalancing_timeout: SERVER_CONFIG
.consumer_group
.rebalancing_timeout
diff --git a/core/configs/src/common/displays.rs b/core/configs/src/common/displays.rs
index 454bba800..f2535cbd6 100644
--- a/core/configs/src/common/displays.rs
+++ b/core/configs/src/common/displays.rs
@@ -106,7 +106,11 @@ impl Display for MessagesMaintenanceConfig {
impl Display for ConsumerGroupConfig {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
- write!(f, "{{ rebalancing_timeout: {} }}", self.rebalancing_timeout)
+ write!(
+ f,
+ "{{ rebalancing_timeout: {}, session_timeout: {}, heartbeat_interval: {} }}",
+ self.rebalancing_timeout, self.session_timeout, self.heartbeat_interval
+ )
}
}
diff --git a/core/configs/src/common/server.rs b/core/configs/src/common/server.rs
index f3cec3b49..dec2379da 100644
--- a/core/configs/src/common/server.rs
+++ b/core/configs/src/common/server.rs
@@ -85,10 +85,17 @@ pub struct HeartbeatConfig {
#[serde_as]
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
+#[serde(default)]
pub struct ConsumerGroupConfig {
#[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub rebalancing_timeout: IggyDuration,
+ #[config_env(leaf)]
+ #[serde_as(as = "DisplayFromStr")]
+ pub session_timeout: IggyDuration,
+ #[config_env(leaf)]
+ #[serde_as(as = "DisplayFromStr")]
+ pub heartbeat_interval: IggyDuration,
}
#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
diff --git a/core/configs/src/server_config/validators.rs b/core/configs/src/server_config/validators.rs
index fef1ebe08..95af784df 100644
--- a/core/configs/src/server_config/validators.rs
+++ b/core/configs/src/server_config/validators.rs
@@ -31,10 +31,16 @@ use crate::common::validators::SEGMENT_MAX_SIZE_BYTES;
use crate::{ConfigEnvMappings, ConfigurationError};
use err_trail::ErrContext;
use iggy_common::{IggyExpiry, MAX_MESSAGE_SIZE_UPPER_BYTES, Validatable};
+use server_common::sharding::{
+ CONSUMER_SESSION_REPORT_TIMEOUT, LIST_CLIENTS_GATHER_TIMEOUT,
+ MIN_CONSUMER_SESSION_HEARTBEAT_SIZE,
+};
use std::ffi::OsStr;
use std::net::SocketAddr;
use std::path::Path;
+const CONSUMER_SESSION_MISSED_REPORT_TOLERANCE: u32 = 1;
+
/// compio-ws (tungstenite 0.29) `write_buffer_size` default. Used to
/// evaluate the `max_write_buffer_size > write_buffer_size` invariant
/// when the operator leaves `write_buffer_size` unset; keep in sync
@@ -255,6 +261,12 @@ impl Validatable<ConfigurationError> for ServerConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if bus_cap < MIN_CONSUMER_SESSION_HEARTBEAT_SIZE as u64 {
+ eprintln!(
+ "{COMPONENT} message_bus.max_message_size ({bus_cap}) must hold a consumer-session heartbeat of at least {MIN_CONSUMER_SESSION_HEARTBEAT_SIZE} bytes"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
// WS frame chain: websocket.max_frame_size <= websocket.max_message_size
// <= message_bus.max_message_size. The bus's WS / WSS install path takes
@@ -344,9 +356,37 @@ impl Validatable<ConfigurationError> for ServerConfig {
format!("{COMPONENT} (error: {e}) - failed to validate quic config")
})?;
- // Both knobs below sit on shared section structs, so the rejects live
+ // These knobs sit on shared section structs, so the rejects live
// here rather than in those types' own `Validatable` impls. `0` /
// `disabled` / `unlimited` all parse to the same zero duration.
+ let session_timeout = self.consumer_group.session_timeout.get_duration();
+ let heartbeat_interval = self.consumer_group.heartbeat_interval.get_duration();
+ if heartbeat_interval.is_zero() {
+ eprintln!(
+ "{COMPONENT} consumer_group.heartbeat_interval ({heartbeat_interval:?}) must be nonzero; consumer_group.session_timeout is {session_timeout:?}"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ // Cover a missed pass, gather/send deadlines, and one interval of scheduling headroom.
+ let minimum_timeout = heartbeat_interval
+ .saturating_add(LIST_CLIENTS_GATHER_TIMEOUT)
+ .saturating_add(CONSUMER_SESSION_REPORT_TIMEOUT)
+ .saturating_mul(CONSUMER_SESSION_MISSED_REPORT_TOLERANCE + 1)
+ .saturating_add(heartbeat_interval);
+ if session_timeout <= minimum_timeout {
+ eprintln!(
+ "{COMPONENT} consumer_group.session_timeout ({session_timeout:?}) must exceed {minimum_timeout:?}, covering three consumer_group.heartbeat_interval periods and two gather/send deadlines, allowing one missed report"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ let cluster_timeout = self.cluster.heartbeat_timeout.get_duration();
+ if self.cluster.enabled && session_timeout <= cluster_timeout {
+ eprintln!(
+ "{COMPONENT} consumer_group.session_timeout ({session_timeout:?}) must exceed cluster.heartbeat_timeout ({cluster_timeout:?})"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
if self
.consumer_group
.rebalancing_timeout
@@ -1060,6 +1100,62 @@ mod tests {
assert!(config.validate().is_err());
}
+ #[test]
+ fn given_invalid_consumer_session_intervals_when_validating_should_reject() {
+ for settings in [
+ "heartbeat_interval = \"0\"",
+ "session_timeout = \"0\"",
+ "session_timeout = \"disabled\"",
+ "session_timeout = \"unlimited\"",
+ "heartbeat_interval = \"500ms\"\nsession_timeout = \"2s\"",
+ "heartbeat_interval = \"500ms\"\nsession_timeout = \"6500ms\"",
+ "heartbeat_interval = \"500ms\"\nsession_timeout = \"7500ms\"",
+ "heartbeat_interval = \"500ms\"\nsession_timeout = \"7700ms\"",
+ "heartbeat_interval = \"20s\"\nsession_timeout = \"27s\"",
+ "heartbeat_interval = \"20s\"\nsession_timeout = \"66s\"",
+ "heartbeat_interval = \"30s\"\nsession_timeout = \"30s\"",
+ "heartbeat_interval = \"31s\"\nsession_timeout = \"30s\"",
+ ] {
+ let config = config_with_override(&format!("[consumer_group]\n{settings}\n"));
+ assert!(config.validate().is_err(), "accepted {settings}");
+ }
+ }
+
+ #[test]
+ fn given_consumer_session_timeout_above_missed_report_budget_when_validating_should_pass() {
+ let config = config_with_override(
+ "[consumer_group]\nheartbeat_interval = \"500ms\"\nsession_timeout = \"7701ms\"\n",
+ );
+ assert!(config.validate().is_ok());
+ }
+
+ #[test]
+ fn given_consumer_heartbeat_frame_limits_when_validating_should_require_one_session() {
+ for (cap, valid) in [
+ (MIN_CONSUMER_SESSION_HEARTBEAT_SIZE - 1, false),
+ (MIN_CONSUMER_SESSION_HEARTBEAT_SIZE, true),
+ ] {
+ let config =
+ config_with_override(&format!("[message_bus]\nmax_message_size = \"{cap} B\"\n"));
+ assert_eq!(config.validate().is_ok(), valid, "frame cap {cap}");
+ }
+ }
+
+ #[test]
+ fn given_consumer_session_timeout_within_cluster_liveness_window_when_validating_should_reject()
+ {
+ for timeout in ["15s", "20s"] {
+ let config = config_with_override(&format!(
+ "[cluster]\nenabled = true\nheartbeat_timeout = \"20s\"\n[consumer_group]\nheartbeat_interval = \"500ms\"\nsession_timeout = \"{timeout}\"\n"
+ ));
+ assert!(config.validate().is_err(), "accepted {timeout}");
+ }
+ let config = config_with_override(
+ "[cluster]\nenabled = true\nheartbeat_timeout = \"20s\"\n[consumer_group]\nheartbeat_interval = \"500ms\"\nsession_timeout = \"21s\"\n",
+ );
+ assert!(config.validate().is_ok());
+ }
+
#[test]
fn given_zero_rebalancing_timeout_when_validating_should_reject() {
let config = config_with_override("[consumer_group]\nrebalancing_timeout = \"0\"\n");
diff --git a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
index 856d7dcb6..97c18f74a 100644
--- a/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
+++ b/core/integration/tests/cli/consumer_group/test_consumer_group_create_command.rs
@@ -30,20 +30,17 @@ struct TestConsumerGroupCreateCmd {
stream_name: String,
topic_id: u32,
topic_name: String,
- group_id: Option<u32>,
group_name: String,
using_stream_id: TestStreamId,
using_topic_id: TestTopicId,
}
impl TestConsumerGroupCreateCmd {
- #[allow(clippy::too_many_arguments)]
fn new(
stream_id: u32,
stream_name: String,
topic_id: u32,
topic_name: String,
- group_id: Option<u32>,
group_name: String,
using_stream_id: TestStreamId,
using_topic_id: TestTopicId,
@@ -53,7 +50,6 @@ impl TestConsumerGroupCreateCmd {
stream_name,
topic_id,
topic_name,
- group_id,
group_name,
using_stream_id,
using_topic_id,
@@ -71,11 +67,6 @@ impl TestConsumerGroupCreateCmd {
TestTopicId::Named => self.topic_name.clone(),
});
- if let Some(group_id) = self.group_id {
- command.push("-g".to_string());
- command.push(format!("{group_id}"));
- }
-
command.push(self.group_name.clone());
command
@@ -125,10 +116,7 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
TestTopicId::Named => self.topic_name.clone(),
};
- let group_id = match self.group_id {
- Some(group_id) => format!("ID: {group_id}"),
- None => "ID auto incremented".to_string(),
- };
+ let group_id = "ID auto incremented";
let message = format!(
"Executing create consumer group: {}, name: {} for topic with ID: {} and stream with ID: {}\nConsumer group: {}, name: {} created for topic with ID: {} and stream with ID: {}\n",
@@ -158,9 +146,6 @@ impl IggyCmdTestCase for TestConsumerGroupCreateCmd {
.unwrap()
.expect("Failed to get consumer group");
assert_eq!(consumer_group_details.name, self.group_name);
- if let Some(group_id) = self.group_id {
- assert_eq!(consumer_group_details.id, group_id);
- }
let topic = client
.delete_topic(
@@ -189,7 +174,6 @@ pub async fn should_be_successful() {
String::from("main"),
1,
String::from("sync"),
- None,
String::from("group1"),
TestStreamId::Numeric,
TestTopicId::Numeric,
@@ -201,7 +185,6 @@ pub async fn should_be_successful() {
String::from("stream"),
3,
String::from("topic"),
- None,
String::from("group3"),
TestStreamId::Named,
TestTopicId::Numeric,
@@ -213,7 +196,6 @@ pub async fn should_be_successful() {
String::from("development"),
1,
String::from("probe"),
- None,
String::from("group7"),
TestStreamId::Numeric,
TestTopicId::Named,
@@ -225,7 +207,6 @@ pub async fn should_be_successful() {
String::from("production"),
5,
String::from("test"),
- None,
String::from("group4"),
TestStreamId::Named,
TestTopicId::Named,
@@ -246,15 +227,14 @@ pub async fn should_help_match() {
Stream ID can be specified as a stream name or ID
Topic ID can be specified as a topic name or ID
-The server assigns the group ID. The legacy --group-id flag is ignored.
+The server assigns the group ID.
Examples:
iggy consumer-group create 1 1 prod
iggy consumer-group create stream 2 test
iggy consumer-group create 2 topic receiver
- iggy consumer-group create -g 4 stream topic group
-{USAGE_PREFIX} consumer-group create [OPTIONS] <STREAM_ID> <TOPIC_ID> <NAME>
+{USAGE_PREFIX} consumer-group create <STREAM_ID> <TOPIC_ID> <NAME>
Arguments:
<STREAM_ID>
@@ -271,9 +251,6 @@ Arguments:
Consumer group name to create
Options:
- -g, --group-id <GROUP_ID>
- Legacy consumer group ID flag (ignored)
-
-h, --help
Print help (see a summary with '-h')
"#,
@@ -293,7 +270,7 @@ pub async fn should_short_help_match() {
format!(
r#"Create consumer group with given name for given stream ID and topic ID.
-{USAGE_PREFIX} consumer-group create [OPTIONS] <STREAM_ID> <TOPIC_ID> <NAME>
+{USAGE_PREFIX} consumer-group create <STREAM_ID> <TOPIC_ID> <NAME>
Arguments:
<STREAM_ID> Stream ID to create consumer group
@@ -301,8 +278,7 @@ Arguments:
<NAME> Consumer group name to create
Options:
- -g, --group-id <GROUP_ID> Legacy consumer group ID flag (ignored)
- -h, --help Print help (see more with '--help')
+ -h, --help Print help (see more with '--help')
"#,
),
))
diff --git a/core/integration/tests/cluster/client_table_restart.rs b/core/integration/tests/cluster/client_table_restart.rs
index e2b87a500..462f2b2fc 100644
--- a/core/integration/tests/cluster/client_table_restart.rs
+++ b/core/integration/tests/cluster/client_table_restart.rs
@@ -74,21 +74,28 @@
use bytes::Bytes;
use iggy::prelude::*;
use iggy_binary_protocol::codec::{WireDecode, WireEncode};
+use iggy_binary_protocol::codes::POLL_MESSAGES_CODE;
use iggy_binary_protocol::consensus::{
Command, Operation, ReplyHeader, RequestHeader, read_size_field, result_code,
result_section_len,
};
+use iggy_binary_protocol::primitives::consumer::WireConsumer;
+use iggy_binary_protocol::primitives::polling_strategy::WirePollingStrategy;
+use iggy_binary_protocol::requests::consumer_groups::JoinConsumerGroupRequest;
+use iggy_binary_protocol::requests::messages::PollMessagesRequest;
use iggy_binary_protocol::requests::streams::CreateStreamRequest;
use iggy_binary_protocol::requests::users::LoginRegisterRequest;
+use iggy_binary_protocol::responses::messages::PollMessagesResponse;
use iggy_binary_protocol::responses::users::LoginRegisterResponse;
use iggy_binary_protocol::{
- ClientVersionInfo, HEADER_SIZE, IGGY_PROTOCOL_VERSION, WireName, WireOptions,
+ ClientVersionInfo, HEADER_SIZE, IGGY_PROTOCOL_VERSION, WireIdentifier, WireName, WireOptions,
};
use integration::harness::TestHarness;
use integration::iggy_harness;
use secrecy::SecretString;
use std::mem::offset_of;
use std::net::SocketAddr;
+use std::str::FromStr;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
@@ -288,10 +295,14 @@ fn request_header(
/// transient rejections: right after boot the single node may not have
/// elected itself yet.
pub(super) async fn register(addr: SocketAddr) -> (TcpStream, u64) {
+ register_client(addr, CLIENT_ID).await
+}
+
+async fn register_client(addr: SocketAddr, client_id: u128) -> (TcpStream, u64) {
let mut stream = TcpStream::connect(addr).await.unwrap();
let deadline = Instant::now() + COMMIT_BUDGET;
loop {
- if let Some(session) = login_on(&mut stream).await {
+ if let Some(session) = login_on(&mut stream, client_id).await {
return (stream, session);
}
assert!(
@@ -302,10 +313,10 @@ pub(super) async fn register(addr: SocketAddr) -> (TcpStream, u64) {
}
}
-/// Root login/register for `CLIENT_ID` on an already-connected socket.
+/// Root login/register on an already-connected socket.
/// `Some(session)` on a committed register, `None` on a transient rejection
/// (right after boot the node may not have elected itself yet).
-async fn login_on(stream: &mut TcpStream) -> Option<u64> {
+async fn login_on(stream: &mut TcpStream, client_id: u128) -> Option<u64> {
let body = LoginRegisterRequest {
version_info: ClientVersionInfo {
protocol_version: IGGY_PROTOCOL_VERSION,
@@ -317,7 +328,8 @@ async fn login_on(stream: &mut TcpStream) -> Option<u64> {
client_context: None,
}
.to_bytes();
- let header = request_header(Operation::Register, 0, 0, body.len());
+ let mut header = request_header(Operation::Register, 0, 0, body.len());
+ header.client = client_id;
match exchange(stream, &header, &body).await.verdict() {
Verdict::Success(reply) => {
@@ -390,7 +402,7 @@ pub(super) async fn resume_request(
// Re-authenticate on the fresh connection. The rebind commits a
// Register, so the epoch strictly advances past the pre-restart one
// (op-derived; regression would mean the fence can be replayed into).
- let resumed = match login_on(&mut stream).await {
+ let resumed = match login_on(&mut stream, CLIENT_ID).await {
Some(resumed) => {
assert!(
resumed > session,
@@ -607,3 +619,378 @@ fn is_transient(code: u32) -> bool {
code == IggyError::TransientNotCommitted.as_code()
|| code == IggyError::TransientNotAccepted.as_code()
}
+
+const LIVENESS_STREAM: &str = "session-liveness";
+const LIVENESS_TOPIC: &str = "events";
+const LIVENESS_GROUP: &str = "readers";
+const LIVENESS_OBSERVATION: Duration = Duration::from_secs(12);
+
+#[iggy_harness(server(
+ metadata.clients_table_max = 4,
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+ sharding.cpu_allocation = "0..1",
+))]
+async fn given_capacity_evicted_member_when_disconnected_should_resume_from_saved_offset(
+ harness: &mut TestHarness,
+) {
+ const CAPACITY_PRESSURE: u32 = 12;
+ let observer = harness.root_client_for_node(0).await.unwrap();
+ create_liveness_group(&observer).await;
+ let stream = Identifier::named(LIVENESS_STREAM).unwrap();
+ let topic = Identifier::named(LIVENESS_TOPIC).unwrap();
+ let group = Identifier::named(LIVENESS_GROUP).unwrap();
+ let consumer = Consumer::group(group.clone());
+ let mut messages = [
+ IggyMessage::from_str("committed").unwrap(),
+ IggyMessage::from_str("pending").unwrap(),
+ ];
+ observer
+ .send_messages(
+ &stream,
+ &topic,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .unwrap();
+ let addr = harness.node(0).tcp_addr().unwrap();
+ let (mut member, session) = register(addr).await;
+ let join = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::named(LIVENESS_STREAM).unwrap(),
+ topic_id: WireIdentifier::named(LIVENESS_TOPIC).unwrap(),
+ group_id: WireIdentifier::named(LIVENESS_GROUP).unwrap(),
+ }
+ .to_bytes();
+ let header = request_header(Operation::JoinConsumerGroup, session, 1, join.len());
+ assert!(matches!(
+ exchange(&mut member, &header, &join).await.verdict(),
+ Verdict::Success(_)
+ ));
+ let original = liveness_group(&observer).await;
+ let mut poll_request = PollMessagesRequest {
+ consumer: WireConsumer::consumer_group(WireIdentifier::named(LIVENESS_GROUP).unwrap()),
+ stream_id: WireIdentifier::named(LIVENESS_STREAM).unwrap(),
+ topic_id: WireIdentifier::named(LIVENESS_TOPIC).unwrap(),
+ partition_id: Some(0),
+ strategy: WirePollingStrategy::first(),
+ count: 1,
+ auto_commit: true,
+ };
+ let first_poll = poll_request.to_bytes();
+ let mut first_header = request_header(Operation::NonReplicated, session, 0, first_poll.len());
+ first_header.reserved[..size_of::<u32>()].copy_from_slice(&POLL_MESSAGES_CODE.to_le_bytes());
+ let deadline = Instant::now() + COMMIT_BUDGET;
+ loop {
+ let Exchange::Reply {
+ status: 0, body, ..
+ } = exchange(&mut member, &first_header, &first_poll).await
+ else {
+ panic!("original member could not poll its assigned partition");
+ };
+ let mut response = PollMessagesResponse::decode(&body).unwrap();
+ if let Some(message) = response.messages.next() {
+ let message = message.unwrap();
+ assert_eq!(message.offset, 0);
+ assert_eq!(message.payload, messages[0].payload.as_ref());
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "seeded messages were not readable"
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+ assert_eq!(
+ observer
+ .get_consumer_offset(&consumer, &stream, &topic, Some(0))
+ .await
+ .unwrap()
+ .unwrap()
+ .stored_offset,
+ 0
+ );
+ let mut connections = Vec::new();
+ for index in 1..=CAPACITY_PRESSURE {
+ let (connection, _) = register_client(addr, CLIENT_ID + u128::from(index)).await;
+ connections.push(connection);
+ }
+ poll_request.auto_commit = false;
+ let poll = poll_request.to_bytes();
+ let mut header = request_header(Operation::NonReplicated, session, 0, poll.len());
+ header.reserved[..size_of::<u32>()].copy_from_slice(&POLL_MESSAGES_CODE.to_le_bytes());
+ let deadline = Instant::now() + LIVENESS_OBSERVATION;
+ while Instant::now() < deadline {
+ assert!(
+ matches!(
+ exchange(&mut member, &header, &poll).await,
+ Exchange::Reply { status: 0, .. }
+ ),
+ "live member lost its polling assignment after capacity eviction"
+ );
+ assert_eq!(liveness_group(&observer).await.members_count, 1);
+ sleep(RETRY_PAUSE).await;
+ }
+ let replacement = harness.root_client_for_node(0).await.unwrap();
+ replacement
+ .join_consumer_group(&stream, &topic, &group)
+ .await
+ .unwrap();
+ let waiting = liveness_group(&observer).await;
+ assert_eq!(waiting.members_count, 2);
+ assert_eq!(
+ waiting
+ .members
+ .iter()
+ .find(|member| member.id == original.members[0].id)
+ .unwrap()
+ .partitions,
+ vec![0]
+ );
+ assert!(
+ replacement
+ .poll_messages(
+ &stream,
+ &topic,
+ None,
+ &consumer,
+ &PollingStrategy::next(),
+ 1,
+ false
+ )
+ .await
+ .unwrap()
+ .messages
+ .is_empty(),
+ "replacement must wait while the original owns the only partition"
+ );
+ drop(member);
+ let deadline = Instant::now() + RESUME_BUDGET;
+ loop {
+ let recovered = replacement
+ .poll_messages(
+ &stream,
+ &topic,
+ None,
+ &consumer,
+ &PollingStrategy::next(),
+ 1,
+ false,
+ )
+ .await
+ .unwrap();
+ if let Some(message) = recovered.messages.first() {
+ assert_eq!(message.header.offset, 1);
+ assert_eq!(message.payload, messages[1].payload);
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "replacement never resumed after the capacity-evicted member disconnected"
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+ let recovered = liveness_group(&observer).await;
+ assert_eq!(recovered.id, original.id, "expiry must preserve the group");
+ assert_eq!(recovered.members_count, 1);
+ assert_ne!(recovered.members[0].id, original.members[0].id);
+ assert_eq!(recovered.members[0].partitions, vec![0]);
+ drop(connections);
+}
+
+#[iggy_harness(server(
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "5s",
+ consumer_group.session_timeout = "22s",
+ sharding.cpu_allocation = "0..1",
+))]
+async fn given_many_recovered_members_when_expiring_should_drain_without_another_interval(
+ harness: &mut TestHarness,
+) {
+ // Exceeds the per-pass logout cap, so recovery needs multiple passes.
+ const MEMBER_COUNT: u32 = 300;
+ const DRAIN_BUDGET: Duration = Duration::from_secs(4);
+ const FIRST_EXPIRY_BUDGET: Duration = Duration::from_secs(45);
+ let observer = harness.root_client_for_node(0).await.unwrap();
+ create_liveness_group(&observer).await;
+ let body = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::named(LIVENESS_STREAM).unwrap(),
+ topic_id: WireIdentifier::named(LIVENESS_TOPIC).unwrap(),
+ group_id: WireIdentifier::named(LIVENESS_GROUP).unwrap(),
+ }
+ .to_bytes();
+ let mut connections = Vec::with_capacity(MEMBER_COUNT as usize);
+ for member in 0..MEMBER_COUNT {
+ let client_id = CLIENT_ID + u128::from(member);
+ let (mut connection, session) =
+ register_client(harness.node(0).tcp_addr().unwrap(), client_id).await;
+ let mut header = request_header(Operation::JoinConsumerGroup, session, 1, body.len());
+ header.client = client_id;
+ assert!(matches!(
+ exchange(&mut connection, &header, &body).await.verdict(),
+ Verdict::Success(_)
+ ));
+ connections.push(connection);
+ }
+ assert_eq!(liveness_group(&observer).await.members_count, MEMBER_COUNT);
+ harness.kill_node(0).unwrap();
+ drop(connections);
+ harness.restart_node(0).unwrap();
+ let observer = harness.root_client_for_node(0).await.unwrap();
+ let deadline = Instant::now() + FIRST_EXPIRY_BUDGET;
+ while liveness_group(&observer).await.members_count == MEMBER_COUNT {
+ assert!(
+ Instant::now() < deadline,
+ "recovered members never began expiring"
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+ let deadline = Instant::now() + DRAIN_BUDGET;
+ loop {
+ let remaining = liveness_group(&observer).await.members_count;
+ if remaining == 0 {
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "cleanup waited another interval with {remaining} members remaining"
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+}
+
+#[iggy_harness(cluster_nodes = 3, server(
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+ sharding.cpu_allocation = "0..1",
+))]
+async fn given_backup_member_when_primary_restarts_should_preserve_membership(
+ harness: &mut TestHarness,
+) {
+ let observer = harness.root_client_for_node(0).await.unwrap();
+ let (connection, primary, backup) = bind_group_member_on_backup(harness, &observer).await;
+ let before = liveness_group(&observer).await;
+ sleep(LIVENESS_OBSERVATION).await;
+ let renewed = liveness_group(&observer).await;
+ assert_eq!(renewed.members_count, 1);
+ assert_eq!(renewed.members[0].id, before.members[0].id);
+
+ harness.kill_node(primary).unwrap();
+ harness.restart_node(primary).unwrap();
+ sleep(LIVENESS_OBSERVATION).await;
+ let observer = harness.root_client_for_node(backup).await.unwrap();
+ let after = liveness_group(&observer).await;
+ assert_eq!(
+ after.members_count, 1,
+ "the new primary must honor a live backup's heartbeats"
+ );
+ assert_eq!(after.members[0].id, before.members[0].id);
+ assert_eq!(after.members[0].partitions, before.members[0].partitions);
+ drop(connection);
+}
+
+#[iggy_harness(cluster_nodes = 3, server(
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+ sharding.cpu_allocation = "0..1",
+))]
+async fn given_backup_member_when_host_crashes_should_expire_membership(harness: &mut TestHarness) {
+ let observer = harness.root_client_for_node(0).await.unwrap();
+ let (connection, _, backup) = bind_group_member_on_backup(harness, &observer).await;
+ assert_eq!(liveness_group(&observer).await.members_count, 1);
+ harness.kill_node(backup).unwrap();
+ let deadline = Instant::now() + RESUME_BUDGET;
+ while liveness_group(&observer).await.members_count != 0 {
+ assert!(
+ Instant::now() < deadline,
+ "dead backup member was not evicted without restarting its host"
+ );
+ sleep(RETRY_PAUSE).await;
+ }
+ drop(connection);
+}
+
+async fn liveness_group(observer: &IggyClient) -> ConsumerGroupDetails {
+ observer
+ .get_consumer_group(
+ &Identifier::named(LIVENESS_STREAM).unwrap(),
+ &Identifier::named(LIVENESS_TOPIC).unwrap(),
+ &Identifier::named(LIVENESS_GROUP).unwrap(),
+ )
+ .await
+ .unwrap()
+ .unwrap()
+}
+
+async fn bind_group_member_on_backup(
+ harness: &TestHarness,
+ observer: &IggyClient,
+) -> (TcpStream, usize, usize) {
+ create_liveness_group(observer).await;
+ let stream = Identifier::named(LIVENESS_STREAM).unwrap();
+ let topic = Identifier::named(LIVENESS_TOPIC).unwrap();
+ let group = Identifier::named(LIVENESS_GROUP).unwrap();
+ let roster = observer.get_cluster_metadata().await.unwrap();
+ let leader_port = roster
+ .nodes
+ .iter()
+ .find(|node| node.role == ClusterNodeRole::Leader)
+ .unwrap()
+ .endpoints
+ .tcp;
+ let primary = (0..harness.cluster_size())
+ .find(|&index| harness.node(index).tcp_addr().unwrap().port() == leader_port)
+ .unwrap();
+ let backup = (0..harness.cluster_size())
+ .find(|&index| index != primary && index != 0)
+ .unwrap();
+ let (mut original, session) = register(harness.node(primary).tcp_addr().unwrap()).await;
+ let body = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::named(LIVENESS_STREAM).unwrap(),
+ topic_id: WireIdentifier::named(LIVENESS_TOPIC).unwrap(),
+ group_id: WireIdentifier::named(LIVENESS_GROUP).unwrap(),
+ }
+ .to_bytes();
+ let header = request_header(Operation::JoinConsumerGroup, session, 1, body.len());
+ assert!(matches!(
+ exchange(&mut original, &header, &body).await.verdict(),
+ Verdict::Success(_)
+ ));
+
+ // Rebind the same identity on a backup without SDK leader redirection.
+ // Closing the old connection must also fence its delayed disconnect logout.
+ let (backup_connection, new_session) = register(harness.node(backup).tcp_addr().unwrap()).await;
+ assert!(new_session > session);
+ drop(original);
+ let members = observer
+ .get_consumer_group(&stream, &topic, &group)
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(members.members_count, 1);
+ (backup_connection, primary, backup)
+}
+
+async fn create_liveness_group(observer: &IggyClient) {
+ observer.create_stream(LIVENESS_STREAM).await.unwrap();
+ let stream = Identifier::named(LIVENESS_STREAM).unwrap();
+ let topic = Identifier::named(LIVENESS_TOPIC).unwrap();
+ observer
+ .create_topic(
+ &stream,
+ LIVENESS_TOPIC,
+ &TopicCreateOptions {
+ partitions_count: Some(1),
+ ..Default::default()
+ },
+ )
+ .await
+ .unwrap();
+ observer
+ .create_consumer_group(&stream, &topic, LIVENESS_GROUP)
+ .await
+ .unwrap();
+}
diff --git a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
index ec130dc66..1a34e9aa5 100644
--- a/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
+++ b/core/integration/tests/server/scenarios/stale_client_consumer_group_scenario.rs
@@ -17,7 +17,7 @@
use futures::StreamExt;
use iggy::prelude::*;
-use iggy_common::Credentials;
+use iggy_common::{Credentials, Durability};
use integration::iggy_harness;
use secrecy::SecretString;
use std::str::FromStr;
@@ -37,6 +37,254 @@ const CONSUMER_GROUP_NAME: &str = "stale-test-cg";
const PARTITIONS_COUNT: u32 = 1;
const TOTAL_MESSAGES: u32 = 10;
+enum Restart {
+ Client,
+ ServerShutdown,
+ ServerCrash,
+ ClusterCrash,
+}
+
+#[iggy_harness(cluster_nodes = 1, server(
+ heartbeat.enabled = true,
+ heartbeat.interval = "2s",
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+))]
+async fn given_client_disconnect_when_member_is_replaced_should_resume_consumption(
+ harness: &mut integration::harness::TestHarness,
+) {
+ verify_replacement_after_restart(harness, Restart::Client).await;
+}
+
+#[iggy_harness(cluster_nodes = 1, server(
+ heartbeat.enabled = true,
+ heartbeat.interval = "2s",
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+))]
+async fn given_server_shutdown_when_member_is_replaced_should_resume_consumption(
+ harness: &mut integration::harness::TestHarness,
+) {
+ verify_replacement_after_restart(harness, Restart::ServerShutdown).await;
+}
+
+#[iggy_harness(cluster_nodes = 1, server(
+ heartbeat.enabled = true,
+ heartbeat.interval = "2s",
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+))]
+async fn given_server_crash_when_member_is_replaced_should_resume_consumption(
+ harness: &mut integration::harness::TestHarness,
+) {
+ verify_replacement_after_restart(harness, Restart::ServerCrash).await;
+}
+
+#[iggy_harness(cluster_nodes = 1, server(
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+))]
+async fn given_heartbeat_disabled_when_member_is_replaced_after_crash_should_resume_consumption(
+ harness: &mut integration::harness::TestHarness,
+) {
+ verify_replacement_after_restart(harness, Restart::ServerCrash).await;
+}
+
+#[iggy_harness(cluster_nodes = 3, server(
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+))]
+async fn given_cluster_node_restart_when_member_is_replaced_should_resume_consumption(
+ harness: &mut integration::harness::TestHarness,
+) {
+ verify_replacement_after_restart(harness, Restart::ServerCrash).await;
+}
+
+#[iggy_harness(cluster_nodes = 3, server(
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+))]
+async fn given_cluster_restart_when_member_is_replaced_should_resume_consumption(
+ harness: &mut integration::harness::TestHarness,
+) {
+ verify_replacement_after_restart(harness, Restart::ClusterCrash).await;
+}
+
+#[iggy_harness(cluster_nodes = 3, server(
+ heartbeat.enabled = false,
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
+))]
+async fn given_live_member_when_backup_restarts_should_preserve_membership(
+ harness: &mut integration::harness::TestHarness,
+) {
+ let consumer_client = create_client(&harness.server().raw_tcp_addr().unwrap(), "1h").await;
+ setup_resources(&consumer_client, Durability::Persisted).await;
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let group_id = Identifier::named(CONSUMER_GROUP_NAME).unwrap();
+ consumer_client
+ .join_consumer_group(&stream, &topic, &group_id)
+ .await
+ .unwrap();
+ let before = consumer_client
+ .get_consumer_group(&stream, &topic, &group_id)
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(before.members_count, 1);
+
+ harness.kill_node(2).unwrap();
+ harness.restart_node(2).unwrap();
+ sleep(STALE_EVICTION_TIMEOUT).await;
+
+ let observer = harness.root_client_for_node(0).await.unwrap();
+ let after = observer
+ .get_consumer_group(&stream, &topic, &group_id)
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(
+ after.members_count, 1,
+ "a peer restart must preserve a live remote member"
+ );
+ assert_eq!(after.members[0].id, before.members[0].id);
+ assert_eq!(after.members[0].partitions, before.members[0].partitions);
+ consumer_client.get_me().await.unwrap();
+}
+
+async fn verify_replacement_after_restart(
+ harness: &mut integration::harness::TestHarness,
+ restart: Restart,
+) {
+ let server_addr = harness.server().raw_tcp_addr().unwrap();
+ let original = create_client(&server_addr, "500ms").await;
+ setup_resources(&original, Durability::Persisted).await;
+ let stream = Identifier::named(STREAM_NAME).unwrap();
+ let topic = Identifier::named(TOPIC_NAME).unwrap();
+ let group_id = Identifier::named(CONSUMER_GROUP_NAME).unwrap();
+ let consumer = Consumer::group(group_id.clone());
+ original
+ .join_consumer_group(&stream, &topic, &group_id)
+ .await
+ .unwrap();
+ let before = original
+ .poll_messages(
+ &stream,
+ &topic,
+ None,
+ &consumer,
+ &PollingStrategy::next(),
+ 1,
+ true,
+ )
+ .await
+ .unwrap();
+ assert_eq!(before.messages.len(), 1);
+ match restart {
+ Restart::Client => {}
+ Restart::ServerShutdown => harness.stop_node(0).unwrap(),
+ Restart::ServerCrash => harness.kill_node(0).unwrap(),
+ Restart::ClusterCrash => {
+ for node in 0..harness.cluster_size() {
+ harness.kill_node(node).unwrap();
+ }
+ }
+ }
+ original.disconnect().await.unwrap();
+ drop(original);
+ if matches!(restart, Restart::ClusterCrash) {
+ for node in 0..harness.cluster_size() {
+ harness.restart_node(node).unwrap();
+ }
+ } else if !matches!(restart, Restart::Client) {
+ harness.restart_node(0).unwrap();
+ }
+
+ let replacement = create_client(&server_addr, "500ms").await;
+ replacement
+ .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+ .await
+ .unwrap();
+ replacement
+ .join_consumer_group(&stream, &topic, &group_id)
+ .await
+ .unwrap();
+
+ // Metadata login can finish before the partition's recovered WAL is applied.
+ let deadline = tokio::time::Instant::now() + STALE_EVICTION_TIMEOUT;
+ let direct = loop {
+ let direct = replacement
+ .poll_messages(
+ &stream,
+ &topic,
+ Some(0),
+ &Consumer::default(),
+ &PollingStrategy::offset(1),
+ 1,
+ false,
+ )
+ .await
+ .unwrap();
+ if !direct.messages.is_empty() {
+ break direct;
+ }
+ assert!(
+ tokio::time::Instant::now() < deadline,
+ "retained data was not readable within {STALE_EVICTION_TIMEOUT:?}"
+ );
+ sleep(STALE_EVICTION_RETRY_INTERVAL).await;
+ };
+ assert_eq!(direct.messages.len(), 1, "retained data must be readable");
+
+ let deadline = tokio::time::Instant::now() + STALE_EVICTION_TIMEOUT;
+ loop {
+ let polled = replacement
+ .poll_messages(
+ &stream,
+ &topic,
+ None,
+ &consumer,
+ &PollingStrategy::next(),
+ 1,
+ true,
+ )
+ .await
+ .expect("replacement group poll must succeed");
+ if !polled.messages.is_empty() {
+ assert_eq!(polled.messages[0].payload, direct.messages[0].payload);
+ let group = replacement
+ .get_consumer_group(&stream, &topic, &group_id)
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(
+ group.members_count, 1,
+ "only the replacement member remains"
+ );
+ break;
+ }
+ if tokio::time::Instant::now() >= deadline {
+ let group = replacement
+ .get_consumer_group(&stream, &topic, &group_id)
+ .await
+ .unwrap()
+ .unwrap();
+ let offset = replacement
+ .get_consumer_offset(&consumer, &stream, &topic, Some(0))
+ .await
+ .unwrap();
+ panic!(
+ "replacement polled empty for {STALE_EVICTION_TIMEOUT:?} despite readable data; group={group:?}; offset={offset:?}"
+ );
+ }
+ sleep(STALE_EVICTION_RETRY_INTERVAL).await;
+ }
+}
+
async fn create_client(server_addr: &str, heartbeat_interval: &str) -> IggyClient {
let config = TcpClientConfig {
server_address: server_addr.to_string(),
@@ -71,7 +319,7 @@ async fn create_reconnecting_client(server_addr: &str) -> IggyClient {
IggyClient::create(ClientWrapper::Tcp(client), None, None)
}
-async fn setup_resources(client: &IggyClient) {
+async fn setup_resources(client: &IggyClient, durability: Durability) {
client
.login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
.await
@@ -86,6 +334,8 @@ async fn setup_resources(client: &IggyClient) {
&TopicCreateOptions {
partitions_count: Some(PARTITIONS_COUNT),
message_expiry: Some(IggyExpiry::NeverExpire),
+ durability,
+ consumer_offset_durability: durability,
..TopicCreateOptions::default()
},
)
@@ -120,6 +370,8 @@ async fn setup_resources(client: &IggyClient) {
#[iggy_harness(server(
heartbeat.enabled = true,
heartbeat.interval = "2s",
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
))]
async fn should_handle_stale_client_with_manual_reconnection(
harness: &integration::harness::TestHarness,
@@ -127,7 +379,7 @@ async fn should_handle_stale_client_with_manual_reconnection(
let server_addr = harness.server().raw_tcp_addr().unwrap();
let setup_client = create_client(&server_addr, "500ms").await;
- setup_resources(&setup_client).await;
+ setup_resources(&setup_client, Durability::default()).await;
// Client with 1h heartbeat will become stale
let stale_client = create_client(&server_addr, "1h").await;
@@ -263,6 +515,8 @@ async fn should_handle_stale_client_with_manual_reconnection(
#[iggy_harness(server(
heartbeat.enabled = true,
heartbeat.interval = "2s",
+ consumer_group.heartbeat_interval = "500ms",
+ consumer_group.session_timeout = "8s",
))]
async fn should_handle_stale_client_with_auto_reconnection(
harness: &integration::harness::TestHarness,
@@ -270,7 +524,7 @@ async fn should_handle_stale_client_with_auto_reconnection(
let server_addr = harness.server().raw_tcp_addr().unwrap();
let setup_client = create_client(&server_addr, "500ms").await;
- setup_resources(&setup_client).await;
+ setup_resources(&setup_client, Durability::default()).await;
let consumer_client = create_reconnecting_client(&server_addr).await;
// Note: auto_login is enabled in create_reconnecting_client, so no manual login needed
diff --git a/core/metadata/src/impls/metadata.rs b/core/metadata/src/impls/metadata.rs
index 6a74224fd..953f0e850 100644
--- a/core/metadata/src/impls/metadata.rs
+++ b/core/metadata/src/impls/metadata.rs
@@ -27,12 +27,12 @@ use crate::stm::user::{DeletePersonalAccessTokenRequest, Users};
use crate::stm::{ConsensusGroupAllocator, StateMachine};
use consensus::{
CLIENTS_TABLE_MAX, Canceled, ClientTable, ClientTableSnapshot, CommitLogEvent, CommitReply,
- Consensus, EvictionContext, FatalReason, Pipeline, PipelineEntry, Plane, PlaneIdentity,
- PlaneKind, PreflightOutcome, PrepareRollback, Project, ReplicaLogContext, RequestLogEvent,
- Sequencer, SessionEnd, SimEventKind, VsrConsensus, ack_preflight, ack_quorum_reached,
- apply_preflight_consensus_plane, build_eviction_message, build_reply_message,
- build_reply_message_with, build_result_rejection_reply, emit_sim_event, fatal,
- fence_old_prepare_by_commit, is_caught_up_primary,
+ Consensus, DISCONNECT_LOGOUT_REQUEST_ID, EvictionContext, FatalReason, Pipeline, PipelineEntry,
+ Plane, PlaneIdentity, PlaneKind, PreflightOutcome, PrepareRollback, Project, ReplicaLogContext,
+ RequestLogEvent, Sequencer, SessionEnd, SimEventKind, VsrConsensus, ack_preflight,
+ ack_quorum_reached, apply_preflight_consensus_plane, build_eviction_message,
+ build_reply_message, build_reply_message_with, build_result_rejection_reply, emit_sim_event,
+ fatal, fence_old_prepare_by_commit, is_caught_up_primary,
panic_if_hash_chain_would_break_in_same_view, peek_committable_head, pipeline_prepare_common,
register_preflight, replicate_preflight, replicate_to_next_in_chain, request_preflight,
send_eviction_to_client, send_prepare_ok as send_prepare_ok_common, verify_prepare_integrity,
@@ -599,7 +599,7 @@ fn require_shard_zero<'a, T>(
///
/// The backup commit walk's per-op logic, shared so the simulator's WAL
/// reconstruction reaches identical state from the same log through one apply path.
-/// Register creates or rebinds a session (no state-machine op); Logout drops the
+/// Register creates or rebinds a session and refreshes its memberships; Logout drops the
/// session and rebalances consumer groups; every other op applies to the state
/// machine and caches the reply for at-most-once dedup. `fire_notifier` runs the
/// post-commit hook (a no-op during reconstruction, before it is wired). Does not
@@ -633,7 +633,9 @@ pub fn apply_committed_prepare<M>(
{
let header = *prepare.header();
if header.operation == Operation::Register {
- // Register: commit_register creates the session, no state-machine op.
+ mux_stm
+ .streams()
+ .refresh_consumer_group_session(header.client, header.op);
if table_mutations_allowed {
let reply = build_reply_message(&header, &bytes::Bytes::new());
client_table
@@ -2313,6 +2315,68 @@ where
}
}
+ /// Remove an expired consumer-group member only if its session is unchanged.
+ /// The caller must establish timeout expiry on the caught-up metadata primary.
+ /// `None` fences legacy members with no persisted session or client-table entry.
+ /// Returns `Ok(None)` when the session changed or membership already ended.
+ ///
+ /// # Errors
+ /// Returns a submission error while consensus cannot accept or commit the logout.
+ ///
+ /// # Panics
+ /// Requires metadata shard 0 and a nonzero client id.
+ #[allow(clippy::future_not_send)]
+ pub async fn submit_expired_logout_in_process(
+ &self,
+ client_id: u128,
+ expected_session: Option<u64>,
+ ) -> Result<Option<u64>, MetadataSubmitError> {
+ assert_ne!(client_id, 0, "client_id 0 is reserved for internal use");
+ let consensus = self
+ .consensus
+ .as_ref()
+ .expect("session expiry runs on shard 0");
+ if !consensus.is_primary() || !consensus.is_normal() || consensus.is_transferring() {
+ return Err(MetadataSubmitError::NotPrimary);
+ }
+ if !is_caught_up_primary(consensus) {
+ return Err(MetadataSubmitError::NotCaughtUp);
+ }
+ if consensus.pipeline_has_message_from_client(client_id) {
+ return Err(MetadataSubmitError::InProgress);
+ }
+ if consensus.pipeline_is_full() {
+ return Err(MetadataSubmitError::PipelineFull);
+ }
+
+ if self
+ .client_table
+ .borrow()
+ .get_epoch(client_id)
+ .or_else(|| self.mux_stm.streams().consumer_group_session(client_id))
+ != expected_session
+ || !self
+ .mux_stm
+ .streams()
+ .read(|inner| inner.consumer_group_members.contains_key(&client_id))
+ {
+ return Ok(None);
+ }
+ // Do not enqueue this check: a later register must never be followed
+ // by a logout whose liveness proof belonged to the previous session.
+ let header = RoutedRequestHeader {
+ client: client_id,
+ request: DISCONNECT_LOGOUT_REQUEST_ID,
+ group: server_common::sharding::METADATA_GROUP,
+ ..RoutedRequestHeader::default()
+ };
+ let prepare = build_prepare_message(consensus, &header, Operation::Logout, &[]);
+ self.dispatch_prepare_and_await(consensus, prepare)
+ .await
+ .map(|reply| Some(reply.header().commit))
+ .map_err(|Canceled| MetadataSubmitError::Canceled)
+ }
+
/// 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
@@ -2646,11 +2710,9 @@ where
) -> Result<Message<ReplyHeader>, Canceled> {
consensus.verify_pipeline();
let receiver = consensus.pipeline_message_with_subscriber(PlaneKind::Metadata, &prepare);
- // Register is the one op whose admission requires the catch-up gate
- // (double-register epoch bump); its submit path checks the gate and
- // the check-to-dispatch section is synchronous. Non-register ops
- // dispatch mid-window by design (they pipeline behind the in-flight
- // batch, like the wire path always has).
+ // Register and expired-session Logout require the catch-up gate for
+ // their epoch checks. Both check-to-dispatch sections are synchronous.
+ // Other submits can pipeline behind the in-flight batch.
debug_assert!(
prepare.header().operation != Operation::Register || is_caught_up_primary(consensus),
"dispatch_prepare_and_await: register dispatched with the catch-up gate closed"
@@ -2870,7 +2932,9 @@ where
// the single-threaded shard and keeps the head revalidation
// sound.
let reply = if prepare_header.operation == Operation::Register {
- // Register: commit_register creates session, no SM.
+ self.mux_stm
+ .streams()
+ .refresh_consumer_group_session(prepare_header.client, prepare_header.op);
let reply = build_reply_message(&prepare_header, &bytes::Bytes::new());
if self.client_table_mutation_allowed(prepare_header.op) {
self.client_table.borrow_mut().commit_register(
@@ -4199,12 +4263,16 @@ fn unreplayable_secret_refusal(
#[cfg(test)]
mod tests {
use super::*;
- use crate::stm::stream::Streams;
+ use crate::stm::StateHandler;
+ use crate::stm::consumer_group::JoinConsumerGroupRequest;
+ use crate::stm::stream::{Streams, StreamsInner};
use crate::stm::user::Users;
use consensus::LocalPipeline;
use iggy_binary_protocol::WireOptions;
+ use iggy_binary_protocol::requests::consumer_groups::CreateConsumerGroupRequest;
+ use iggy_binary_protocol::requests::streams::CreateStreamRequest;
use iggy_binary_protocol::requests::topics::CreateTopicRequest;
- use iggy_common::variadic;
+ use iggy_common::{IggyTimestamp, variadic};
use journal::prepare_journal::PrepareJournal;
use message_bus::{
BusMessage, ClientForwardFn, ConnectionLostFn, JoinHandle, ReplicaForwardFn, SendError,
@@ -5689,6 +5757,333 @@ mod tests {
);
}
+ #[compio::test]
+ async fn expired_logout_waits_for_catchup_and_preserves_registered_members() {
+ const CLIENT: u128 = 1;
+ const USER: u32 = 7;
+ let (_dir, metadata) = metadata_with_group_member(CLIENT).await;
+ let session = metadata
+ .submit_register_in_process(CLIENT, USER)
+ .await
+ .unwrap()
+ .epoch;
+ let consensus = metadata.consensus.as_ref().unwrap();
+ consensus.advance_commit_max(session + 1);
+ assert_eq!(
+ metadata
+ .submit_expired_logout_in_process(CLIENT, Some(session))
+ .await,
+ Err(MetadataSubmitError::NotCaughtUp)
+ );
+ assert_eq!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(CLIENT)
+ .len(),
+ 1
+ );
+ }
+
+ #[compio::test]
+ async fn expired_logout_removes_only_the_observed_session() {
+ const CLIENT: u128 = 1;
+ const USER: u32 = 7;
+ let (_dir, metadata) = metadata_with_group_member(CLIENT).await;
+ let session = metadata
+ .submit_register_in_process(CLIENT, USER)
+ .await
+ .unwrap()
+ .epoch;
+ assert_eq!(
+ metadata
+ .submit_expired_logout_in_process(CLIENT, Some(session))
+ .await
+ .unwrap(),
+ Some(session + 1)
+ );
+ assert!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(CLIENT)
+ .is_empty()
+ );
+ assert_eq!(metadata.client_table.borrow().get_epoch(CLIENT), None);
+ assert_eq!(metadata.mux_stm.streams().consumer_group_count(), 1);
+ }
+
+ #[compio::test]
+ async fn expired_logout_removes_a_member_without_a_client_table_entry() {
+ const CLIENT: u128 = 1;
+ let (_dir, metadata) = metadata_with_group_member(CLIENT).await;
+
+ assert_eq!(
+ metadata
+ .submit_expired_logout_in_process(CLIENT, Some(1))
+ .await
+ .unwrap(),
+ Some(2)
+ );
+ assert_eq!(
+ metadata
+ .submit_expired_logout_in_process(CLIENT, None)
+ .await
+ .unwrap(),
+ None
+ );
+
+ assert!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(CLIENT)
+ .is_empty()
+ );
+ assert_eq!(metadata.mux_stm.streams().consumer_group_count(), 1);
+ assert_eq!(metadata.consensus.as_ref().unwrap().commit_min(), 2);
+ }
+
+ #[compio::test]
+ async fn stale_logout_preserves_a_newer_session_and_its_membership() {
+ const CLIENT: u128 = 1;
+ const OLD_SESSION: u64 = 1;
+ const USER: u32 = 7;
+ let (_dir, metadata) = metadata_with_group_member(CLIENT).await;
+ let new_session = metadata
+ .submit_register_in_process(CLIENT, USER)
+ .await
+ .unwrap()
+ .epoch;
+
+ metadata
+ .submit_logout_in_process(CLIENT, OLD_SESSION, DISCONNECT_LOGOUT_REQUEST_ID)
+ .await
+ .unwrap();
+ for expired_session in [None, Some(OLD_SESSION)] {
+ assert_eq!(
+ metadata
+ .submit_expired_logout_in_process(CLIENT, expired_session)
+ .await
+ .unwrap(),
+ None
+ );
+ }
+
+ assert_eq!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(CLIENT)
+ .len(),
+ 1
+ );
+ assert_eq!(
+ metadata.client_table.borrow().get_epoch(CLIENT),
+ Some(new_session)
+ );
+ assert_eq!(
+ metadata.consensus.as_ref().unwrap().commit_min(),
+ new_session
+ );
+ }
+
+ #[compio::test]
+ async fn expired_logout_keeps_the_member_fence_after_capacity_and_fence_eviction() {
+ const CLIENT: u128 = 1;
+ const USER: u32 = 7;
+ let (_dir, metadata) = metadata_with_group_member(CLIENT).await;
+ *metadata.client_table.borrow_mut() = ClientTable::new(1);
+ let old_session = metadata
+ .submit_register_in_process(CLIENT, USER)
+ .await
+ .unwrap()
+ .epoch;
+ let session = metadata
+ .submit_register_in_process(CLIENT, USER)
+ .await
+ .unwrap()
+ .epoch;
+ for (client, epoch) in [(CLIENT, session), (CLIENT + 1, session + 2)] {
+ if client != CLIENT {
+ assert_eq!(
+ metadata
+ .submit_register_in_process(client, USER)
+ .await
+ .unwrap()
+ .epoch,
+ epoch
+ );
+ assert_eq!(
+ metadata
+ .client_table
+ .borrow()
+ .user_id_for_session(CLIENT, session),
+ Some(USER)
+ );
+ }
+ let mut request = create_stream_request(client, 1, &format!("fence-{client}"));
+ bytemuck::checked::from_bytes_mut::<RoutedRequestHeader>(
+ &mut request.as_mut_slice()[..size_of::<RoutedRequestHeader>()],
+ )
+ .session = epoch;
+ metadata.submit_request_in_process(request).await.unwrap();
+ }
+ metadata
+ .submit_register_in_process(CLIENT + 2, USER)
+ .await
+ .unwrap();
+ assert_eq!(metadata.client_table.borrow().get_epoch(CLIENT), None);
+ assert_eq!(
+ metadata
+ .client_table
+ .borrow()
+ .user_id_for_session(CLIENT, session),
+ None
+ );
+ assert_eq!(
+ metadata.mux_stm.streams().consumer_group_session(CLIENT),
+ Some(session)
+ );
+ for stale in [None, Some(old_session)] {
+ assert_eq!(
+ metadata
+ .submit_expired_logout_in_process(CLIENT, stale)
+ .await
+ .unwrap(),
+ None
+ );
+ }
+ assert!(
+ metadata
+ .submit_expired_logout_in_process(CLIENT, Some(session))
+ .await
+ .unwrap()
+ .is_some()
+ );
+ assert!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(CLIENT)
+ .is_empty()
+ );
+ }
+
+ #[compio::test]
+ async fn replayed_register_refreshes_member_epoch_even_below_the_table_frontier() {
+ const CLIENT: u128 = 1;
+ let (_dir, metadata) = metadata_with_group_member(CLIENT).await;
+ let request = RoutedRequestHeader {
+ client: CLIENT,
+ ..Default::default()
+ };
+ let prepare = build_prepare_message(
+ metadata.consensus.as_ref().unwrap(),
+ &request,
+ Operation::Register,
+ &[],
+ );
+ let epoch = prepare.header().op;
+ apply_committed_prepare(
+ &*metadata.mux_stm,
+ &metadata.client_table,
+ false,
+ |_| {},
+ prepare,
+ );
+ assert_eq!(metadata.client_table.borrow().get_epoch(CLIENT), None);
+ assert_eq!(
+ metadata.mux_stm.streams().consumer_group_session(CLIENT),
+ Some(epoch)
+ );
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn metadata_with_group_member(
+ client_id: u128,
+ ) -> (
+ tempfile::TempDir,
+ IggyMetadata<VsrConsensus<NoopBus>, PrepareJournal, (), TestMux>,
+ ) {
+ const USER: u32 = 7;
+ let mut inner = StreamsInner::new();
+ let timestamp = IggyTimestamp::now();
+ let _ = StateHandler::apply(
+ &CreateStreamRequest {
+ name: WireName::new("stream").unwrap(),
+ options: WireOptions::empty(),
+ },
+ &mut inner,
+ timestamp,
+ );
+ let _ = StateHandler::apply(
+ &PersistedCreateTopicRequest {
+ request: CreateTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ name: WireName::new("topic").unwrap(),
+ options: WireOptions::empty(),
+ },
+ created_view: 0,
+ derived_options: WireOptions::empty(),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ },
+ &mut inner,
+ timestamp,
+ );
+ let _ = StateHandler::apply(
+ &CreateConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ name: WireName::new("group").unwrap(),
+ },
+ &mut inner,
+ timestamp,
+ );
+ let _ = StateHandler::apply(
+ &JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ group_id: WireIdentifier::numeric(0),
+ client_id,
+ in_flight: Vec::new(),
+ session: None,
+ },
+ &mut inner,
+ timestamp,
+ );
+ let dir = tempfile::tempdir().unwrap();
+ let mut metadata = metadata_plane();
+ metadata.journal = Some(
+ PrepareJournal::open(&dir.path().join("journal.wal"), 0)
+ .await
+ .unwrap(),
+ );
+ metadata.mux_stm = Rc::new(TestMux::new((Users::default(), (inner.into(), ()))));
+ metadata
+ .submit_register_in_process(client_id, USER)
+ .await
+ .unwrap();
+ metadata.client_table.borrow_mut().remove_client(
+ client_id,
+ USER,
+ SessionEnd::DisconnectCleanup,
+ );
+ assert_eq!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(client_id)
+ .len(),
+ 1
+ );
+ (dir, metadata)
+ }
+
/// Register is the one op that still honors the catch-up gate (its
/// admission races a committed-but-unapplied register; a double commit
/// bumps the epoch past the first reply's and fences a live client).
diff --git a/core/metadata/src/impls/recovery.rs b/core/metadata/src/impls/recovery.rs
index 60ffcf58f..bd22c4d32 100644
--- a/core/metadata/src/impls/recovery.rs
+++ b/core/metadata/src/impls/recovery.rs
@@ -358,7 +358,7 @@ pub async fn recover<M>(
journal_slots: usize,
clients_table_max: usize,
seed_baseline: impl FnOnce(&M),
- on_replayed_logout: impl Fn(&M, u128, iggy_common::IggyTimestamp),
+ on_replayed_session: impl Fn(&M, &PrepareHeader),
) -> Result<RecoveredMetadata<M>, RecoveryError>
where
M: StateMachine<Input = Message<PrepareHeader>, Error = IggyError>
@@ -589,8 +589,8 @@ where
continue;
}
- // Register/Logout mutate the client table and skip the state
- // machine, mirroring the commit paths (`on_ack` / `commit_journal`).
+ // Register/Logout apply their session effects through the callback,
+ // mirroring the commit paths (`on_ack` / `commit_journal`).
//
// Epochs come back identical on every replica: they are the register's
// own commit op, so replay reads them out of the log rather than
@@ -614,6 +614,7 @@ where
if header.operation == Operation::Register {
let reply = build_reply_message(header, &bytes::Bytes::new());
client_table.commit_register(header.client, header.user_id, reply);
+ on_replayed_session(&mux_stm, header);
last_applied_op = Some(header.op);
continue;
}
@@ -627,11 +628,7 @@ where
// commit path: the caller drops the client from its consumer
// groups (`remove_consumer_group_member`) so replay and live
// apply converge on the same group membership.
- on_replayed_logout(
- &mux_stm,
- header.client,
- iggy_common::IggyTimestamp::from(header.timestamp),
- );
+ on_replayed_session(&mux_stm, header);
last_applied_op = Some(header.op);
continue;
}
@@ -967,7 +964,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -993,7 +990,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1029,7 +1026,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1087,7 +1084,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1147,7 +1144,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await;
@@ -1198,7 +1195,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1236,7 +1233,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1306,7 +1303,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1383,7 +1380,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1449,7 +1446,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1506,7 +1503,11 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, client, _| replayed_logouts.borrow_mut().push(client),
+ |_, header| {
+ if header.operation == Operation::Logout {
+ replayed_logouts.borrow_mut().push(header.client);
+ }
+ },
)
.await
.unwrap();
@@ -1574,7 +1575,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await;
assert!(matches!(
@@ -1607,7 +1608,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await;
assert!(matches!(
@@ -1639,7 +1640,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1692,7 +1693,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await;
match result {
@@ -1733,7 +1734,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
{
@@ -1798,7 +1799,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
@@ -1851,7 +1852,7 @@ mod tests {
journal::prepare_journal::DEFAULT_SLOT_COUNT,
CLIENTS_TABLE_MAX,
|_| {},
- |_, _, _| {},
+ |_, _| {},
)
.await
.unwrap();
diff --git a/core/metadata/src/stm/consumer_group.rs b/core/metadata/src/stm/consumer_group.rs
index 0e70642bb..3b3923ffd 100644
--- a/core/metadata/src/stm/consumer_group.rs
+++ b/core/metadata/src/stm/consumer_group.rs
@@ -45,6 +45,7 @@ use iggy_binary_protocol::requests::consumer_groups::{
};
use iggy_binary_protocol::responses::consumer_groups::consumer_group_response::ConsumerGroupResponse;
use iggy_binary_protocol::responses::consumer_groups::get_consumer_group::ConsumerGroupDetailsResponse;
+use iggy_common::IggyTimestamp;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
@@ -66,6 +67,8 @@ pub struct PendingRevocation {
pub struct ConsumerGroupMember {
pub id: usize,
pub client_id: u128,
+ /// Retained independently of client-table capacity eviction. None for legacy snapshots.
+ pub session: Option<u64>,
pub partitions: Vec<usize>,
/// Partitions marked for cooperative handoff away from this member but not
/// yet moved. The member still owns them (poll + commit) until completion.
@@ -78,6 +81,7 @@ impl ConsumerGroupMember {
Self {
id,
client_id,
+ session: None,
partitions: Vec::new(),
pending_revocations: Vec::new(),
}
@@ -392,6 +396,8 @@ pub struct JoinConsumerGroupRequest {
pub group_id: WireIdentifier,
pub client_id: u128,
pub in_flight: Vec<u32>,
+ /// Authenticated bind epoch, absent only in legacy records. `PrepareHeader` does not retain it.
+ pub session: Option<u64>,
}
impl WireEncode for JoinConsumerGroupRequest {
@@ -402,6 +408,7 @@ impl WireEncode for JoinConsumerGroupRequest {
+ 16
+ 4
+ self.in_flight.len() * 4
+ + self.session.map_or(0, |_| size_of::<u64>())
}
fn encode(&self, buf: &mut BytesMut) {
@@ -416,6 +423,9 @@ impl WireEncode for JoinConsumerGroupRequest {
for partition_id in &self.in_flight {
buf.put_u32_le(*partition_id);
}
+ if let Some(session) = self.session {
+ buf.put_u64_le(session);
+ }
}
}
@@ -440,6 +450,14 @@ impl WireDecode for JoinConsumerGroupRequest {
in_flight.push(read_u32_le(buf, pos)?);
pos += 4;
}
+ // Old WAL records end after in_flight; new joins retain the request epoch.
+ let session = if pos == buf.len() {
+ None
+ } else {
+ let session = read_u64_le(buf, pos)?;
+ pos += size_of::<u64>();
+ Some(session).filter(|&epoch| epoch != 0)
+ };
Ok((
Self {
stream_id,
@@ -447,6 +465,7 @@ impl WireDecode for JoinConsumerGroupRequest {
group_id,
client_id,
in_flight,
+ session,
},
pos,
))
@@ -591,7 +610,7 @@ impl StateHandler for DeleteConsumerGroupRequest {
// offsets on the topic's surviving partitions.
state.revision = state.revision.wrapping_add(1);
// The dropped group may have held pending revocations.
- state.recompute_pending_revocations_count();
+ state.recompute_consumer_group_metadata();
ApplyReply::ok(Bytes::new())
}
}
@@ -627,15 +646,17 @@ impl StateHandler for JoinConsumerGroupRequest {
// 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 {
+ .iter_mut()
+ .find(|(_, member)| member.client_id == self.client_id);
+ if let Some((_, member)) = already {
+ member.session = self.session.or(member.session);
return ApplyReply::ok(Bytes::new());
}
let member_key = group
.members
.insert(ConsumerGroupMember::new(0, self.client_id));
group.members[member_key].id = member_key;
+ group.members[member_key].session = self.session;
// 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
@@ -644,7 +665,7 @@ impl StateHandler for JoinConsumerGroupRequest {
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();
+ state.recompute_consumer_group_metadata();
ApplyReply::ok(Bytes::new())
}
}
@@ -692,7 +713,7 @@ impl StateHandler for LeaveConsumerGroupRequest {
};
group.members.remove(key);
group.rebalance_members(&partition_ids);
- state.recompute_pending_revocations_count();
+ state.recompute_consumer_group_metadata();
ApplyReply::ok(Bytes::new())
}
}
@@ -757,7 +778,7 @@ impl StateHandler for RemoveConsumerGroupMemberRequest {
// Only perturb the reconciler when a membership actually changed.
if any_removed {
state.revision = state.revision.wrapping_add(1);
- state.recompute_pending_revocations_count();
+ state.recompute_consumer_group_metadata();
}
ApplyReply::ok(Bytes::new())
}
@@ -833,7 +854,35 @@ impl StateHandler for CompleteConsumerGroupRevocationRequest {
});
if completed {
state.revision = state.revision.wrapping_add(1);
- state.recompute_pending_revocations_count();
+ state.recompute_consumer_group_metadata();
+ }
+ ApplyReply::ok(Bytes::new())
+ }
+}
+
+/// A committed Register refreshes all memberships of that identity.
+#[derive(Debug, Clone)]
+pub struct RefreshConsumerGroupSessionRequest {
+ pub client_id: u128,
+ pub session: u64,
+}
+
+impl StateHandler for RefreshConsumerGroupSessionRequest {
+ type State = StreamsInner;
+
+ fn apply(&self, state: &mut StreamsInner, _timestamp: IggyTimestamp) -> ApplyReply {
+ if let Some(memberships) = state.consumer_group_members.get(&self.client_id) {
+ for &(stream_id, topic_id, group_id, member_id) in memberships {
+ if let Some(member) = state
+ .items
+ .get_mut(stream_id)
+ .and_then(|stream| stream.topics.get_mut(topic_id))
+ .and_then(|topic| topic.consumer_groups.get_mut(&group_id))
+ .and_then(|group| group.members.get_mut(member_id))
+ {
+ member.session = Some(self.session);
+ }
+ }
}
ApplyReply::ok(Bytes::new())
}
@@ -856,6 +905,8 @@ pub struct ConsumerGroupMemberSnapshot {
/// trailing element).
#[serde(default)]
pub pending_revocations: Vec<(usize, usize, u64)>,
+ #[serde(default)]
+ pub session: Option<u64>,
}
/// Consumer group snapshot representation for serialization (nested under the
@@ -884,6 +935,7 @@ impl ConsumerGroupSnapshot {
ConsumerGroupMemberSnapshot {
id: member.id,
client_id: member.client_id,
+ session: member.session,
partitions: member.partitions.clone(),
pending_revocations: member
.pending_revocations
@@ -919,6 +971,7 @@ impl ConsumerGroupSnapshot {
ConsumerGroupMember {
id: member_snap.id,
client_id: member_snap.client_id,
+ session: member_snap.session,
partitions: member_snap.partitions,
pending_revocations: member_snap
.pending_revocations
@@ -947,14 +1000,182 @@ impl ConsumerGroupSnapshot {
#[cfg(test)]
mod tests {
use super::*;
+ use crate::stm::snapshot::Snapshotable;
+ use crate::stm::stream::Streams;
use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment;
- use iggy_binary_protocol::requests::streams::CreateStreamRequest;
+ use iggy_binary_protocol::requests::streams::{CreateStreamRequest, DeleteStreamRequest};
use iggy_binary_protocol::requests::topics::{
- CreateTopicRequest, CreateTopicWithAssignmentsRequest,
+ CreateTopicRequest, CreateTopicWithAssignmentsRequest, DeleteTopicRequest,
};
use iggy_binary_protocol::{WireName, WireOptions};
use iggy_common::IggyTimestamp;
+ #[test]
+ fn session_refresh_index_tracks_membership_removal_and_rejoin() {
+ const CLIENT: u128 = 7;
+ const OTHER: u128 = 8;
+ let mut state = streams_with_topic();
+ assert_eq!(create_group(&mut state, "first").code, 0);
+ assert_eq!(create_group(&mut state, "second").code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, CLIENT).code, 0);
+ assert_eq!(join(&mut state, 0, 0, 1, CLIENT).code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, CLIENT).code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, OTHER).code, 0);
+ assert_eq!(state.consumer_group_members[&CLIENT].len(), 2);
+ assert_eq!(leave(&mut state, 0, 0, 0, CLIENT).code, 0);
+ assert_eq!(state.consumer_group_members[&CLIENT].len(), 1);
+ assert_eq!(
+ StateHandler::apply(
+ &DeleteConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ group_id: WireIdentifier::numeric(1),
+ },
+ &mut state,
+ IggyTimestamp::now(),
+ )
+ .code,
+ 0
+ );
+ assert!(!state.consumer_group_members.contains_key(&CLIENT));
+ assert_eq!(
+ StateHandler::apply(
+ &RemoveConsumerGroupMemberRequest { client_id: OTHER },
+ &mut state,
+ IggyTimestamp::now(),
+ )
+ .code,
+ 0
+ );
+ assert!(state.consumer_group_members.is_empty());
+ assert_eq!(join(&mut state, 0, 0, 0, CLIENT).code, 0);
+ let streams = Streams::from(state);
+ streams.refresh_consumer_group_session(CLIENT, 11);
+ assert_eq!(streams.consumer_group_session(CLIENT), Some(11));
+ assert_eq!(streams.consumer_group_session(OTHER), None);
+ }
+
+ #[test]
+ fn deleting_a_parent_removes_its_session_refresh_index() {
+ const CLIENT: u128 = 7;
+ for delete_stream in [false, true] {
+ let mut state = streams_with_topic();
+ assert_eq!(create_group(&mut state, "group").code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, CLIENT).code, 0);
+ let reply = if delete_stream {
+ StateHandler::apply(
+ &DeleteStreamRequest {
+ stream_id: WireIdentifier::numeric(0),
+ },
+ &mut state,
+ IggyTimestamp::now(),
+ )
+ } else {
+ StateHandler::apply(
+ &DeleteTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ },
+ &mut state,
+ IggyTimestamp::now(),
+ )
+ };
+ assert_eq!(reply.code, 0);
+ assert!(state.consumer_group_members.is_empty());
+ }
+ }
+
+ #[test]
+ fn session_refresh_preserves_other_clients_after_boot_and_in_place_restore() {
+ const CLIENT: u128 = 7;
+ const OTHER: u128 = 8;
+ let mut state = streams_with_topic();
+ assert_eq!(create_group(&mut state, "first").code, 0);
+ assert_eq!(create_group(&mut state, "second").code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, CLIENT).code, 0);
+ assert_eq!(join(&mut state, 0, 0, 1, CLIENT).code, 0);
+ assert_eq!(join(&mut state, 0, 0, 0, OTHER).code, 0);
+ let original = Streams::from(state);
+ let boot = Streams::from_snapshot(original.to_snapshot()).unwrap();
+ let mut restored = StreamsInner::new();
+ restored.restore_in_place(original.to_snapshot());
+ for streams in [boot, Streams::from(restored)] {
+ for epoch in [11, 12, 13] {
+ streams.refresh_consumer_group_session(CLIENT, epoch);
+ streams.refresh_consumer_group_session(u128::MAX, epoch);
+ assert!(streams.read(|inner| {
+ let topic = &inner.items[0].topics[0];
+ for group in topic.consumer_groups.values() {
+ for (_, member) in &group.members {
+ assert_eq!(
+ member.session,
+ (member.client_id == CLIENT).then_some(epoch)
+ );
+ }
+ }
+ !inner.consumer_group_members.contains_key(&u128::MAX)
+ }));
+ }
+ }
+ }
+
+ #[test]
+ fn member_session_snapshot_round_trip_preserves_fence_and_reads_legacy_members() {
+ const CLIENT: u128 = 7;
+ const SESSION: u64 = 11;
+ let mut group = ConsumerGroup::new(0, Arc::from("group"));
+ let mut member = ConsumerGroupMember::new(0, CLIENT);
+ member.session = Some(SESSION);
+ member.partitions.push(2);
+ group.members.insert(member);
+ let snapshot = ConsumerGroupSnapshot::from_group(&group);
+ let encoded = rmp_serde::to_vec(&snapshot).unwrap();
+ let restored = rmp_serde::from_slice::<ConsumerGroupSnapshot>(&encoded)
+ .unwrap()
+ .into_group();
+ assert_eq!(restored.members[0].session, Some(SESSION));
+ assert_eq!(restored.members[0].partitions, [2]);
+
+ let legacy = rmp_serde::to_vec(&(
+ 0usize,
+ CLIENT,
+ vec![2usize],
+ Vec::<(usize, usize, u64)>::new(),
+ ))
+ .unwrap();
+ let restored = rmp_serde::from_slice::<ConsumerGroupMemberSnapshot>(&legacy).unwrap();
+ assert_eq!(restored.session, None);
+ assert_eq!(restored.client_id, CLIENT);
+ assert_eq!(restored.partitions, [2]);
+ }
+
+ #[test]
+ fn replicated_join_preserves_session_and_decodes_legacy_payloads() {
+ let mut join = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ group_id: WireIdentifier::numeric(0),
+ client_id: 7,
+ in_flight: vec![1, 2],
+ session: None,
+ };
+ let legacy = join.to_bytes();
+ assert_eq!(
+ JoinConsumerGroupRequest::decode_from(&legacy)
+ .unwrap()
+ .session,
+ None
+ );
+ join.session = Some(11);
+ let encoded = join.to_bytes();
+ let decoded = JoinConsumerGroupRequest::decode_from(&encoded).unwrap();
+ assert_eq!(decoded.session, join.session);
+ assert_eq!(decoded.in_flight, join.in_flight);
+ for length in legacy.len() + 1..encoded.len() {
+ assert!(JoinConsumerGroupRequest::decode_from(&encoded[..length]).is_err());
+ }
+ }
+
// Groups co-locate in the topic node, so an apply resolves its parent through
// `topic_mut`. Build a `StreamsInner` holding stream 0 / topic 0 via the same
// handlers the commit path drives.
@@ -1012,6 +1233,7 @@ mod tests {
group_id: WireIdentifier::numeric(group),
client_id,
in_flight: Vec::new(),
+ session: None,
},
state,
IggyTimestamp::now(),
diff --git a/core/metadata/src/stm/mod.rs b/core/metadata/src/stm/mod.rs
index 833f196cc..b2bf76a30 100644
--- a/core/metadata/src/stm/mod.rs
+++ b/core/metadata/src/stm/mod.rs
@@ -384,12 +384,14 @@ macro_rules! define_state {
/// # Requirements
/// Each listed operation must have a corresponding `{Operation}Request` wire type
/// that implements `WireDecode` and `StateHandler<State = {$state}Inner>`.
+/// Internal commands only implement `StateHandler` and are never decoded from wire data.
#[macro_export]
macro_rules! collect_handlers {
(
$state:ident {
$($operation:ident),* $(,)?
}
+ $(internal { $($internal:ident),* $(,)? })?
) => {
paste::paste! {
#[derive(Debug, Clone)]
@@ -397,6 +399,9 @@ macro_rules! collect_handlers {
$(
$operation([<$operation Request>], ::iggy_common::IggyTimestamp),
)*
+ $($(
+ $internal([<$internal Request>], ::iggy_common::IggyTimestamp),
+ )*)?
/// Replace the whole state from a snapshot section, in place.
/// Never parsed off the wire (state transfer installs it via
/// `RestoreSnapshotInPlace`); absorbed on both left-right
@@ -448,6 +453,11 @@ macro_rules! collect_handlers {
$crate::stm::StateHandler::apply(payload, self, *ts)
},
)*
+ $($(
+ [<$state Command>]::$internal(payload, ts) => {
+ $crate::stm::StateHandler::apply(payload, self, *ts)
+ },
+ )*)?
[<$state Command>]::RestoreSnapshot(snapshot) => {
self.restore_in_place(snapshot.clone());
$crate::stm::result::ApplyReply::default()
diff --git a/core/metadata/src/stm/stream.rs b/core/metadata/src/stm/stream.rs
index 7a46b9e15..f6715128a 100644
--- a/core/metadata/src/stm/stream.rs
+++ b/core/metadata/src/stm/stream.rs
@@ -18,7 +18,8 @@
use crate::stm::StateHandler;
use crate::stm::consumer_group::{
CompleteConsumerGroupRevocationRequest, ConsumerGroup, ConsumerGroupSnapshot,
- JoinConsumerGroupRequest, LeaveConsumerGroupRequest, RemoveConsumerGroupMemberRequest,
+ JoinConsumerGroupRequest, LeaveConsumerGroupRequest, RefreshConsumerGroupSessionRequest,
+ RemoveConsumerGroupMemberRequest,
};
use crate::stm::id_slab::IdSlab;
use crate::stm::result::{
@@ -858,6 +859,9 @@ define_state! {
// decide whether to wake the reconciler. Deterministic (same ops, same
// recompute on every replica).
pending_revocations_count: u64,
+ // Derived with the revocation count after membership changes and restore.
+ // Locations are (stream, topic, group, member); never persisted.
+ consumer_group_members: AHashMap<u128, Vec<(usize, usize, u64, usize)>>,
// Shared aggregate stats, one `Arc` per stream/topic across both
// left-right buffers (see `StatsRegistry`). Not snapshotted -- rebuilt
// as streams/topics restore.
@@ -994,27 +998,33 @@ collect_handlers! {
CompleteConsumerGroupRevocation,
TruncatePartition,
}
+ internal { RefreshConsumerGroupSession }
}
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) {
+ /// Rebuild derived group metadata after membership/revocation changes or restore.
+ /// Session refreshes use the membership index without rescanning unrelated clients.
+ pub(crate) fn recompute_consumer_group_metadata(&mut self) {
let mut count: u64 = 0;
- for (_, stream) in &self.items {
- for (_, topic) in &stream.topics {
+ for memberships in self.consumer_group_members.values_mut() {
+ memberships.clear();
+ }
+ for (stream_id, stream) in &self.items {
+ for (topic_id, topic) in &stream.topics {
for group in topic.consumer_groups.values() {
- for (_, member) in &group.members {
+ for (member_id, member) in &group.members {
count += member.pending_revocations.len() as u64;
+ self.consumer_group_members
+ .entry(member.client_id)
+ .or_default()
+ .push((stream_id, topic_id, group.id, member_id));
}
}
}
}
self.pending_revocations_count = count;
+ self.consumer_group_members
+ .retain(|_, memberships| !memberships.is_empty());
}
/// Resolve a wire stream identifier to its committed slab id, `None` when
@@ -1553,6 +1563,48 @@ impl Streams {
})
}
+ /// The persisted session fence survives client-table eviction and replica restart.
+ #[must_use]
+ pub fn consumer_group_session(&self, client_id: u128) -> Option<u64> {
+ self.inner.read(|inner| {
+ inner
+ .consumer_group_members
+ .get(&client_id)?
+ .iter()
+ .filter_map(|&(stream_id, topic_id, group_id, member_id)| {
+ inner
+ .items
+ .get(stream_id)?
+ .topics
+ .get(topic_id)?
+ .consumer_groups
+ .get(&group_id)?
+ .members
+ .get(member_id)?
+ .session
+ })
+ .max()
+ })
+ }
+
+ /// Apply the session fence from a committed Register, including on replay.
+ pub fn refresh_consumer_group_session(&self, client_id: u128, session: u64) {
+ if session == 0
+ || !self
+ .inner
+ .read(|inner| inner.consumer_group_members.contains_key(&client_id))
+ {
+ return;
+ }
+ let cmd = StreamsCommand::RefreshConsumerGroupSession(
+ RefreshConsumerGroupSessionRequest { client_id, session },
+ IggyTimestamp::from(0),
+ );
+ if let Err(error) = self.inner.try_apply(cmd) {
+ tracing::error!(client_id, %error, "consumer session refresh dispatched to reader-only Streams STM");
+ }
+ }
+
/// 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
@@ -2069,7 +2121,7 @@ impl StateHandler for DeleteStreamRequest {
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();
+ state.recompute_consumer_group_metadata();
ApplyReply::ok(Bytes::new())
}
}
@@ -2399,7 +2451,7 @@ impl StateHandler for DeleteTopicRequest {
.remove_topic(stream_id, topic_id, &partition_ids);
state.revision = state.revision.wrapping_add(1);
// The dropped topic may have held groups with pending revocations.
- state.recompute_pending_revocations_count();
+ state.recompute_consumer_group_metadata();
ApplyReply::ok(Bytes::new())
}
}
@@ -2828,10 +2880,11 @@ impl StreamsInner {
revision: snapshot.revision,
// Recomputed from the restored groups just below.
pending_revocations_count: 0,
+ consumer_group_members: AHashMap::new(),
last_result: None,
stats_registry,
};
- inner.recompute_pending_revocations_count();
+ inner.recompute_consumer_group_metadata();
inner
}
}
diff --git a/core/server/config.toml b/core/server/config.toml
index 2332d6a30..33fcb3a65 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -18,10 +18,22 @@
# Root directory for server data. Environment override: IGGY_PATH.
path = "local_data"
-# Configuration for consumer group cooperative partition rebalancing.
+# Consumer group rebalancing and session liveness.
[consumer_group]
# Maximum time a partition can remain in pending revocation before being force-transferred to the target member.
rebalancing_timeout = "30s"
+# The metadata primary removes members whose hosting server stops reporting their
+# connections for this long, including after a crash. An isolated live consumer
+# loses its session and must log in again. Each new primary grants a fresh timeout.
+# Expiry is always enabled. Incomplete client lists defer expiry across the cluster
+# until a full timeout has passed without an incomplete report.
+# Must exceed cluster.heartbeat_timeout when clustering is enabled, and
+# three heartbeat_interval periods plus two gather (3s) and send (100ms)
+# deadlines, allowing one missed report before expiry.
+session_timeout = "30s"
+# Server-to-server session reports run even when client heartbeat checks are disabled.
+# Must be nonzero. Use the same settings on all nodes.
+heartbeat_interval = "5s"
[data_maintenance.messages]
# Enables or disables the segment cleaner. It deletes the oldest sealed segments of
@@ -1050,8 +1062,9 @@ transfer_artifact_bytes_max = "1088 MiB"
# call. Hard upper bound: IOV_MAX/2 = 512 on Linux.
max_batch = 256
-# Wire-level cap on a single framed message. Boot refuses a value above the
-# frozen 256 MiB ceiling: segment recovery derives fixed scan and allocation
+# Wire-level cap on a single framed message. Must fit a consumer session report
+# (at least 280 B). Boot refuses a value above the frozen 256 MiB ceiling:
+# segment recovery derives fixed scan and allocation
# limits from the widest LEGAL record, so batches admitted above the ceiling
# would be refused as implausible on a later boot. Raising it past 64 MiB
# also breaks Go clients, whose frame cap is a hard 64 MiB constant.
diff --git a/core/server/src/boot/mod.rs b/core/server/src/boot/mod.rs
index 8d7ad6b0a..2463dbc52 100644
--- a/core/server/src/boot/mod.rs
+++ b/core/server/src/boot/mod.rs
@@ -115,6 +115,8 @@ where
SB: SuperblockStore + 'static,
{
let sessions = Rc::new(RefCell::new(SessionManager::new()));
+ let consumer_group_liveness = Rc::default();
+ let session_timeout = server_config.consumer_group.session_timeout.get_duration();
ShellHandlers {
on_replica_message: make_deferred_replica_message_handler(shard_handle),
on_client_request: make_deferred_client_request_handler(
@@ -124,9 +126,14 @@ where
server_config,
max_tokens_per_user,
),
- on_metadata_submit: make_metadata_submit_handler(shard_handle),
+ on_metadata_submit: make_metadata_submit_handler(
+ shard_handle,
+ &consumer_group_liveness,
+ session_timeout,
+ ),
on_list_clients: make_list_clients_handler(&sessions),
sessions,
+ consumer_group_liveness,
}
}
@@ -529,10 +536,17 @@ async fn shard_main(
|mux_stm| {
ensure_default_root_user(mux_stm);
},
- |mux_stm, client, stamp| {
- mux_stm
- .streams()
- .remove_consumer_group_member(client, stamp);
+ |mux_stm, header| {
+ if header.operation == iggy_binary_protocol::Operation::Register {
+ mux_stm
+ .streams()
+ .refresh_consumer_group_session(header.client, header.op);
+ } else {
+ mux_stm.streams().remove_consumer_group_member(
+ header.client,
+ iggy_common::IggyTimestamp::from(header.timestamp),
+ );
+ }
},
)
.await
@@ -668,6 +682,7 @@ async fn shard_main(
let ShardBuild {
shard,
sessions,
+ consumer_group_liveness,
on_client_request,
shard_handle,
} = Box::pin(build_shard_for_thread(
@@ -862,12 +877,13 @@ async fn shard_main(
} else {
None
};
- let stop_signals = StopSignals {
+ let mut stop_signals = StopSignals {
pump: stop_tx,
reconciler: reconcile_stop_tx,
heartbeat: heartbeat_stop_tx,
pat_cleaner: pat_cleaner_stop,
segment_cleaner: segment_cleaner_stop,
+ consumer_group_liveness: None,
};
// One keep-alive per process, so shard 0 owns it. Started before the
@@ -978,6 +994,25 @@ async fn shard_main(
systemd::notify_ready();
}
+ if shard_id == 0 {
+ let (stop_tx, stop_rx) = channel(1);
+ let cleaner_shard = Rc::clone(&shard);
+ let interval = config.consumer_group.heartbeat_interval.get_duration();
+ let timeout = config.consumer_group.session_timeout.get_duration();
+ let handle = compio::runtime::spawn(async move {
+ crate::consumer_group::liveness::run(
+ cleaner_shard,
+ consumer_group_liveness,
+ stop_rx,
+ interval,
+ timeout,
+ )
+ .await;
+ });
+ bus.track_background(handle);
+ stop_signals.consumer_group_liveness = Some(stop_tx);
+ }
+
bus.token().wait().await;
#[cfg(feature = "systemd")]
if shard_id == 0 {
diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs
index 9f3dd9628..cc5818e48 100644
--- a/core/server/src/boot/recovery.rs
+++ b/core/server/src/boot/recovery.rs
@@ -19,6 +19,7 @@
use crate::boot::topology::{RosterCells, TcpTopology, build_cluster_roster};
use crate::boot::wire_shell_handlers;
+use crate::consumer_group::liveness::ConsumerGroupLiveness;
use crate::partition_helpers::load_partition_or_fence;
use crate::server_error::ServerError;
use crate::session_manager::SessionManager;
@@ -62,6 +63,7 @@ use tracing::{error, info, warn};
pub(in crate::boot) struct ShardBuild {
pub shard: Rc<ServerShard>,
pub sessions: Rc<RefCell<SessionManager>>,
+ pub consumer_group_liveness: Rc<RefCell<ConsumerGroupLiveness>>,
pub on_client_request: RequestHandler,
pub shard_handle: ShellShardHandle<Rc<IggyMessageBus>, PrepareJournal, IggySnapshot>,
}
@@ -259,6 +261,7 @@ pub(in crate::boot) async fn build_shard_for_thread(
on_metadata_submit,
on_list_clients,
sessions,
+ consumer_group_liveness,
} = wire_shell_handlers(
&bus,
&shard_handle,
@@ -328,6 +331,7 @@ pub(in crate::boot) async fn build_shard_for_thread(
Ok(ShardBuild {
shard,
sessions,
+ consumer_group_liveness,
on_client_request,
shard_handle,
})
diff --git a/core/server/src/boot/threads.rs b/core/server/src/boot/threads.rs
index 79760ede0..29d834284 100644
--- a/core/server/src/boot/threads.rs
+++ b/core/server/src/boot/threads.rs
@@ -845,6 +845,7 @@ pub(in crate::boot) struct StopSignals {
pub(in crate::boot) heartbeat: Option<Sender<()>>,
pub(in crate::boot) pat_cleaner: Option<Sender<()>>,
pub(in crate::boot) segment_cleaner: Option<Sender<()>>,
+ pub(in crate::boot) consumer_group_liveness: Option<Sender<()>>,
}
impl StopSignals {
@@ -852,9 +853,14 @@ impl StopSignals {
pub(in crate::boot) fn fire(&self) {
let _ = self.pump.try_send(());
let _ = self.reconciler.try_send(());
- for stop in [&self.heartbeat, &self.pat_cleaner, &self.segment_cleaner]
- .into_iter()
- .flatten()
+ for stop in [
+ &self.heartbeat,
+ &self.pat_cleaner,
+ &self.segment_cleaner,
+ &self.consumer_group_liveness,
+ ]
+ .into_iter()
+ .flatten()
{
let _ = stop.try_send(());
}
diff --git a/core/server/src/consumer_group.rs b/core/server/src/consumer_group.rs
index 47d3d3f72..c74530874 100644
--- a/core/server/src/consumer_group.rs
+++ b/core/server/src/consumer_group.rs
@@ -25,6 +25,8 @@
//! primary enriches the op here before replication, mirroring the PAT mint
//! in [`crate::pat`] and the password hash in [`crate::users`].
+pub mod liveness;
+
use crate::namespace::{resolve_offset_group_id, resolve_partition_namespace};
use crate::shell::{ShellBus, ShellShard};
use crate::wire::{request_body, rewrite_request_body};
@@ -85,6 +87,7 @@ where
group_id: wire.group_id,
client_id,
in_flight,
+ session: Some(request.header().session),
}
.to_bytes()
}
@@ -881,6 +884,7 @@ mod tests {
group_id: GROUP_ID,
client_id: FIRST_CLIENT,
in_flight: Vec::new(),
+ session: None,
}
.to_bytes(),
))
diff --git a/core/server/src/consumer_group/liveness.rs b/core/server/src/consumer_group/liveness.rs
new file mode 100644
index 000000000..2496d37e3
--- /dev/null
+++ b/core/server/src/consumer_group/liveness.rs
@@ -0,0 +1,1162 @@
+// 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.
+
+//! Volatile, primary-owned leases for persisted consumer-group memberships.
+//! A new primary and each newly observed session get a full timeout. Only
+//! server-observed connections renew leases, never recovered client-table rows.
+
+use std::cell::RefCell;
+use std::collections::BTreeMap;
+use std::rc::Rc;
+use std::time::{Duration, Instant};
+
+use bytes::BytesMut;
+use consensus::{Consensus, MetadataHandle};
+use iggy_binary_protocol::{
+ Command, ConsensusHeader, ConsumerSession, ConsumerSessionHeartbeatHeader, HEADER_SIZE,
+ MAX_CONSUMER_SESSIONS_PER_HEARTBEAT, PrepareHeader, WireDecode, WireEncode,
+};
+use journal::superblock::SuperblockStore;
+use journal::{Journal, JournalHandle};
+use message_bus::SendError;
+use message_bus::lifecycle::ShutdownToken;
+use metadata::impls::metadata::StreamsFrontend;
+use server_common::Message;
+use server_common::sharding::CONSUMER_SESSION_REPORT_TIMEOUT;
+use shard::Receiver;
+use tracing::{debug, info, trace, warn};
+
+use crate::shell::{ServerShard, ShellBus, ShellShard};
+
+const MAX_LOGOUTS_PER_PASS: usize = 256;
+const HEARTBEAT_RETRY_INTERVAL: Duration = Duration::from_millis(1);
+
+#[derive(Debug, PartialEq, Eq)]
+enum Pass {
+ Stopped,
+ HitCap,
+ Drained,
+}
+
+#[derive(Debug)]
+struct Lease {
+ session: Option<u64>,
+ last_seen: Instant,
+}
+
+/// Bounded by the current consumer-group members, independent of heartbeat volume.
+#[derive(Default)]
+pub struct ConsumerGroupLiveness {
+ view: Option<u32>,
+ leases: BTreeMap<u128, Lease>,
+ last_incomplete: Option<Instant>,
+ last_incomplete_warning: Option<Instant>,
+ report_offset: usize,
+ report_incomplete: bool,
+}
+
+impl ConsumerGroupLiveness {
+ fn observe_view(&mut self, view: Option<u32>) {
+ if self.view != view {
+ self.leases.clear();
+ self.last_incomplete = None;
+ self.last_incomplete_warning = None;
+ self.view = view;
+ }
+ }
+
+ fn reconcile(&mut self, view: u32, members: &BTreeMap<u128, Option<u64>>, now: Instant) {
+ self.observe_view(Some(view));
+ self.leases
+ .retain(|client_id, _| members.contains_key(client_id));
+ for (&client_id, &session) in members {
+ let lease = self.leases.entry(client_id).or_insert(Lease {
+ session,
+ last_seen: now,
+ });
+ if lease.session != session {
+ *lease = Lease {
+ session,
+ last_seen: now,
+ };
+ }
+ }
+ }
+
+ fn renew(&mut self, session: ConsumerSession, now: Instant) {
+ if let Some(lease) = self.leases.get_mut(&session.client_id)
+ && lease.session == Some(session.session)
+ {
+ lease.last_seen = now;
+ }
+ }
+
+ fn defer_expiry(&mut self, now: Instant, timeout: Duration, replica: u8) {
+ // Recovered memberships do not identify their hosting replica. Until
+ // every reporting node can enumerate its clients, absence is uncertain.
+ if self
+ .last_incomplete_warning
+ .is_none_or(|last| now.saturating_duration_since(last) >= timeout)
+ {
+ warn!(
+ replica,
+ ?timeout,
+ "incomplete consumer session report; deferring session expiry"
+ );
+ self.last_incomplete_warning = Some(now);
+ }
+ self.last_incomplete = Some(now);
+ }
+
+ pub(crate) fn receive(
+ &mut self,
+ cluster: u128,
+ primary_view: Option<u32>,
+ message: &Message<ConsumerSessionHeartbeatHeader>,
+ now: Instant,
+ timeout: Duration,
+ ) {
+ self.observe_view(primary_view);
+ let header = message.header();
+ if primary_view != Some(header.view) || header.cluster != cluster {
+ return;
+ }
+ // Validate the whole batch before refreshing anything.
+ let sessions = message
+ .body()
+ .as_chunks::<{ ConsumerSession::ENCODED_SIZE }>()
+ .0;
+ if let Err(error) = sessions
+ .iter()
+ .try_for_each(|bytes| ConsumerSession::decode(bytes).map(|_| ()))
+ {
+ debug!(
+ ?error,
+ replica = header.replica,
+ "invalid consumer session heartbeat body"
+ );
+ return;
+ }
+ if header.incomplete != 0 {
+ self.defer_expiry(now, timeout, header.replica);
+ }
+ for bytes in sessions {
+ if let Ok((session, _)) = ConsumerSession::decode(bytes) {
+ self.renew(session, now);
+ }
+ }
+ }
+
+ fn expired(
+ &self,
+ view: u32,
+ client_id: u128,
+ session: Option<u64>,
+ now: Instant,
+ timeout: Duration,
+ ) -> bool {
+ self.view == Some(view)
+ && self.last_incomplete.is_none_or(|last_incomplete| {
+ now.saturating_duration_since(last_incomplete) >= timeout
+ })
+ && self.leases.get(&client_id).is_some_and(|lease| {
+ lease.session == session
+ && now.saturating_duration_since(lease.last_seen) >= timeout
+ })
+ }
+}
+
+pub async fn run(
+ shard: Rc<ServerShard>,
+ liveness: Rc<RefCell<ConsumerGroupLiveness>>,
+ stop: Receiver<()>,
+ interval: Duration,
+ timeout: Duration,
+) {
+ let shutdown = shard.bus.token();
+ 'running: while compio::time::timeout(interval, stop.recv()).await.is_err() {
+ loop {
+ match report_and_expire(&shard, &liveness, &stop, &shutdown, timeout).await {
+ Pass::Stopped => break 'running,
+ Pass::HitCap => {}
+ Pass::Drained => break,
+ }
+ }
+ }
+}
+
+async fn report_and_expire<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ liveness: &RefCell<ConsumerGroupLiveness>,
+ stop: &Receiver<()>,
+ shutdown: &ShutdownToken,
+ timeout: Duration,
+) -> Pass
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let metadata = shard.plane.metadata();
+ let Some(consensus) = metadata.consensus.as_ref() else {
+ return Pass::Drained;
+ };
+ let clients = shard.gather_consumer_sessions().await;
+ let now = Instant::now();
+ let view = consensus.view();
+ let serving = consensus.is_normal() && !consensus.is_transferring();
+ let primary_view = (consensus.is_primary() && serving).then_some(view);
+ liveness.borrow_mut().observe_view(primary_view);
+ if !consensus.is_normal() || (consensus.is_primary() && !serving) {
+ return Pass::Drained;
+ }
+
+ let incomplete = !clients.complete;
+ let sessions = clients.clients;
+
+ if primary_view.is_some() {
+ let table = metadata.client_table.borrow();
+ let members = metadata.mux_stm.streams().read(|inner| {
+ inner
+ .items
+ .iter()
+ .flat_map(|(_, stream)| {
+ stream.topics.iter().flat_map(|(_, topic)| {
+ topic.consumer_groups.values().flat_map(|group| {
+ group.members.iter().map(|(_, member)| {
+ (
+ member.client_id,
+ table.get_epoch(member.client_id).or(member.session),
+ )
+ })
+ })
+ })
+ })
+ .fold(
+ BTreeMap::<u128, Option<u64>>::new(),
+ |mut members, (client_id, session)| {
+ members
+ .entry(client_id)
+ .and_modify(|current| *current = (*current).max(session))
+ .or_insert(session);
+ members
+ },
+ )
+ });
+ drop(table);
+ let mut tracker = liveness.borrow_mut();
+ tracker.reconcile(view, &members, now);
+ if incomplete {
+ tracker.defer_expiry(now, timeout, consensus.replica());
+ }
+ for session in sessions {
+ tracker.renew(session, now);
+ }
+ } else {
+ // A backup can lag a freshly joined group. Report all bound sessions;
+ // filtering through its replicated memberships would drop valid renewals.
+ // Even an empty gather must report incompleteness to the primary.
+ return report_sessions(shard, liveness, &sessions, incomplete, stop, shutdown).await;
+ }
+
+ expire_sessions(shard, liveness, stop, shutdown, timeout, now).await
+}
+
+async fn report_sessions<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ liveness: &RefCell<ConsumerGroupLiveness>,
+ sessions: &[ConsumerSession],
+ incomplete: bool,
+ stop: &Receiver<()>,
+ shutdown: &ShutdownToken,
+) -> Pass
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let Some(consensus) = shard.plane.metadata().consensus.as_ref() else {
+ return Pass::Drained;
+ };
+ let batch_size = (shard.bus_max_message_size().saturating_sub(HEADER_SIZE)
+ / ConsumerSession::ENCODED_SIZE)
+ .min(MAX_CONSUMER_SESSIONS_PER_HEARTBEAT);
+ if batch_size == 0 {
+ return Pass::Drained;
+ }
+ let view = consensus.view();
+ let session_count = sessions.len().max(1);
+ let (offset, incomplete) = {
+ let mut tracker = liveness.borrow_mut();
+ let incomplete = incomplete || tracker.report_incomplete;
+ tracker.report_incomplete = true;
+ (tracker.report_offset % session_count, incomplete)
+ };
+ let (before, after) = sessions.split_at(offset);
+ let mut batches = after.chunks(batch_size).chain(before.chunks(batch_size));
+ let first = batches.next().unwrap_or_default();
+ let send = async {
+ let mut offset = offset;
+ for batch in std::iter::once(first).chain(batches) {
+ loop {
+ if stop.try_recv().is_ok() || shutdown.is_triggered() {
+ return Pass::Stopped;
+ }
+ if consensus.view() != view || !consensus.is_normal() || consensus.is_primary() {
+ return Pass::Drained;
+ }
+ let heartbeat = heartbeat_message(
+ consensus.cluster(),
+ view,
+ consensus.replica(),
+ batch,
+ incomplete,
+ );
+ match shard
+ .bus
+ .send_to_replica(consensus.primary_index(view), heartbeat.into_frozen())
+ .await
+ {
+ Ok(()) => {
+ offset = (offset + batch.len()) % session_count;
+ liveness.borrow_mut().report_offset = offset;
+ break;
+ }
+ Err(SendError::Backpressure | SendError::ReplicaForwardFailed(_)) => {
+ shard.bus.sleep(HEARTBEAT_RETRY_INTERVAL).await;
+ }
+ Err(error) => {
+ trace!(
+ ?error,
+ "consumer session heartbeat could not reach metadata primary"
+ );
+ return Pass::Drained;
+ }
+ }
+ }
+ }
+ liveness.borrow_mut().report_incomplete = false;
+ Pass::Drained
+ };
+ shard::bus_timeout(&shard.bus, CONSUMER_SESSION_REPORT_TIMEOUT, send)
+ .await
+ .unwrap_or(Pass::Drained)
+}
+
+async fn expire_sessions<B, MJ, S, SB>(
+ shard: &Rc<ShellShard<B, MJ, S, SB>>,
+ liveness: &RefCell<ConsumerGroupLiveness>,
+ stop: &Receiver<()>,
+ shutdown: &ShutdownToken,
+ timeout: Duration,
+ now: Instant,
+) -> Pass
+where
+ B: ShellBus,
+ MJ: JournalHandle + 'static,
+ MJ::Target: Journal<Entry = Message<PrepareHeader>, Header = PrepareHeader>,
+ S: 'static,
+ SB: SuperblockStore + 'static,
+{
+ let metadata = shard.plane.metadata();
+ let Some(consensus) = metadata.consensus.as_ref() else {
+ return Pass::Drained;
+ };
+ let view = consensus.view();
+ let expired = {
+ let mut tracker = liveness.borrow_mut();
+ if tracker
+ .last_incomplete
+ .is_some_and(|last| now.saturating_duration_since(last) >= timeout)
+ {
+ tracker.last_incomplete = None;
+ tracker.last_incomplete_warning = None;
+ info!(?timeout, "consumer-group session expiry resumed");
+ }
+ tracker
+ .leases
+ .iter()
+ .filter(|(_, lease)| now.saturating_duration_since(lease.last_seen) >= timeout)
+ .take(MAX_LOGOUTS_PER_PASS)
+ .map(|(&client_id, lease)| (client_id, lease.session))
+ .collect::<Vec<_>>()
+ };
+ let hit_cap = expired.len() == MAX_LOGOUTS_PER_PASS;
+ let mut removed = 0;
+ let mut deferred = false;
+ for (client_id, session) in expired {
+ // Submission is not cancel-safe. Stop between logouts and recheck every
+ // fence after the preceding commit, which can yield to a new heartbeat.
+ if stop.try_recv().is_ok() || shutdown.is_triggered() {
+ return Pass::Stopped;
+ }
+ if consensus.view() != view || !metadata.is_caught_up_primary() {
+ return Pass::Drained;
+ }
+ if !liveness
+ .borrow()
+ .expired(view, client_id, session, now, timeout)
+ {
+ continue;
+ }
+ match metadata
+ .submit_expired_logout_in_process(client_id, session)
+ .await
+ {
+ Ok(Some(_)) => {
+ removed += 1;
+ info!(client_id, ?session, "expired consumer-group session");
+ }
+ Ok(None) => {}
+ Err(error) => {
+ deferred = true;
+ trace!(client_id, ?error, "consumer-group session cleanup deferred");
+ }
+ }
+ }
+ if hit_cap && removed > 0 && !deferred {
+ Pass::HitCap
+ } else {
+ Pass::Drained
+ }
+}
+
+#[allow(clippy::cast_possible_truncation)]
+fn heartbeat_message(
+ cluster: u128,
+ view: u32,
+ replica: u8,
+ sessions: &[ConsumerSession],
+ incomplete: bool,
+) -> Message<ConsumerSessionHeartbeatHeader> {
+ let mut body = BytesMut::with_capacity(sessions.len() * ConsumerSession::ENCODED_SIZE);
+ for session in sessions {
+ session.encode(&mut body);
+ }
+ let size = HEADER_SIZE + body.len();
+ let mut message = Message::<ConsumerSessionHeartbeatHeader>::new(size).transmute_header(
+ |_, header: &mut ConsumerSessionHeartbeatHeader| {
+ header.command = Command::ConsumerSessionHeartbeat;
+ header.cluster = cluster;
+ header.view = view;
+ header.replica = replica;
+ header.incomplete = u8::from(incomplete);
+ header.size = size as u32;
+ header.checksum_body = u128::from(iggy_common::calculate_checksum(&body));
+ header.seal();
+ },
+ );
+ message.as_mut_slice()[HEADER_SIZE..size].copy_from_slice(&body);
+ message
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::dispatch::test_support::{SpyBus, TestShard, prepare_message, test_shard};
+ use consensus::client_table::ClientTable;
+ use iggy_binary_protocol::primitives::partition_assignment::CreatedPartitionAssignment;
+ use iggy_binary_protocol::requests::consumer_groups::CreateConsumerGroupRequest;
+ use iggy_binary_protocol::requests::streams::CreateStreamRequest;
+ use iggy_binary_protocol::requests::topics::{
+ CreateTopicRequest, CreateTopicWithAssignmentsRequest,
+ };
+ use iggy_binary_protocol::{Operation, WireIdentifier, WireName, WireOptions};
+ use journal::prepare_journal::PrepareJournal;
+ use message_bus::lifecycle::Shutdown;
+ use metadata::stm::StateMachine;
+ use metadata::stm::consumer_group::JoinConsumerGroupRequest;
+ use server_common::MessageBag;
+ use server_common::sharding::METADATA_GROUP;
+
+ const CLIENT: u128 = 7;
+ const SESSION: u64 = 11;
+ const VIEW: u32 = 3;
+ const CLUSTER: u128 = 42;
+ const TIMEOUT: Duration = Duration::from_secs(30);
+
+ #[compio::test]
+ async fn heartbeat_batches_fit_the_transport_limit_without_losing_sessions() {
+ const SMALL_FRAME_CAP: usize = 4096;
+ let sessions = (0..=MAX_CONSUMER_SESSIONS_PER_HEARTBEAT)
+ .map(|index| ConsumerSession {
+ client_id: CLIENT + index as u128,
+ session: SESSION,
+ })
+ .collect::<Vec<_>>();
+ for cap in [
+ HEADER_SIZE + ConsumerSession::ENCODED_SIZE,
+ SMALL_FRAME_CAP,
+ HEADER_SIZE + MAX_CONSUMER_SESSIONS_PER_HEARTBEAT * ConsumerSession::ENCODED_SIZE,
+ ] {
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, 1, 3, 1));
+ shard.set_bus_max_message_size(cap);
+ let tracker = RefCell::new(ConsumerGroupLiveness::default());
+ let (_stop, receiver) = shard::channel(1);
+ let (_signal, shutdown) = Shutdown::new();
+ assert_eq!(
+ report_sessions(&shard, &tracker, &sessions, false, &receiver, &shutdown).await,
+ Pass::Drained
+ );
+ let frames = bus.replica_sends.borrow();
+ assert!(frames.iter().all(|(_, frame)| frame.len() <= cap));
+ let received = frames
+ .iter()
+ .flat_map(|(_, frame)| {
+ frame[HEADER_SIZE..]
+ .as_chunks::<{ ConsumerSession::ENCODED_SIZE }>()
+ .0
+ .iter()
+ })
+ .map(|bytes| ConsumerSession::decode(bytes).unwrap().0)
+ .collect::<Vec<_>>();
+ assert_eq!(received, sessions);
+ assert!(!tracker.borrow().report_incomplete);
+ }
+ }
+
+ #[compio::test]
+ async fn heartbeat_reports_retry_a_full_peer_queue_until_the_tail_is_sent() {
+ const PEER_QUEUE_CAPACITY: usize = 2;
+ let sessions = (0..=MAX_CONSUMER_SESSIONS_PER_HEARTBEAT * PEER_QUEUE_CAPACITY)
+ .map(|index| ConsumerSession {
+ client_id: CLIENT + index as u128,
+ session: SESSION,
+ })
+ .collect::<Vec<_>>();
+ let bus = SpyBus::default();
+ bus.replica_send_capacity.set(Some(PEER_QUEUE_CAPACITY));
+ let shard = Rc::new(test_shard(&bus, 1, 3, 1));
+ let tracker = RefCell::new(ConsumerGroupLiveness::default());
+ let (_stop, receiver) = shard::channel(1);
+ let (_signal, shutdown) = Shutdown::new();
+ let send = report_sessions(&shard, &tracker, &sessions, false, &receiver, &shutdown);
+ let drain = async {
+ let mut received = Vec::new();
+ while received.len() < sessions.len() {
+ for (_, frame) in bus.replica_sends.borrow_mut().drain(..) {
+ received.extend(
+ frame[HEADER_SIZE..]
+ .as_chunks::<{ ConsumerSession::ENCODED_SIZE }>()
+ .0
+ .iter()
+ .map(|bytes| ConsumerSession::decode(bytes).unwrap().0),
+ );
+ }
+ if received.len() < sessions.len() {
+ compio::time::sleep(HEARTBEAT_RETRY_INTERVAL).await;
+ }
+ }
+ received
+ };
+ let (pass, received) =
+ compio::time::timeout(TIMEOUT, async { futures::join!(send, drain) })
+ .await
+ .expect("heartbeat sender must yield to the peer writer");
+ assert_eq!(pass, Pass::Drained);
+ assert_eq!(received, sessions);
+ assert!(!tracker.borrow().report_incomplete);
+ }
+
+ #[compio::test]
+ async fn timed_out_reports_resume_at_the_unsent_session_and_defer_expiry() {
+ let sessions = (0..=MAX_CONSUMER_SESSIONS_PER_HEARTBEAT)
+ .map(|index| ConsumerSession {
+ client_id: CLIENT + index as u128,
+ session: SESSION,
+ })
+ .collect::<Vec<_>>();
+ let bus = SpyBus::default();
+ bus.replica_send_capacity.set(Some(1));
+ let shard = Rc::new(test_shard(&bus, 1, 3, 1));
+ let tracker = RefCell::new(ConsumerGroupLiveness::default());
+ let (stop, receiver) = shard::channel(1);
+ let (_signal, shutdown) = Shutdown::new();
+ assert_eq!(
+ report_sessions(&shard, &tracker, &sessions, false, &receiver, &shutdown).await,
+ Pass::Drained
+ );
+ assert!(tracker.borrow().report_incomplete);
+ assert_eq!(
+ tracker.borrow().report_offset,
+ MAX_CONSUMER_SESSIONS_PER_HEARTBEAT
+ );
+ bus.replica_sends.borrow_mut().clear();
+ bus.replica_send_capacity.set(None);
+ assert_eq!(
+ report_sessions(&shard, &tracker, &sessions, false, &receiver, &shutdown).await,
+ Pass::Drained
+ );
+ {
+ let frames = bus.replica_sends.borrow();
+ assert_eq!(
+ ConsumerSession::decode(&frames[0].1[HEADER_SIZE..])
+ .unwrap()
+ .0,
+ sessions[MAX_CONSUMER_SESSIONS_PER_HEARTBEAT]
+ );
+ assert!(frames.iter().all(|(_, frame)| {
+ frame[std::mem::offset_of!(ConsumerSessionHeartbeatHeader, incomplete)] == 1
+ }));
+ }
+ assert!(!tracker.borrow().report_incomplete);
+ bus.replica_sends.borrow_mut().clear();
+ stop.try_send(()).unwrap();
+ assert_eq!(
+ report_sessions(&shard, &tracker, &sessions, false, &receiver, &shutdown).await,
+ Pass::Stopped
+ );
+ assert!(bus.replica_sends.borrow().is_empty());
+ }
+
+ #[compio::test]
+ async fn mixed_legacy_membership_epochs_do_not_block_automatic_expiry() {
+ let (_dir, shard) = shard_with_members(1).await;
+ let metadata = shard.plane.metadata();
+ let topic = CreateTopicWithAssignmentsRequest {
+ request: CreateTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ name: WireName::new("legacy").unwrap(),
+ options: WireOptions::empty(),
+ },
+ created_view: 0,
+ derived_options: WireOptions::empty(),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 2,
+ }],
+ };
+ let group = CreateConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(1),
+ name: WireName::new("legacy").unwrap(),
+ };
+ let legacy_join = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(1),
+ group_id: WireIdentifier::numeric(0),
+ client_id: CLIENT,
+ in_flight: Vec::new(),
+ session: None,
+ };
+ for (operation, body) in [
+ (Operation::CreateTopicWithAssignments, topic.to_bytes()),
+ (Operation::CreateConsumerGroup, group.to_bytes()),
+ (Operation::JoinConsumerGroup, legacy_join.to_bytes()),
+ ] {
+ assert_eq!(
+ metadata
+ .mux_stm
+ .update(prepare_message(operation, CLIENT, 1, &body))
+ .unwrap()
+ .code,
+ 0
+ );
+ }
+ assert_eq!(metadata.client_table.borrow().get_epoch(CLIENT), None);
+ let tracker = RefCell::new(ConsumerGroupLiveness::default());
+ let (_stop, receiver) = shard::channel(1);
+ let (_signal, shutdown) = Shutdown::new();
+ report_and_expire(&shard, &tracker, &receiver, &shutdown, TIMEOUT).await;
+ assert_eq!(tracker.borrow().leases[&CLIENT].session, Some(SESSION));
+ let expired_at = Instant::now() + TIMEOUT * 2;
+ assert_eq!(
+ expire_sessions(&shard, &tracker, &receiver, &shutdown, TIMEOUT, expired_at).await,
+ Pass::Drained
+ );
+ assert!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_memberships(CLIENT)
+ .is_empty()
+ );
+ }
+
+ #[compio::test]
+ async fn transferring_backup_still_reports_while_transferring_primary_does_not() {
+ for replica in [0, 1] {
+ let bus = SpyBus::default();
+ let shard = Rc::new(test_shard(&bus, replica, 3, 1));
+ let consensus = shard.plane.metadata().consensus.as_ref().unwrap();
+ consensus.begin_state_transfer_await();
+ assert!(consensus.is_normal());
+ assert!(consensus.is_transferring());
+ let (_stop, receiver) = shard::channel(1);
+ let (_signal, shutdown) = Shutdown::new();
+ let tracker = RefCell::new(ConsumerGroupLiveness::default());
+ assert_eq!(
+ report_and_expire(&shard, &tracker, &receiver, &shutdown, TIMEOUT).await,
+ Pass::Drained
+ );
+ if replica == 0 {
+ assert!(bus.replica_sends.borrow().is_empty());
+ } else {
+ let (target, header) = bus.sole_replica_send::<ConsumerSessionHeartbeatHeader>();
+ assert_eq!(target, 0);
+ assert_eq!(header.replica, replica);
+ assert_eq!(header.view, consensus.view());
+ assert_eq!(header.incomplete, 1);
+ }
+ }
+ }
+
+ #[test]
+ fn one_lost_report_with_gather_delays_does_not_expire_live_members() {
+ const INTERVAL: Duration = Duration::from_secs(20);
+ const VALID_TIMEOUT: Duration = Duration::from_secs(67);
+ let now = Instant::now();
+ let mut tracker = ConsumerGroupLiveness::default();
+ tracker.reconcile(VIEW, &BTreeMap::from([(CLIENT, Some(SESSION))]), now);
+ let next_report = now
+ + (INTERVAL
+ + server_common::sharding::LIST_CLIENTS_GATHER_TIMEOUT
+ + CONSUMER_SESSION_REPORT_TIMEOUT)
+ * 2;
+ assert!(!tracker.expired(VIEW, CLIENT, Some(SESSION), next_report, VALID_TIMEOUT));
+ tracker.receive(
+ CLUSTER,
+ Some(VIEW),
+ &heartbeat_message(
+ CLUSTER,
+ VIEW,
+ 1,
+ &[ConsumerSession {
+ client_id: CLIENT,
+ session: SESSION,
+ }],
+ false,
+ ),
+ next_report,
+ VALID_TIMEOUT,
+ );
+ assert!(!tracker.expired(
+ VIEW,
+ CLIENT,
+ Some(SESSION),
+ now + VALID_TIMEOUT,
+ VALID_TIMEOUT
+ ));
+ assert!(tracker.expired(
+ VIEW,
+ CLIENT,
+ Some(SESSION),
+ next_report + VALID_TIMEOUT,
+ VALID_TIMEOUT
+ ));
+ }
+
+ #[compio::test]
+ async fn capped_expiry_with_a_fenced_candidate_continues_but_deferral_stops() {
+ let (_dir, shard) = shard_with_members(MAX_LOGOUTS_PER_PASS + 1).await;
+ let metadata = shard.plane.metadata();
+ let now = Instant::now();
+ let members = (0..=MAX_LOGOUTS_PER_PASS)
+ .map(|index| (CLIENT + index as u128, Some(SESSION)))
+ .collect();
+ let tracker = RefCell::new(ConsumerGroupLiveness::default());
+ tracker.borrow_mut().reconcile(0, &members, now);
+ metadata
+ .mux_stm
+ .streams()
+ .refresh_consumer_group_session(CLIENT, SESSION + 1);
+ let (_stop, receiver) = shard::channel(1);
+ let (_signal, shutdown) = Shutdown::new();
+
+ tracker.borrow_mut().defer_expiry(now + TIMEOUT, TIMEOUT, 1);
+ assert_eq!(
+ expire_sessions(
+ &shard,
+ &tracker,
+ &receiver,
+ &shutdown,
+ TIMEOUT,
+ now + TIMEOUT
+ )
+ .await,
+ Pass::Drained
+ );
+ assert_eq!(metadata.consensus.as_ref().unwrap().commit_min(), 0);
+
+ assert_eq!(
+ expire_sessions(
+ &shard,
+ &tracker,
+ &receiver,
+ &shutdown,
+ TIMEOUT,
+ now + TIMEOUT * 2
+ )
+ .await,
+ Pass::HitCap
+ );
+ assert_eq!(
+ metadata.consensus.as_ref().unwrap().commit_min(),
+ (MAX_LOGOUTS_PER_PASS - 1) as u64
+ );
+ assert_eq!(
+ metadata.mux_stm.streams().consumer_group_session(CLIENT),
+ Some(SESSION + 1)
+ );
+ assert_eq!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_session(CLIENT + MAX_LOGOUTS_PER_PASS as u128),
+ Some(SESSION)
+ );
+
+ let remaining = BTreeMap::from([
+ (CLIENT, Some(SESSION + 1)),
+ (CLIENT + MAX_LOGOUTS_PER_PASS as u128, Some(SESSION)),
+ ]);
+ tracker
+ .borrow_mut()
+ .reconcile(0, &remaining, now + TIMEOUT * 2);
+ assert_eq!(
+ expire_sessions(
+ &shard,
+ &tracker,
+ &receiver,
+ &shutdown,
+ TIMEOUT,
+ now + TIMEOUT * 2
+ )
+ .await,
+ Pass::Drained
+ );
+ assert_eq!(
+ metadata
+ .mux_stm
+ .streams()
+ .consumer_group_session(CLIENT + MAX_LOGOUTS_PER_PASS as u128),
+ None
+ );
+ assert_eq!(
+ metadata.mux_stm.streams().consumer_group_session(CLIENT),
+ Some(SESSION + 1)
+ );
+ }
+
+ async fn shard_with_members(count: usize) -> (tempfile::TempDir, Rc<TestShard>) {
+ let dir = tempfile::tempdir().unwrap();
+ let mut shard = test_shard(&SpyBus::default(), 0, 1, 1);
+ let metadata = shard.plane.metadata_mut();
+ metadata.journal = Some(
+ PrepareJournal::open(&dir.path().join("journal.wal"), 0)
+ .await
+ .unwrap(),
+ );
+ metadata.client_table = RefCell::new(ClientTable::new(1));
+ let create_stream = CreateStreamRequest {
+ name: WireName::new("stream").unwrap(),
+ options: WireOptions::empty(),
+ };
+ let create_topic = CreateTopicWithAssignmentsRequest {
+ request: CreateTopicRequest {
+ stream_id: WireIdentifier::numeric(0),
+ partitions_count: 1,
+ name: WireName::new("topic").unwrap(),
+ options: WireOptions::empty(),
+ },
+ created_view: 0,
+ derived_options: WireOptions::empty(),
+ partitions: vec![CreatedPartitionAssignment {
+ partition_id: 0,
+ consensus_group_id: 1,
+ }],
+ };
+ let create_group = CreateConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ name: WireName::new("group").unwrap(),
+ };
+ for (operation, body) in [
+ (Operation::CreateStream, create_stream.to_bytes()),
+ (
+ Operation::CreateTopicWithAssignments,
+ create_topic.to_bytes(),
+ ),
+ (Operation::CreateConsumerGroup, create_group.to_bytes()),
+ ] {
+ assert_eq!(
+ metadata
+ .mux_stm
+ .update(prepare_message(operation, CLIENT, 1, &body))
+ .unwrap()
+ .code,
+ 0
+ );
+ }
+ for index in 0..count {
+ let client_id = CLIENT + index as u128;
+ let join = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::numeric(0),
+ topic_id: WireIdentifier::numeric(0),
+ group_id: WireIdentifier::numeric(0),
+ client_id,
+ in_flight: Vec::new(),
+ session: None,
+ };
+ assert_eq!(
+ metadata
+ .mux_stm
+ .update(prepare_message(
+ Operation::JoinConsumerGroup,
+ client_id,
+ 1,
+ &join.to_bytes()
+ ))
+ .unwrap()
+ .code,
+ 0
+ );
+ metadata
+ .mux_stm
+ .streams()
+ .refresh_consumer_group_session(client_id, SESSION);
+ }
+ (dir, Rc::new(shard))
+ }
+
+ #[test]
+ fn recovered_members_receive_full_timeout_including_missing_client_table_entries() {
+ let now = Instant::now();
+ let mut tracker = ConsumerGroupLiveness::default();
+ tracker.reconcile(
+ VIEW,
+ &BTreeMap::from([(CLIENT, Some(SESSION)), (CLIENT + 1, None)]),
+ now,
+ );
+ assert!(!tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT / 2, TIMEOUT));
+ assert!(tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT, TIMEOUT));
+ assert!(tracker.expired(VIEW, CLIENT + 1, None, now + TIMEOUT, TIMEOUT));
+ tracker.reconcile(VIEW, &BTreeMap::new(), now + TIMEOUT);
+ assert!(tracker.leases.is_empty());
+ }
+
+ #[test]
+ fn remote_heartbeats_renew_only_matching_sessions() {
+ let now = Instant::now();
+ let mut tracker = ConsumerGroupLiveness::default();
+ tracker.reconcile(
+ VIEW,
+ &BTreeMap::from([(CLIENT, Some(SESSION)), (CLIENT + 1, Some(SESSION))]),
+ now,
+ );
+ let message = heartbeat_message(
+ CLUSTER,
+ VIEW,
+ 2,
+ &[ConsumerSession {
+ client_id: CLIENT,
+ session: SESSION,
+ }],
+ false,
+ );
+ tracker.receive(CLUSTER, Some(VIEW), &message, now + TIMEOUT / 2, TIMEOUT);
+ assert!(!tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT, TIMEOUT));
+ assert!(tracker.expired(VIEW, CLIENT + 1, Some(SESSION), now + TIMEOUT, TIMEOUT));
+ assert!(tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT * 2, TIMEOUT));
+
+ tracker.reconcile(
+ VIEW,
+ &BTreeMap::from([(CLIENT, Some(SESSION + 1))]),
+ now + TIMEOUT,
+ );
+ tracker.receive(CLUSTER, Some(VIEW), &message, now + TIMEOUT * 2, TIMEOUT);
+ assert!(tracker.expired(VIEW, CLIENT, Some(SESSION + 1), now + TIMEOUT * 2, TIMEOUT));
+ }
+
+ #[test]
+ fn promotion_grants_full_grace_and_rejects_old_view_or_foreign_heartbeats() {
+ let now = Instant::now();
+ let members = BTreeMap::from([(CLIENT, Some(SESSION))]);
+ let mut tracker = ConsumerGroupLiveness::default();
+ tracker.reconcile(VIEW, &members, now);
+ tracker.observe_view(None);
+ tracker.reconcile(VIEW + 1, &members, now + TIMEOUT);
+ assert!(!tracker.expired(VIEW + 1, CLIENT, Some(SESSION), now + TIMEOUT, TIMEOUT));
+ for (cluster, view) in [(CLUSTER + 1, VIEW + 1), (CLUSTER, VIEW)] {
+ let message = heartbeat_message(
+ cluster,
+ view,
+ 2,
+ &[ConsumerSession {
+ client_id: CLIENT,
+ session: SESSION,
+ }],
+ false,
+ );
+ tracker.receive(
+ CLUSTER,
+ Some(VIEW + 1),
+ &message,
+ now + TIMEOUT * 2,
+ TIMEOUT,
+ );
+ }
+ assert!(tracker.expired(VIEW + 1, CLIENT, Some(SESSION), now + TIMEOUT * 2, TIMEOUT));
+ assert!(!tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT * 2, TIMEOUT));
+ }
+
+ #[test]
+ fn unknown_sessions_do_not_grow_the_tracker_and_malformed_batches_do_not_renew() {
+ let now = Instant::now();
+ let mut tracker = ConsumerGroupLiveness::default();
+ tracker.reconcile(VIEW, &BTreeMap::from([(CLIENT, Some(SESSION))]), now);
+ for sessions in [
+ vec![ConsumerSession {
+ client_id: CLIENT + 1,
+ session: SESSION,
+ }],
+ vec![
+ ConsumerSession {
+ client_id: CLIENT,
+ session: SESSION,
+ },
+ ConsumerSession {
+ client_id: 0,
+ session: 0,
+ },
+ ],
+ ] {
+ let message = heartbeat_message(CLUSTER, VIEW, 2, &sessions, false);
+ tracker.receive(CLUSTER, Some(VIEW), &message, now + TIMEOUT, TIMEOUT);
+ }
+ assert_eq!(tracker.leases.len(), 1);
+ assert!(tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT, TIMEOUT));
+ }
+
+ #[test]
+ fn incomplete_local_gathers_grant_another_timeout_before_expiry() {
+ let now = Instant::now();
+ let mut tracker = ConsumerGroupLiveness::default();
+ let members = BTreeMap::from([(CLIENT, Some(SESSION))]);
+ tracker.reconcile(VIEW, &members, now);
+ tracker.defer_expiry(now + TIMEOUT, TIMEOUT, 0);
+ tracker.reconcile(VIEW, &members, now + TIMEOUT + TIMEOUT / 2);
+ assert!(!tracker.expired(
+ VIEW,
+ CLIENT,
+ Some(SESSION),
+ now + TIMEOUT + TIMEOUT / 2,
+ TIMEOUT
+ ));
+ assert!(tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT * 2, TIMEOUT));
+ }
+
+ #[test]
+ fn incomplete_remote_gathers_defer_expiry_even_without_sessions() {
+ let now = Instant::now();
+ let mut tracker = ConsumerGroupLiveness::default();
+ tracker.reconcile(VIEW, &BTreeMap::from([(CLIENT, Some(SESSION))]), now);
+ let incomplete = heartbeat_message(CLUSTER, VIEW, 2, &[], true);
+ let routed = MessageBag::try_from(incomplete.into_generic())
+ .unwrap()
+ .into_generic()
+ .try_into_typed::<ConsumerSessionHeartbeatHeader>()
+ .unwrap();
+ for tick in 1..=3 {
+ let received = now + TIMEOUT * tick;
+ tracker.receive(CLUSTER, Some(VIEW), &routed, received, TIMEOUT);
+ assert!(!tracker.expired(VIEW, CLIENT, Some(SESSION), received, TIMEOUT));
+ }
+ let complete = heartbeat_message(CLUSTER, VIEW, 2, &[], false);
+ tracker.receive(
+ CLUSTER,
+ Some(VIEW),
+ &complete,
+ now + TIMEOUT * 3 + TIMEOUT / 2,
+ TIMEOUT,
+ );
+ assert!(!tracker.expired(
+ VIEW,
+ CLIENT,
+ Some(SESSION),
+ now + TIMEOUT * 3 + TIMEOUT / 2,
+ TIMEOUT
+ ));
+ assert!(
+ tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT * 4, TIMEOUT),
+ "a crashed or recovered node must stop deferring expiry"
+ );
+ }
+
+ #[test]
+ fn foreign_or_malformed_incomplete_reports_cannot_defer_expiry() {
+ let now = Instant::now();
+ let mut tracker = ConsumerGroupLiveness::default();
+ tracker.reconcile(VIEW, &BTreeMap::from([(CLIENT, Some(SESSION))]), now);
+ for message in [
+ heartbeat_message(CLUSTER + 1, VIEW, 2, &[], true),
+ heartbeat_message(CLUSTER, VIEW + 1, 2, &[], true),
+ heartbeat_message(
+ CLUSTER,
+ VIEW,
+ 2,
+ &[ConsumerSession {
+ client_id: 0,
+ session: SESSION,
+ }],
+ true,
+ ),
+ ] {
+ tracker.receive(CLUSTER, Some(VIEW), &message, now + TIMEOUT, TIMEOUT);
+ }
+ assert!(tracker.expired(VIEW, CLIENT, Some(SESSION), now + TIMEOUT, TIMEOUT));
+ }
+
+ #[test]
+ fn bounded_heartbeat_round_trips_through_replica_routing() {
+ let sessions = vec![
+ ConsumerSession {
+ client_id: CLIENT,
+ session: SESSION
+ };
+ MAX_CONSUMER_SESSIONS_PER_HEARTBEAT
+ ];
+ let message = heartbeat_message(CLUSTER, VIEW, 2, &sessions, false);
+ assert_eq!(
+ u128::from(iggy_common::calculate_checksum(message.body())),
+ message.header().checksum_body
+ );
+ let bag = MessageBag::try_from(message.into_generic()).unwrap();
+ assert_eq!(bag.command(), Command::ConsumerSessionHeartbeat);
+ assert_eq!(
+ bag.routing(),
+ (iggy_binary_protocol::Operation::Reserved, METADATA_GROUP)
+ );
+ let typed = bag
+ .into_generic()
+ .try_into_typed::<ConsumerSessionHeartbeatHeader>()
+ .unwrap();
+ assert_eq!(
+ typed.body().len(),
+ sessions.len() * ConsumerSession::ENCODED_SIZE
+ );
+ let mut corrupted = typed.into_generic();
+ corrupted.as_mut_slice()[std::mem::offset_of!(ConsumerSessionHeartbeatHeader, view)] ^= 1;
+ assert!(MessageBag::try_from(corrupted).is_err());
+ }
+}
diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs
index 1d44012c8..87ecc81a1 100644
--- a/core/server/src/dispatch/mod.rs
+++ b/core/server/src/dispatch/mod.rs
@@ -72,7 +72,7 @@ use journal::{Journal, JournalHandle};
use message_bus::client_listener::RequestHandler;
use message_bus::replica::listener::MessageHandler;
use server_common::Message;
-use shard::{ConnectedClientInfo, ListClientsHandler};
+use shard::{ListClientsHandler, ListClientsReply};
use std::cell::RefCell;
use std::collections::VecDeque;
use std::rc::Rc;
@@ -104,11 +104,13 @@ type ActiveClientRequests = Rc<RefCell<AHashSet<u128>>>;
/// [`shard::IggyShard::list_all_clients`].
pub fn make_list_clients_handler(sessions: &Rc<RefCell<SessionManager>>) -> ListClientsHandler {
let sessions = Rc::clone(sessions);
- Rc::new(move |reply| {
- let clients: Vec<ConnectedClientInfo> = sessions.borrow().iter_clients().collect();
- // Best-effort: the gather side bounds itself by count + timeout, so
- // a dropped reply (receiver gone) just means this shard is omitted.
- let _ = reply.try_send(clients);
+ Rc::new(move |reply| match reply {
+ ListClientsReply::Clients(reply) => {
+ let _ = reply.try_send(sessions.borrow().iter_clients().collect());
+ }
+ ListClientsReply::Sessions(reply) => {
+ let _ = reply.try_send(sessions.borrow().iter_consumer_sessions().collect());
+ }
})
}
diff --git a/core/server/src/dispatch/session_ops.rs b/core/server/src/dispatch/session_ops.rs
index b756fef09..b68fee2ba 100644
--- a/core/server/src/dispatch/session_ops.rs
+++ b/core/server/src/dispatch/session_ops.rs
@@ -1059,7 +1059,7 @@ pub(in crate::dispatch) fn submit_disconnect_logout<B, MJ, S, SB>(
warn!(
vsr_client_id,
?error,
- "disconnect logout submit failed; peer slots may linger until eviction"
+ "disconnect logout submit failed; consumer-group cleanup will retry after session expiry"
);
}
});
diff --git a/core/server/src/dispatch/submit.rs b/core/server/src/dispatch/submit.rs
index 983a67844..3224c65cf 100644
--- a/core/server/src/dispatch/submit.rs
+++ b/core/server/src/dispatch/submit.rs
@@ -24,6 +24,7 @@
//! their replica forwards) delegate to `session_ops`, which owns that
//! machinery.
+use crate::consumer_group::liveness::ConsumerGroupLiveness;
use crate::dispatch::session_ops::{
answer_forwarded_logout, answer_forwarded_register, submit_logout_local_or_forward,
submit_register_local_or_forward,
@@ -31,13 +32,15 @@ use crate::dispatch::session_ops::{
use crate::dispatch::upgrade_shard_handle;
use crate::reply_frame::committed_reply_header;
use crate::shell::{ShellBus, ShellShard, ShellShardHandle};
-use consensus::MetadataHandle;
+use consensus::{Consensus, MetadataHandle};
use iggy_binary_protocol::{GenericHeader, PrepareHeader, RoutedRequestHeader};
use iggy_common::IggyError;
use journal::superblock::SuperblockStore;
use journal::{Journal, JournalHandle};
use server_common::Message;
+use std::cell::RefCell;
use std::rc::Rc;
+use std::time::{Duration, Instant};
use tracing::warn;
/// Handler shard 0 runs for an inbound [`shard::MetadataSubmit`]: a peer
@@ -49,6 +52,8 @@ use tracing::warn;
#[allow(clippy::too_many_lines)]
pub fn make_metadata_submit_handler<B, MJ, S, SB>(
shard_handle: &ShellShardHandle<B, MJ, S, SB>,
+ liveness: &Rc<RefCell<ConsumerGroupLiveness>>,
+ session_timeout: Duration,
) -> shard::MetadataSubmitHandler
where
B: ShellBus,
@@ -58,13 +63,30 @@ where
SB: SuperblockStore + 'static,
{
let shard_handle = Rc::clone(shard_handle);
+ let liveness = Rc::clone(liveness);
Rc::new(move |submit| {
let Some(shard) = upgrade_shard_handle(&shard_handle) else {
return;
};
let bus = shard.bus.clone();
+ let liveness = Rc::clone(&liveness);
bus.spawn(async move {
match submit {
+ shard::MetadataSubmit::ConsumerSessionHeartbeat(message) => {
+ let metadata = shard.plane.metadata();
+ if let Some(consensus) = metadata.consensus.as_ref() {
+ liveness.borrow_mut().receive(
+ consensus.cluster(),
+ (consensus.is_primary()
+ && consensus.is_normal()
+ && !consensus.is_transferring())
+ .then_some(consensus.view()),
+ &message,
+ Instant::now(),
+ session_timeout,
+ );
+ }
+ }
shard::MetadataSubmit::AttachConsumerSession {
vsr_client_id,
session,
diff --git a/core/server/src/dispatch/test_support.rs b/core/server/src/dispatch/test_support.rs
index c5bb3867f..8d4adc2b2 100644
--- a/core/server/src/dispatch/test_support.rs
+++ b/core/server/src/dispatch/test_support.rs
@@ -61,6 +61,7 @@ pub type RecordedReplicaSends = Rc<RefCell<Vec<(u8, Vec<u8>)>>>;
pub struct SpyBus {
pub client_replies: RecordedReplies,
pub replica_sends: RecordedReplicaSends,
+ pub replica_send_capacity: Rc<Cell<Option<usize>>>,
/// Installs of the client connection-lost hook, one per handler built
/// on this bus.
pub connection_lost_hooks: Rc<Cell<usize>>,
@@ -103,6 +104,13 @@ impl MessageBus for SpyBus {
replica: u8,
data: Frozen<MESSAGE_ALIGN>,
) -> Result<(), SendError> {
+ if self
+ .replica_send_capacity
+ .get()
+ .is_some_and(|capacity| self.replica_sends.borrow().len() >= capacity)
+ {
+ return Err(SendError::Backpressure);
+ }
self.replica_sends
.borrow_mut()
.push((replica, data.as_slice().to_vec()));
diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs
index 85894ccb9..5fa44b2b0 100644
--- a/core/server/src/partition_reconciler.rs
+++ b/core/server/src/partition_reconciler.rs
@@ -1781,6 +1781,7 @@ mod tests {
group_id: WireIdentifier::numeric(group_id),
client_id,
in_flight: Vec::new(),
+ session: None,
};
mux.update(build_prepare(op, Operation::JoinConsumerGroup, &req))
.expect("JoinConsumerGroup apply succeeds");
diff --git a/core/server/src/session_manager.rs b/core/server/src/session_manager.rs
index 685864aa5..bf3cc52f8 100644
--- a/core/server/src/session_manager.rs
+++ b/core/server/src/session_manager.rs
@@ -28,6 +28,7 @@
use crate::cluster_meta::ClusterRoster;
use ahash::AHashMap;
use consensus::client_table::SessionAttachment;
+use iggy_binary_protocol::ConsumerSession;
use iggy_common::IggyError;
use message_bus::installer::conn_info::ClientTransportKind;
use shard::ConnectedClientInfo;
@@ -497,6 +498,19 @@ impl SessionManager {
.iter()
.map(|(&id, conn)| record_from(id, conn))
}
+
+ pub fn iter_consumer_sessions(&self) -> impl Iterator<Item = ConsumerSession> + '_ {
+ self.connections.values().filter_map(|connection| {
+ if let ConnectionState::Bound {
+ client_id, session, ..
+ } = connection.state
+ {
+ Some(ConsumerSession { client_id, session })
+ } else {
+ None
+ }
+ })
+ }
}
impl Default for SessionManager {
@@ -535,15 +549,12 @@ impl std::error::Error for SessionError {}
/// Flatten a connection + its id into a [`ConnectedClientInfo`].
fn record_from(connection_id: u128, conn: &Connection) -> ConnectedClientInfo {
- let user_id = match conn.state {
- ConnectionState::Authenticated { user_id } | ConnectionState::Bound { user_id, .. } => {
- Some(user_id)
- }
- ConnectionState::Connected => None,
- };
- let vsr_client_id = match conn.state {
- ConnectionState::Bound { client_id, .. } => Some(client_id),
- ConnectionState::Authenticated { .. } | ConnectionState::Connected => None,
+ let (user_id, vsr_client_id) = match conn.state {
+ ConnectionState::Bound {
+ user_id, client_id, ..
+ } => (Some(user_id), Some(client_id)),
+ ConnectionState::Authenticated { user_id } => (Some(user_id), None),
+ ConnectionState::Connected => (None, None),
};
ConnectedClientInfo {
client_id: connection_id,
@@ -668,6 +679,47 @@ mod tests {
assert_eq!(mgr.iter_clients().count(), 1);
}
+ #[test]
+ fn consumer_session_gather_includes_only_live_bound_connections() {
+ const CLIENT: u128 = 7;
+ const USER: u32 = 1;
+ let mut sessions = SessionManager::new();
+ for connection in [1, 2] {
+ sessions.ensure_connection(connection, addr(5000), ClientTransportKind::Tcp);
+ sessions.record_sdk_info(
+ connection,
+ ClientSdkInfo {
+ sdk_name: "rust-sdk".to_owned(),
+ sdk_version: "1.0.0".to_owned(),
+ protocol_version: 1,
+ },
+ );
+ }
+ assert_eq!(sessions.iter_consumer_sessions().count(), 0);
+ sessions.login(1, USER).unwrap();
+ assert_eq!(sessions.iter_consumer_sessions().count(), 0);
+ sessions.bind_session(1, CLIENT, 11).unwrap();
+ assert_eq!(
+ sessions.iter_consumer_sessions().collect::<Vec<_>>(),
+ [ConsumerSession {
+ client_id: CLIENT,
+ session: 11
+ }]
+ );
+ sessions.login(2, USER).unwrap();
+ sessions.bind_session(2, CLIENT, 12).unwrap();
+ assert_eq!(sessions.iter_clients().count(), 2);
+ assert_eq!(
+ sessions.iter_consumer_sessions().collect::<Vec<_>>(),
+ [ConsumerSession {
+ client_id: CLIENT,
+ session: 12
+ }]
+ );
+ sessions.remove_connection(2);
+ assert_eq!(sessions.iter_consumer_sessions().count(), 0);
+ }
+
#[test]
fn full_lifecycle() {
let mut mgr = SessionManager::new();
diff --git a/core/server/src/shell.rs b/core/server/src/shell.rs
index 5c95fef00..0a186c208 100644
--- a/core/server/src/shell.rs
+++ b/core/server/src/shell.rs
@@ -23,6 +23,7 @@
//! Everything here is type- and config-level; construction (wiring the
//! handlers against a live bus) stays in [`crate::boot`].
+use crate::consumer_group::liveness::ConsumerGroupLiveness;
use crate::session_manager::SessionManager;
use configs::server::ServerConfig;
use consensus::{ConsensusTimers, VsrConsensus};
@@ -93,6 +94,8 @@ pub struct ShellHandlers {
/// Bound by the client-request handler, read by the get-clients
/// handler; the caller keeps it to reach locally-homed sessions.
pub sessions: Rc<RefCell<SessionManager>>,
+ /// Volatile leases shared by replica heartbeat ingress and shard 0's expiry task.
+ pub consumer_group_liveness: Rc<RefCell<ConsumerGroupLiveness>>,
}
impl ShellHandlers {
@@ -108,6 +111,7 @@ impl ShellHandlers {
on_metadata_submit: Rc::new(|_| {}),
on_list_clients: Rc::new(|_| {}),
sessions: Rc::new(RefCell::new(SessionManager::new())),
+ consumer_group_liveness: Rc::default(),
}
}
}
diff --git a/core/server_common/src/consensus_message.rs b/core/server_common/src/consensus_message.rs
index 0e4af1e11..077d09bd5 100644
--- a/core/server_common/src/consensus_message.rs
+++ b/core/server_common/src/consensus_message.rs
@@ -19,8 +19,8 @@ use crate::iobuf::{Frozen, Owned};
use crate::sharding::METADATA_GROUP;
use aligned_vec::{AVec, ConstAlign};
use iggy_binary_protocol::{
- Command, CommitHeader, ConsensusError, ConsensusHeader, DoViewChangeHeader,
- ForwardLogoutHeader, ForwardLogoutResultHeader, ForwardRegisterHeader,
+ Command, CommitHeader, ConsensusError, ConsensusHeader, ConsumerSessionHeartbeatHeader,
+ DoViewChangeHeader, ForwardLogoutHeader, ForwardLogoutResultHeader, ForwardRegisterHeader,
ForwardRegisterResultHeader, GenericHeader, HEADER_SIZE, Operation, PrepareHeader,
PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader, RequestHeader,
RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
@@ -644,6 +644,8 @@ pub enum MessageBag {
ForwardLogout(Message<ForwardLogoutHeader>),
/// The primary's verdict, routed back to the parked logout by nonce.
ForwardLogoutResult(Message<ForwardLogoutResultHeader>),
+ /// A backup reports session liveness to the metadata primary without a reply.
+ ConsumerSessionHeartbeat(Message<ConsumerSessionHeartbeatHeader>),
}
impl MessageBag {
@@ -725,6 +727,9 @@ impl MessageBag {
Self::ForwardRegister(message) => (message.header().operation(), METADATA_GROUP),
Self::ForwardRegisterResult(message) => (message.header().operation(), METADATA_GROUP),
Self::ForwardLogout(message) => (message.header().operation(), METADATA_GROUP),
+ Self::ConsumerSessionHeartbeat(message) => {
+ (message.header().operation(), METADATA_GROUP)
+ }
Self::ForwardLogoutResult(message) => (message.header().operation(), METADATA_GROUP),
}
}
@@ -756,6 +761,7 @@ impl MessageBag {
Self::ForwardRegister(message) => message.into_generic(),
Self::ForwardRegisterResult(message) => message.into_generic(),
Self::ForwardLogout(message) => message.into_generic(),
+ Self::ConsumerSessionHeartbeat(message) => message.into_generic(),
Self::ForwardLogoutResult(message) => message.into_generic(),
}
}
@@ -781,6 +787,7 @@ impl MessageBag {
Self::ForwardRegister(message) => message.header().command,
Self::ForwardRegisterResult(message) => message.header().command,
Self::ForwardLogout(message) => message.header().command,
+ Self::ConsumerSessionHeartbeat(message) => message.header().command,
Self::ForwardLogoutResult(message) => message.header().command,
}
}
@@ -806,6 +813,7 @@ impl MessageBag {
Self::ForwardRegister(message) => message.header().size(),
Self::ForwardRegisterResult(message) => message.header().size(),
Self::ForwardLogout(message) => message.header().size(),
+ Self::ConsumerSessionHeartbeat(message) => message.header().size(),
Self::ForwardLogoutResult(message) => message.header().size(),
}
}
@@ -831,6 +839,7 @@ impl MessageBag {
Self::ForwardRegister(message) => message.header().operation(),
Self::ForwardRegisterResult(message) => message.header().operation(),
Self::ForwardLogout(message) => message.header().operation(),
+ Self::ConsumerSessionHeartbeat(message) => message.header().operation(),
Self::ForwardLogoutResult(message) => message.header().operation(),
}
}
@@ -963,6 +972,9 @@ where
Command::ForwardLogoutResult => Ok(Self::ForwardLogoutResult(
value.try_into_typed::<ForwardLogoutResultHeader>()?,
)),
+ Command::ConsumerSessionHeartbeat => Ok(Self::ConsumerSessionHeartbeat(
+ value.try_into_typed::<ConsumerSessionHeartbeatHeader>()?,
+ )),
// Reply / Eviction are server-to-client frames; they do not
// appear on the inbound dispatch path.
Command::Reply | Command::Eviction => Err(ConsensusError::ClientBoundCommand(command)),
diff --git a/core/server_common/src/sharding/mod.rs b/core/server_common/src/sharding/mod.rs
index 58974159a..6e938c595 100644
--- a/core/server_common/src/sharding/mod.rs
+++ b/core/server_common/src/sharding/mod.rs
@@ -28,3 +28,13 @@ pub use namespace::{
};
pub use partition_location::PartitionLocation;
pub use shard_id::ShardId;
+
+/// Maximum time a client-list gather waits for all shard replies.
+pub const LIST_CLIENTS_GATHER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
+
+/// Maximum time spent retrying one gathered consumer-session report.
+pub const CONSUMER_SESSION_REPORT_TIMEOUT: std::time::Duration =
+ std::time::Duration::from_millis(100);
+
+pub const MIN_CONSUMER_SESSION_HEARTBEAT_SIZE: usize =
+ iggy_binary_protocol::HEADER_SIZE + iggy_binary_protocol::ConsumerSession::ENCODED_SIZE;
diff --git a/core/shard/src/lib.rs b/core/shard/src/lib.rs
index dd188fc0d..501c5fa82 100644
--- a/core/shard/src/lib.rs
+++ b/core/shard/src/lib.rs
@@ -41,12 +41,13 @@ use consensus::{
use crossfire::AsyncRxTrait;
use futures::FutureExt;
use iggy_binary_protocol::{
- CHECKSUM_UNSEALED, Command, CommitHeader, ConsensusHeader, DoViewChangeHeader,
- ForwardLogoutHeader, ForwardLogoutResultHeader, ForwardRegisterHeader,
- ForwardRegisterResultHeader, GenericHeader, Operation, PrepareHeader, PrepareOkHeader,
- RepairPrepareHeader, RepairRangeReplyHeader, RequestPreparesHeader, RequestStartViewHeader,
- RequestStateChunkHeader, RequestStateTransferHeader, RoutedRequestHeader,
- StartViewChangeHeader, StartViewHeader, StateChunkHeader, StateTransferTargetHeader,
+ CHECKSUM_UNSEALED, Command, CommitHeader, ConsensusHeader, ConsumerSession,
+ ConsumerSessionHeartbeatHeader, DoViewChangeHeader, ForwardLogoutHeader,
+ ForwardLogoutResultHeader, ForwardRegisterHeader, ForwardRegisterResultHeader, GenericHeader,
+ Operation, PrepareHeader, PrepareOkHeader, RepairPrepareHeader, RepairRangeReplyHeader,
+ RequestPreparesHeader, RequestStartViewHeader, RequestStateChunkHeader,
+ RequestStateTransferHeader, RoutedRequestHeader, StartViewChangeHeader, StartViewHeader,
+ StateChunkHeader, StateTransferTargetHeader,
};
#[cfg(feature = "simulator")]
use iggy_common::PartitionStats;
@@ -67,7 +68,9 @@ use partitions::state_transfer::TransferArtifact;
use partitions::{
FatalCommit, IggyPartition, IggyPartitions, PollFragments, PollingArgs, PollingConsumer,
};
-use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use server_common::sharding::{
+ IggyNamespace, LIST_CLIENTS_GATHER_TIMEOUT, PartitionLocation, ShardId,
+};
use server_common::{MESSAGE_ALIGN, Message, MessageBag, iobuf::Frozen};
use shards_table::ShardsTable;
use std::cell::{Cell, RefCell};
@@ -211,6 +214,8 @@ pub fn channel<T: Send + 'static>(capacity: usize) -> (Sender<T>, Receiver<T>) {
/// Logout preserves them so its caller can distinguish an unknown outcome
/// from a request that never entered the primary pipeline.
pub enum MetadataSubmit {
+ /// Replica-link liveness report routed to shard 0; no proposal or reply.
+ ConsumerSessionHeartbeat(Message<ConsumerSessionHeartbeatHeader>),
AttachConsumerSession {
vsr_client_id: u128,
session: u64,
@@ -320,12 +325,20 @@ pub struct ConnectedClientInfo {
/// Handler each shard runs for an inbound [`LifecycleFrame::ListClients`].
/// The server wires it to read the shard's `SessionManager` and push its
/// connected clients back over the carried reply sender.
-pub type ListClientsHandler = Rc<dyn Fn(Sender<Vec<ConnectedClientInfo>>)>;
+pub type ListClientsHandler = Rc<dyn Fn(ListClientsReply)>;
-/// Per-shard reply budget for the `list_all_clients` gather. A shard that
-/// doesn't answer within this window is skipped (partial result) so one
-/// wedged shard can't hang the read.
-const LIST_CLIENTS_GATHER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
+#[derive(Debug)]
+pub enum ListClientsReply {
+ Clients(Sender<Vec<ConnectedClientInfo>>),
+ Sessions(Sender<Vec<ConsumerSession>>),
+}
+
+/// Best-effort client list plus whether every shard answered.
+pub struct GatheredClients<T = ConnectedClientInfo> {
+ pub clients: Vec<T>,
+ /// False if any shard rejected the request or failed to reply before the deadline.
+ pub complete: bool,
+}
/// A read executed on the shard that owns a partition: a message poll or a
/// consumer-offset lookup. Carried by [`LifecycleFrame::PartitionRead`];
@@ -772,7 +785,7 @@ pub enum LifecycleFrame {
/// shard knows only its own connections). See
/// [`IggyShard::list_all_clients`].
ListClients {
- reply: Sender<Vec<ConnectedClientInfo>>,
+ reply: ListClientsReply,
},
/// Execute a partition read (message poll / consumer-offset lookup) on
/// the shard that owns `namespace` and push the result back over
@@ -1969,12 +1982,32 @@ where
/// treat the result as best-effort-complete.
#[allow(clippy::future_not_send)]
pub async fn list_all_clients(&self) -> Vec<ConnectedClientInfo> {
+ self.gather_clients().await.clients
+ }
+
+ /// Gather administrative client details with completeness information.
+ #[allow(clippy::future_not_send)]
+ pub async fn gather_clients(&self) -> GatheredClients {
+ self.gather_client_info(ListClientsReply::Clients).await
+ }
+
+ /// Gather bound session identities with completeness information for expiry decisions.
+ #[allow(clippy::future_not_send)]
+ pub async fn gather_consumer_sessions(&self) -> GatheredClients<ConsumerSession> {
+ self.gather_client_info(ListClientsReply::Sessions).await
+ }
+
+ #[allow(clippy::future_not_send)]
+ async fn gather_client_info<ClientInfo: Send + 'static>(
+ &self,
+ reply: fn(Sender<Vec<ClientInfo>>) -> ListClientsReply,
+ ) -> GatheredClients<ClientInfo> {
let shard_count = self.shard_count as usize;
- let (reply_tx, reply_rx) = channel::<Vec<ConnectedClientInfo>>(shard_count.max(1));
+ let (reply_tx, reply_rx) = channel::<Vec<ClientInfo>>(shard_count.max(1));
let mut expected = 0usize;
for sender in &self.senders {
let frame = ShardFrame::lifecycle(LifecycleFrame::ListClients {
- reply: reply_tx.clone(),
+ reply: reply(reply_tx.clone()),
});
if let Err(error) = sender.try_send(frame) {
tracing::warn!(
@@ -2022,7 +2055,10 @@ where
"list_all_clients: gather timed out; returning partial result"
);
}
- clients
+ GatheredClients {
+ clients,
+ complete: received == shard_count,
+ }
}
/// Run a partition read on the shard owning `namespace` and await the reply.
@@ -3238,6 +3274,13 @@ where
MessageBag::ForwardRegisterResult(ref msg) => {
self.on_forward_register_result(*msg.header());
}
+ MessageBag::ConsumerSessionHeartbeat(msg) => {
+ if self.peer_is_known(msg.header().replica, "ConsumerSessionHeartbeat")
+ && control_suffix_body_verified(&msg, msg.header().checksum_body).is_some()
+ {
+ (self.on_metadata_submit)(MetadataSubmit::ConsumerSessionHeartbeat(msg));
+ }
+ }
MessageBag::ForwardLogout(ref msg) => self.on_forward_logout(*msg.header()),
MessageBag::ForwardLogoutResult(ref msg) => {
self.on_forward_logout_result(*msg.header());
diff --git a/core/shard/src/poll/completion_tests.rs b/core/shard/src/poll/completion_tests.rs
index 2d0d8cc83..203ed3047 100644
--- a/core/shard/src/poll/completion_tests.rs
+++ b/core/shard/src/poll/completion_tests.rs
@@ -190,6 +190,7 @@ async fn given_pending_attached_poll_when_metadata_changes_should_fence_only_aff
group_id: WireIdentifier::numeric(u32::try_from(GROUP).unwrap()),
client_id: CLIENT,
in_flight: Vec::new(),
+ session: None,
}
.to_bytes(),
),
diff --git a/core/simulator/README.md b/core/simulator/README.md
index f1829b6c7..bf1e2a576 100644
--- a/core/simulator/README.md
+++ b/core/simulator/README.md
@@ -202,6 +202,7 @@ Constructors: `Simulator::new` (one shard per replica), `with_shards` (metadata
## Not modelled
+- Client heartbeat verification and periodic consumer-group session expiry. Shell mode runs the real disconnect callback, so tests can reproduce a rejected or skipped disconnect `Logout`. The consumer-group disconnect tests invoke the expiry cleanup operation explicitly; they do not cover the production timer or automatic liveness reporting.
- Storage faults, beyond two knobs on the superblock. `SimSuperblock::set_fail_writes` and `set_yield_writes` inject a persistent write fault and an fsync-wide suspension point; `MemStorage` under the journal never fails and never tears a write. Partition superblocks are storeless, which leaves partition view recovery untested.
- Segment files. Partition messages live in memory; they survive a restart only because the harness carries `RetainedPartitionState` across the rebuild.
- Partition-plane durability. Production's `load_partition` restores the view alone, so a restarted replica rejoins at op 0, invisible to quorum. `--restore-partition-frontier` looks past that at a system more durable than Iggy is.
diff --git a/core/simulator/src/client.rs b/core/simulator/src/client.rs
index 1c78d8b05..a8ab1010b 100644
--- a/core/simulator/src/client.rs
+++ b/core/simulator/src/client.rs
@@ -19,7 +19,7 @@ use bytes::{Bytes, BytesMut};
use iggy_binary_protocol::codes::{GET_STREAM_CODE, POLL_MESSAGES_CODE};
use iggy_binary_protocol::primitives::consumer::WireConsumer;
use iggy_binary_protocol::requests::consumer_groups::{
- CreateConsumerGroupRequest, DeleteConsumerGroupRequest,
+ CreateConsumerGroupRequest, DeleteConsumerGroupRequest, JoinConsumerGroupRequest,
};
use iggy_binary_protocol::requests::consumer_offsets::{
DeleteConsumerOffsetRequest, StoreConsumerOffsetRequest,
@@ -427,6 +427,22 @@ impl SimClient {
self.build_request(Operation::CreateConsumerGroup, &wire.to_bytes())
}
+ /// # Panics
+ /// Panics if any identifier is not a valid name.
+ pub fn join_consumer_group(
+ &self,
+ stream: &str,
+ topic: &str,
+ group: &str,
+ ) -> Message<RoutedRequestHeader> {
+ let wire = JoinConsumerGroupRequest {
+ stream_id: WireIdentifier::named(stream).expect("stream name must be valid"),
+ topic_id: WireIdentifier::named(topic).expect("topic name must be valid"),
+ group_id: WireIdentifier::named(group).expect("group name must be valid"),
+ };
+ self.build_request(Operation::JoinConsumerGroup, &wire.to_bytes())
+ }
+
/// # Panics
/// Panics if `stream`, `topic`, or `group` is not a valid `WireName`.
pub fn delete_consumer_group(
diff --git a/core/simulator/src/lib.rs b/core/simulator/src/lib.rs
index 8077668ed..cae0418c3 100644
--- a/core/simulator/src/lib.rs
+++ b/core/simulator/src/lib.rs
@@ -1689,9 +1689,279 @@ mod tests {
use bytes::Bytes;
use consensus::Status;
use futures::FutureExt;
- use iggy_binary_protocol::{AckLevel, RoutedRequestHeader};
+ use iggy_binary_protocol::{AckLevel, RoutedRequestHeader, WireIdentifier};
use iggy_common::ConsumerKind;
- use server_common::sharding::IggyNamespace;
+ use server_common::sharding::{IggyNamespace, LIST_CLIENTS_GATHER_TIMEOUT};
+
+ const DISCONNECT_STREAM: &str = "sim-stream-0";
+ const DISCONNECT_TOPIC: &str = "sim-topic-0-0";
+ const DISCONNECT_GROUP: &str = "disconnect-recovery";
+ const DISCONNECT_PROGRESS_STEPS: usize = 200;
+
+ #[test]
+ fn given_pending_metadata_when_disconnected_session_expires_should_reassign_partition() {
+ let (mut sim, original, replacement) = consumer_group_disconnect_fixture(3, 2);
+ let pending = original.create_stream("disconnect-in-flight");
+ sim.submit_request(original.client_id(), 0, pending.into_generic());
+ assert!(
+ (0..DISCONNECT_PROGRESS_STEPS).any(|_| {
+ sim.step();
+ sim.metadata_consensus(0)
+ .unwrap()
+ .pipeline_has_message_from_client(original.client_id())
+ }),
+ "disconnect must happen while the original client's metadata request is pending"
+ );
+ sim.outboxes[0].notify_client_connection_lost(original.client_id());
+ sim.run_pumps();
+ assert!(
+ (0..DISCONNECT_PROGRESS_STEPS).any(|_| {
+ sim.step();
+ !sim.metadata_consensus(0)
+ .unwrap()
+ .pipeline_has_message_from_client(original.client_id())
+ }),
+ "the pending request must finish after the disconnect"
+ );
+ sim.shell_login(&replacement);
+ let joined = submit_and_wait_for_reply(
+ &mut sim,
+ replacement.client_id(),
+ 0,
+ replacement.join_consumer_group(DISCONNECT_STREAM, DISCONNECT_TOPIC, DISCONNECT_GROUP),
+ );
+ assert_eq!(joined.header().status, 0);
+ for _ in 0..DISCONNECT_PROGRESS_STEPS {
+ sim.step();
+ }
+ recover_disconnected_group(&mut sim, &original, &replacement);
+ }
+
+ #[test]
+ fn given_evicted_client_table_entry_when_disconnected_session_expires_should_reassign_partition()
+ {
+ let (mut sim, original, replacement) = consumer_group_disconnect_fixture(1, 1);
+ sim.shell_login(&replacement);
+ assert_eq!(
+ sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .client_table
+ .borrow()
+ .get_epoch(original.client_id()),
+ None,
+ "the new registration must evict the original client-table entry"
+ );
+ sim.outboxes[0].notify_client_connection_lost(original.client_id());
+ sim.run_pumps();
+ let joined = submit_and_wait_for_reply(
+ &mut sim,
+ replacement.client_id(),
+ 0,
+ replacement.join_consumer_group(DISCONNECT_STREAM, DISCONNECT_TOPIC, DISCONNECT_GROUP),
+ );
+ assert_eq!(joined.header().status, 0);
+ for _ in 0..DISCONNECT_PROGRESS_STEPS {
+ sim.step();
+ }
+ recover_disconnected_group(&mut sim, &original, &replacement);
+ }
+
+ fn consumer_group_disconnect_fixture(
+ replica_count: u8,
+ client_capacity: usize,
+ ) -> (Simulator, SimClient, SimClient) {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let original = SimClient::new(1);
+ let replacement = SimClient::new(2);
+ let mut sim = Simulator::with_shards_shell(
+ usize::from(replica_count),
+ 1,
+ [original.client_id(), replacement.client_id()].into_iter(),
+ packet::PacketSimulatorOptions {
+ node_count: replica_count,
+ client_count: 2,
+ seed: 0x4273,
+ ..packet::PacketSimulatorOptions::default()
+ },
+ );
+ for replica in &sim.replicas {
+ replica.shards[0]
+ .plane
+ .metadata()
+ .client_table
+ .borrow_mut()
+ .set_capacity(client_capacity);
+ }
+ let namespace = IggyNamespace::new(0, 0, 0);
+ sim.init_partition(namespace);
+ sim.seed_stream_topic_partition(namespace);
+ sim.shell_login(&original);
+ for request in [
+ original.create_consumer_group(DISCONNECT_STREAM, DISCONNECT_TOPIC, DISCONNECT_GROUP),
+ original.join_consumer_group(DISCONNECT_STREAM, DISCONNECT_TOPIC, DISCONNECT_GROUP),
+ ] {
+ let reply = submit_and_wait_for_reply(&mut sim, original.client_id(), 0, request);
+ assert_eq!(reply.header().status, 0);
+ }
+ assert_eq!(
+ disconnect_group_assignment(&sim, original.client_id()),
+ Some(vec![0]),
+ "original consumer must own the only partition before disconnecting"
+ );
+ (sim, original, replacement)
+ }
+
+ fn disconnect_group_assignment(sim: &Simulator, client_id: u128) -> Option<Vec<u32>> {
+ sim.replicas[0].shards[0]
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_member_assignment(
+ &WireIdentifier::numeric(0),
+ &WireIdentifier::numeric(0),
+ &WireIdentifier::named(DISCONNECT_GROUP).unwrap(),
+ client_id,
+ )
+ .map(|(_, partitions)| partitions)
+ }
+
+ fn recover_disconnected_group(
+ sim: &mut Simulator,
+ original: &SimClient,
+ replacement: &SimClient,
+ ) {
+ assert_eq!(
+ disconnect_group_assignment(sim, original.client_id()),
+ Some(vec![0]),
+ "failed disconnect cleanup must leave the original owning the only partition"
+ );
+ assert_eq!(
+ disconnect_group_assignment(sim, replacement.client_id()),
+ Some(vec![]),
+ "the replacement joins successfully but cannot poll any partition"
+ );
+ let owner = Rc::clone(&sim.replicas[0].shards[0]);
+ let session = owner
+ .plane
+ .metadata()
+ .mux_stm
+ .streams()
+ .consumer_group_session(original.client_id())
+ .expect("the orphan retains its session even after client-table eviction");
+ let original_id = original.client_id();
+ let replacement_id = replacement.client_id();
+ let (complete, completed) = shard::channel(1);
+ // The simulator does not run the wall-clock liveness task. Exercise its
+ // committed cleanup separately from timeout selection and reporting.
+ sim.executor.spawn(async move {
+ let live = owner.gather_consumer_sessions().await;
+ assert!(live.complete);
+ assert!(
+ live.clients
+ .iter()
+ .all(|client| client.client_id != original_id),
+ "the real disconnect callback must remove the original connection"
+ );
+ assert!(
+ live.clients
+ .iter()
+ .any(|client| client.client_id == replacement_id)
+ );
+ let cleanup = owner
+ .plane
+ .metadata()
+ .submit_expired_logout_in_process(original_id, Some(session))
+ .await;
+ complete.try_send(cleanup).unwrap();
+ });
+ let cleanup = (0..DISCONNECT_PROGRESS_STEPS)
+ .find_map(|_| {
+ sim.step();
+ completed.recv().now_or_never()
+ })
+ .expect("session expiry cleanup must complete")
+ .expect("cleanup result channel stays open")
+ .expect("the primary must accept expiry cleanup");
+ assert!(cleanup.is_some(), "expiry must commit a Logout");
+ assert_eq!(disconnect_group_assignment(sim, original_id), None);
+ assert_eq!(
+ disconnect_group_assignment(sim, replacement_id),
+ Some(vec![0]),
+ "expiry must give the existing replacement the orphan's partition"
+ );
+ }
+
+ #[test]
+ fn given_a_shard_that_stops_answering_when_gathering_clients_should_report_incomplete() {
+ server_common::MemoryPool::init_pool(&server_common::MemoryPoolSettings {
+ enabled: false,
+ size: iggy_common::IggyByteSize::from(0u64),
+ bucket_capacity: 1,
+ });
+ let network_options = packet::PacketSimulatorOptions {
+ node_count: 1,
+ client_count: 1,
+ seed: 0x4273,
+ ..packet::PacketSimulatorOptions::default()
+ };
+ let mut sim = Simulator::with_shards_shell(1, 2, std::iter::once(1), network_options);
+ sim.shell_login(&SimClient::new(1));
+ for missing_shard in [false, true] {
+ if missing_shard {
+ sim.executor.abort(sim.replicas[0].pump_tasks[1]);
+ }
+ let shard = Rc::clone(&sim.replicas[0].shards[0]);
+ let result = Rc::new(RefCell::new(None));
+ let gathered = Rc::clone(&result);
+ sim.executor.spawn(async move {
+ *gathered.borrow_mut() = Some(futures::join!(
+ shard.gather_clients(),
+ shard.gather_consumer_sessions()
+ ));
+ });
+ assert!(matches!(
+ sim.executor.run_until_stalled(POLL_BUDGET),
+ RunOutcome::Quiescent { .. }
+ ));
+ if missing_shard {
+ assert!(
+ result.borrow().is_none(),
+ "gather must wait for the missing shard until its deadline"
+ );
+ sim.executor.advance_time(LIST_CLIENTS_GATHER_TIMEOUT);
+ assert!(matches!(
+ sim.executor.run_until_stalled(POLL_BUDGET),
+ RunOutcome::Quiescent { .. }
+ ));
+ }
+ let (gathered, sessions) = result
+ .borrow_mut()
+ .take()
+ .expect("gather must complete within its deadline");
+ assert_eq!(gathered.complete, !missing_shard);
+ assert_eq!(sessions.complete, !missing_shard);
+ if !missing_shard {
+ assert!(
+ gathered
+ .clients
+ .iter()
+ .any(|client| client.vsr_client_id == Some(1))
+ );
+ assert!(
+ sessions
+ .clients
+ .iter()
+ .any(|session| session.client_id == 1)
+ );
+ }
+ }
+ }
pub fn submit_and_wait_for_reply(
sim: &mut Simulator,
diff --git a/core/simulator/src/packet.rs b/core/simulator/src/packet.rs
index 0f3e8684d..279e73ebb 100644
--- a/core/simulator/src/packet.rs
+++ b/core/simulator/src/packet.rs
@@ -335,7 +335,7 @@ pub struct PacketSimulator {
/// One past the highest [`Command`] discriminant, sizing [`COMMAND_LABELS`] and
/// the delivery counters. Raising it is part of adding a command.
-pub const COMMAND_COUNT_MAX: usize = 30;
+pub const COMMAND_COUNT_MAX: usize = 31;
/// Names for each [`Command`] discriminant, so a coverage report reads as
/// protocol rather than as integers. Indexed by discriminant; the trailing
@@ -371,13 +371,14 @@ pub const COMMAND_LABELS: [&str; COMMAND_COUNT_MAX] = [
"ForwardRegisterResult",
"ForwardLogout",
"ForwardLogoutResult",
+ "ConsumerSessionHeartbeat",
];
const _: () = {
// Adding a command without extending the table would report it under the wrong
// name or index past the end of `command_counts`. Asserts the TABLE's length
- // rather than pinning one variant to the end: `ForwardLogoutResult ==
- // COMMAND_COUNT_MAX - 1` still holds after a `NewThing = 30` is appended past
+ // rather than pinning one variant to the end: `ConsumerSessionHeartbeat ==
+ // COMMAND_COUNT_MAX - 1` still holds after a `NewThing = 31` is appended past
// it, so that form passed exactly when it needed to fire.
assert!(COMMAND_LABELS.len() == COMMAND_COUNT_MAX);
assert!(enumset::EnumSet::<Command>::variant_count() as usize == COMMAND_COUNT_MAX);
diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs
index 3b963168b..8ee2d0365 100644
--- a/core/simulator/src/replica.rs
+++ b/core/simulator/src/replica.rs
@@ -407,8 +407,18 @@ pub fn new_shard(
on_client_request,
on_metadata_submit,
on_list_clients,
- // Step 6 keeps this to register client sessions; unused shell-off.
+ // TODO: Model automatic consumer liveness in shell-mode simulations:
+ // - Retain both handles. Run heartbeat verification on every shard and
+ // consumer_group::liveness::run on shard 0 of every replica for session
+ // heartbeat reporting and expiry. Support simulated shard/bus types.
+ // - Drive client heartbeats, task timers, and connection/lease timestamps
+ // with virtual time instead of compio timers and Instant::now().
+ // - Stop these tasks on replica shutdown/crash and recreate them on restart.
+ // - Test missed heartbeats and failed disconnect Logout by advancing virtual
+ // time, without calling expiry cleanup directly. Verify partition reassignment,
+ // saved offsets, and preservation of live or reconnected consumers.
sessions: _,
+ consumer_group_liveness: _,
} = if shell {
wire_shell_handlers(
&SharedSimOutbox(Rc::clone(bus)),