krabka_client_admin/
consumer_group_listings.rs

1//! [`AdminClient::list_consumer_groups`], Apache Kafka's older
2//! `listConsumerGroups`.
3//!
4//! Kafka's `listConsumerGroups` sends `ListGroups` to every broker with the
5//! state and type filters of its options, and keeps each listed group whose
6//! protocol type is `consumer` or empty. That is
7//! [`AdminClient::list_groups`] with the protocol types `""` and
8//! `"consumer"`, so this call runs it and reads each listing as Kafka's
9//! `ConsumerGroupListing`.
10
11use std::collections::BTreeSet;
12
13use crate::{
14    AdminClient, AdminError,
15    groups::{GroupListing, GroupState, GroupType, ListGroupsError, ListGroupsOptions},
16};
17
18/// The protocol type of a classic consumer group
19/// (`ConsumerProtocol.PROTOCOL_TYPE`).
20const CONSUMER_PROTOCOL_TYPE: &str = "consumer";
21
22/// The options of [`AdminClient::list_consumer_groups`], as Apache Kafka's
23/// `ListConsumerGroupsOptions` holds them. An empty set does not filter.
24#[derive(Debug, Clone, Default, PartialEq, Eq, bon::Builder)]
25pub struct ListConsumerGroupsOptions {
26    /// The broker lists only groups in these states (`states_filter`).
27    #[builder(default)]
28    pub group_states: BTreeSet<GroupState>,
29    /// The broker lists only groups of these types (`types_filter`).
30    #[builder(default)]
31    pub types: BTreeSet<GroupType>,
32}
33
34/// One group of [`AdminClient::list_consumer_groups`], as Apache Kafka's
35/// `ConsumerGroupListing` holds it.
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct ConsumerGroupListing {
38    pub group_id: String,
39    /// Whether the group has no protocol type.
40    pub is_simple_consumer_group: bool,
41    /// `None` when the broker sends no state (`ListGroups` v3 or lower).
42    pub group_state: Option<GroupState>,
43    /// `None` when the broker sends no type (`ListGroups` v4 or lower).
44    pub group_type: Option<GroupType>,
45}
46
47impl From<GroupListing> for ConsumerGroupListing {
48    fn from(listing: GroupListing) -> Self {
49        Self {
50            is_simple_consumer_group: listing.protocol_type.is_empty(),
51            group_id: listing.group_id,
52            group_state: listing.group_state,
53            group_type: listing.group_type,
54        }
55    }
56}
57
58/// The result of [`AdminClient::list_consumer_groups`], as Apache Kafka's
59/// `ListConsumerGroupsResult` holds it.
60#[derive(Debug, Clone, PartialEq, Eq)]
61pub struct ListConsumerGroupsResult {
62    /// The consumer groups of every broker that answered, one entry for each
63    /// group id, in group id order.
64    pub valid: Vec<ConsumerGroupListing>,
65    /// One entry for each broker that failed, in metadata order.
66    pub errors: Vec<ListGroupsError>,
67}
68
69impl ListConsumerGroupsResult {
70    /// The groups when no broker failed, as Kafka's
71    /// `ListConsumerGroupsResult.all` gives them.
72    ///
73    /// # Errors
74    /// Returns the first broker failure.
75    pub fn all(self) -> Result<Vec<ConsumerGroupListing>, ListGroupsError> {
76        match self.errors.into_iter().next() {
77            Some(error) => Err(error),
78            None => Ok(self.valid),
79        }
80    }
81}
82
83impl AdminClient {
84    /// Lists the consumer groups of the whole cluster, as Apache Kafka's
85    /// `listConsumerGroups` does.
86    ///
87    /// The call sends `ListGroups` to every broker with the state and type
88    /// filters of `options`, and keeps each group whose protocol type is
89    /// `consumer` or empty. See [`AdminClient::list_groups`] for the fan-out,
90    /// the retries, the per-broker errors and the version that each filter
91    /// needs.
92    ///
93    /// # Errors
94    /// Returns an error when the `Metadata` request fails, or when the
95    /// metadata names no broker after the timeout. A failure on one broker is
96    /// an entry in [`ListConsumerGroupsResult::errors`].
97    pub async fn list_consumer_groups(
98        &self,
99        options: &ListConsumerGroupsOptions,
100    ) -> Result<ListConsumerGroupsResult, AdminError> {
101        let result = self
102            .list_groups(
103                &ListGroupsOptions::builder()
104                    .group_states(options.group_states.clone())
105                    .types(options.types.clone())
106                    .protocol_types(maplit::btreeset! {
107                        String::new(),
108                        CONSUMER_PROTOCOL_TYPE.to_owned(),
109                    })
110                    .build(),
111            )
112            .await?;
113        Ok(ListConsumerGroupsResult {
114            valid: result
115                .valid
116                .into_iter()
117                .map(ConsumerGroupListing::from)
118                .collect(),
119            errors: result.errors,
120        })
121    }
122}
123
124#[cfg(test)]
125mod tests {
126    use std::sync::{Arc, Mutex};
127
128    use assert2::assert;
129    use krabka_client_core::MockReply;
130    use krabka_protocol::owned::{
131        list_groups_request::{self, ListGroupsRequest},
132        list_groups_response::{ListGroupsResponse, ListedGroup},
133        metadata_request,
134        metadata_response::{MetadataResponse, MetadataResponseBroker},
135    };
136
137    use super::*;
138    use crate::partition_leaders::test_support::{
139        ApiVersionRange, decode_request, encode_response, fast_admin, scripted_broker,
140    };
141
142    fn listed(group_id: &str, protocol_type: &str, group_type: &str) -> ListedGroup {
143        ListedGroup {
144            group_id: group_id.to_owned(),
145            protocol_type: protocol_type.to_owned(),
146            group_state: "Stable".to_owned(),
147            group_type: group_type.to_owned(),
148            ..Default::default()
149        }
150    }
151
152    /// Kafka's `listConsumerGroups` sends the state and type filters and
153    /// keeps the groups whose protocol type is `consumer` or empty, marking a
154    /// group with no protocol type as simple.
155    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
156    async fn list_consumer_groups_keeps_consumer_protocol_groups() {
157        let requests = Arc::new(Mutex::new(Vec::new()));
158        let seen = Arc::clone(&requests);
159        let broker = scripted_broker(
160            vec![
161                ApiVersionRange::builder()
162                    .api_key(list_groups_request::API_KEY)
163                    .max_version(5)
164                    .build(),
165                ApiVersionRange::builder()
166                    .api_key(metadata_request::API_KEY)
167                    .min_version(12)
168                    .max_version(12)
169                    .build(),
170            ],
171            move |api_key, version, body, own| match api_key {
172                metadata_request::API_KEY => MockReply::Respond(encode_response(
173                    &MetadataResponse {
174                        brokers: vec![MetadataResponseBroker {
175                            node_id: 1,
176                            host: own.ip().to_string(),
177                            port: i32::from(own.port()),
178                            ..Default::default()
179                        }],
180                        ..Default::default()
181                    },
182                    version,
183                    true,
184                )),
185                list_groups_request::API_KEY => {
186                    let flexible = version >= list_groups_request::FLEXIBLE_MIN;
187                    let request: ListGroupsRequest = decode_request(body, version, flexible);
188                    seen.lock().expect("requests lock").push((version, request));
189                    MockReply::Respond(encode_response(
190                        &ListGroupsResponse {
191                            groups: vec![
192                                listed("kip848", "consumer", "Consumer"),
193                                listed("simple", "", "Classic"),
194                                listed("connect", "connect", "Classic"),
195                                listed("share", "share", "Share"),
196                            ],
197                            ..Default::default()
198                        },
199                        version,
200                        flexible,
201                    ))
202                }
203                _ => MockReply::Silent,
204            },
205        )
206        .await;
207        let admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
208
209        let result = admin
210            .list_consumer_groups(
211                &ListConsumerGroupsOptions::builder()
212                    .group_states(maplit::btreeset! {
213                        GroupState::Stable,
214                    })
215                    .types(BTreeSet::new())
216                    .build(),
217            )
218            .await
219            .expect("list_consumer_groups succeeds");
220
221        broker.stop();
222        let listing = |group_id: &str, is_simple_consumer_group, group_type| ConsumerGroupListing {
223            group_id: group_id.to_owned(),
224            is_simple_consumer_group,
225            group_state: Some(GroupState::Stable),
226            group_type: Some(group_type),
227        };
228        assert!(
229            (result, requests.lock().expect("requests lock").clone())
230                == (
231                    ListConsumerGroupsResult {
232                        valid: vec![
233                            listing("kip848", false, GroupType::Consumer),
234                            listing("simple", true, GroupType::Classic),
235                        ],
236                        errors: Vec::new(),
237                    },
238                    vec![(
239                        5,
240                        ListGroupsRequest {
241                            states_filter: vec!["Stable".to_owned()],
242                            ..Default::default()
243                        }
244                    )],
245                )
246        );
247    }
248}
krabka_client_admin/consumer_group_listings.rs

krabka_client_admin/
consumer_group_listings.rs
