krabka_client_consumer/
group_metadata.rs

1//! KIP-447 consumer group metadata for a transactional producer's
2//! `send_offsets_to_transaction`.
3//!
4//! The group coordinator uses it to fence zombie producers with the consumer
5//! group's generation for a classic group, or with the member epoch for a
6//! KIP-848 next-gen group. This removes the need for one producer per input
7//! partition.
8
9/// The identity that a consumer presents to a transactional producer for
10/// KIP-447 offset-commit fencing.
11///
12/// This mirrors the JVM's
13/// `org.apache.kafka.clients.consumer.ConsumerGroupMetadata`.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct ConsumerGroupMetadata {
16    /// The consumer group id.
17    pub group_id: String,
18    /// Classic-group generation id, or the member epoch for a KIP-848 next-gen
19    /// group. The client sends it verbatim in
20    /// `TxnOffsetCommitRequest.generation_id`, which matches the JVM wire
21    /// convention. The coordinator interprets it per group kind.
22    pub generation_id: i32,
23    /// The member id that the coordinator assigned at join time. It is empty
24    /// for a simple consumer, which uses manual assignment and has no group
25    /// membership.
26    pub member_id: String,
27    /// `group.instance.id` for a static member. `None` for a dynamic member.
28    pub group_instance_id: Option<String>,
29}
30
31impl ConsumerGroupMetadata {
32    /// Metadata for a producer that commits offsets to a group it is not a
33    /// member of, that is a manual partition assignment or a simple consumer.
34    ///
35    /// The group coordinator applies no generation or member fencing to this
36    /// shape.
37    #[must_use]
38    pub fn for_group(group_id: impl Into<String>) -> Self {
39        Self {
40            group_id: group_id.into(),
41            generation_id: -1,
42            member_id: String::new(),
43            group_instance_id: None,
44        }
45    }
46}
47
48#[cfg(test)]
49mod tests {
50    use super::*;
51
52    #[test]
53    fn for_group_is_simple_consumer_shape() {
54        let m = ConsumerGroupMetadata::for_group("g");
55        assert2::assert!(
56            m == ConsumerGroupMetadata {
57                group_id: "g".into(),
58                generation_id: -1,
59                member_id: String::new(),
60                group_instance_id: None,
61            }
62        );
63    }
64}
krabka_client_consumer/group_metadata.rs

krabka_client_consumer/
group_metadata.rs
