Struct ConsumerGroup 

pub struct ConsumerGroup { /* private fields */ }

Kafka-style consumer group coordinator.

Manages consumer membership, partition assignment, offset tracking, and rebalancing across the group.

Implementations

impl ConsumerGroup

pub fn new( group_id: impl Into<String>, partition_count: u32, strategy: AssignmentStrategy, heartbeat_timeout_ms: u64, session_timeout_ms: u64, ) -> Self

Create a new consumer group.

partition_count is the number of topic partitions to manage. heartbeat_timeout_ms controls when a consumer is marked lagging. session_timeout_ms controls when a consumer is evicted.

pub fn group_id(&self) -> &str

Return the group identifier.

pub fn partition_count(&self) -> u32

Return the number of partitions.

pub fn generation(&self) -> u64

Return the current generation.

pub fn coordinator(&self) -> Option<&str>

Return the current coordinator, if any.

pub fn strategy(&self) -> &AssignmentStrategy

Return the current assignment strategy.

pub fn set_strategy(&mut self, strategy: AssignmentStrategy)

Set the assignment strategy.

pub fn member_count(&self) -> usize

Number of registered members.

pub fn get_member(&self, consumer_id: &str) -> Option<&ConsumerMember>

Get a member by ID.

pub fn get_offset(&self, partition: u32) -> Option<&PartitionOffset>

Get partition offset information.

pub fn rebalance_history(&self) -> &VecDeque<RebalanceEvent>

Return the rebalance history.

pub fn join( &mut self, consumer_id: impl Into<String>, now_ms: u64, ) -> Result<(), GroupError>

Join a consumer to the group.

Returns an error if the consumer already exists. The first consumer to join becomes the coordinator.

pub fn leave(&mut self, consumer_id: &str) -> Result<(), GroupError>

Remove a consumer from the group.

If the removed consumer was the coordinator, elects a new one.

pub fn heartbeat( &mut self, consumer_id: &str, now_ms: u64, ) -> Result<(), GroupError>

Record a heartbeat from a consumer.

pub fn check_heartbeats(&mut self, now_ms: u64) -> Vec<String>

Check all consumers for heartbeat timeouts.

Marks consumers as Lagging if they exceed heartbeat_timeout_ms, and evicts (marks Dead and removes partitions) those exceeding session_timeout_ms.

Returns the IDs of consumers that were evicted.

pub fn elect_coordinator(&mut self)

Elect a coordinator from the active members.

Selects the lexicographically smallest active consumer ID.

pub fn commit_offset( &mut self, partition: u32, offset: u64, ) -> Result<(), GroupError>

Commit an offset for a partition.

pub fn update_log_end_offset( &mut self, partition: u32, offset: u64, ) -> Result<(), GroupError>

Update the log-end offset for a partition (producer side).

pub fn total_lag(&self) -> u64

Calculate the total consumer lag across all partitions.

pub fn consumer_lag(&self) -> HashMap<String, u64>

Calculate per-consumer lag: sum of lags on partitions assigned to each consumer.

pub fn rebalance(&mut self, now_ms: u64) -> Result<RebalanceResult, GroupError>

Trigger a rebalance using the current assignment strategy.

Returns the result with partition assignments and movement count.

pub fn current_assignments(&self) -> HashMap<String, Vec<u32>>

Get current partition assignments as consumer_id -> partitions.

pub fn stats(&self) -> GroupStats

Aggregate statistics for the group.