Method join

nodedb/src/event/cdc/consumer_group/assignment.rs:74–88

Register a consumer joining a group. Triggers rebalance.

(&self, tenant_id: u64, stream: &str, group: &str, consumer_id: &str)

    /// Register a consumer joining a group. Triggers rebalance.
    pub fn join(&self, tenant_id: u64, stream: &str, group: &str, consumer_id: &str) {
        let key = (tenant_id, stream.to_string(), group.to_string());
        let mut groups = self.groups.write().unwrap_or_else(|p| p.into_inner());
        let state = groups.entry(key).or_insert_with(GroupAssignment::new);
        state.consumers.insert(consumer_id.to_string());
        state.rebalance();

        tracing::debug!(
            stream,
            group,
            consumer_id,
            total_consumers = state.consumers.len(),
            "consumer joined, rebalanced"
        );
    }

    /// Deregister a consumer leaving a group. Triggers rebalance.
    pub fn leave(&self, tenant_id: u64, stream: &str, group: &str, consumer_id: &str) {