krabka_protocol/krabka/barrier/
describe_barrier_groups.rs

1//! `DescribeBarrierGroups`, api key 1011.
2//!
3//! The request reads the definition of one or more barrier groups. An empty
4//! `groups` array asks for every group that the coordinator holds.
5
6use bytes::{Buf, BufMut};
7
8#[cfg(test)]
9use crate::krabka::test_support::sample_tagged_fields;
10use crate::{
11    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
12    krabka::common::{
13        I16_LEN, I32_LEN, I64_LEN, array_len, check_version, decode_array, decode_string_array,
14        encode_array, encode_string_array, read_unknown_tagged_fields, string_array_len,
15        unknown_tagged_fields_len, write_unknown_tagged_fields,
16    },
17    primitives::{
18        fixed::{get_i16, get_i32, get_i64, put_i16, put_i32, put_i64},
19        string_bytes::{
20            compact_nullable_string_len, compact_string_len, get_compact_nullable_string_owned,
21            get_compact_string_owned, put_compact_nullable_string, put_compact_string,
22        },
23    },
24};
25
26/// Api key of `DescribeBarrierGroups`.
27pub const API_KEY: i16 = 1011;
28/// Lowest version of `DescribeBarrierGroups` that this build speaks.
29pub const MIN_VERSION: i16 = 0;
30/// Highest version of `DescribeBarrierGroups` that this build speaks.
31pub const MAX_VERSION: i16 = 0;
32/// First version with flexible framing from KIP-482.
33pub const FLEXIBLE_MIN: i16 = 0;
34
35/// Reports whether `version` uses flexible framing from KIP-482.
36///
37/// The request header takes the same framing as the request body, so a caller
38/// that writes the header needs this answer.
39#[inline]
40#[must_use]
41pub fn is_flexible(version: i16) -> bool {
42    version >= FLEXIBLE_MIN
43}
44
45/// Reads the definition of one or more barrier groups.
46#[derive(Debug, Clone, PartialEq, Eq, Default)]
47pub struct DescribeBarrierGroupsRequest {
48    /// Names of the groups to describe. An empty array asks for every group.
49    pub groups: Vec<String>,
50    /// Tagged fields that this build does not know.
51    pub unknown_tagged_fields: UnknownTaggedFields,
52}
53
54impl Encode for DescribeBarrierGroupsRequest {
55    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
56        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
57        encode_string_array(buf, &self.groups)?;
58        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
59        Ok(())
60    }
61
62    fn encoded_len(&self, _version: i16) -> usize {
63        string_array_len(&self.groups) + unknown_tagged_fields_len(&self.unknown_tagged_fields)
64    }
65}
66
67impl Decode<'_> for DescribeBarrierGroupsRequest {
68    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
69        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
70        Ok(Self {
71            groups: decode_string_array(buf)?,
72            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
73        })
74    }
75}
76
77impl ProtocolRequest for DescribeBarrierGroupsRequest {
78    const API_KEY: i16 = API_KEY;
79    const MIN_VERSION: i16 = MIN_VERSION;
80    const MAX_VERSION: i16 = MAX_VERSION;
81    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
82    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
83
84    type Response = DescribeBarrierGroupsResponse;
85}
86
87/// The definition of one barrier group, or the error that hid it.
88#[derive(Debug, Clone, PartialEq, Eq)]
89pub struct DescribedBarrierGroup {
90    /// Name of the barrier group.
91    pub group: String,
92    /// Kafka error code. `0` means the rest of the fields describe the group.
93    pub error_code: i16,
94    /// Text that explains a non-zero `error_code`. `None` when the coordinator
95    /// has nothing to add to the code.
96    pub error_message: Option<String>,
97    /// Topics that the group covers.
98    pub topics: Vec<String>,
99    /// Period between two automatic injections, in milliseconds. `-1` means
100    /// that periodic injection is off for this group.
101    pub interval_ms: i64,
102    /// Number of published cuts that the coordinator keeps for this group.
103    pub retained_cuts: i32,
104    /// Epoch of the last cut that the coordinator started for this group.
105    /// `-1` means that the group has no cut yet.
106    pub last_epoch: i64,
107    /// Node id of the broker that coordinates this group. `-1` means that no
108    /// broker owns the group right now.
109    pub coordinator_id: i32,
110    /// Tagged fields that this build does not know.
111    pub unknown_tagged_fields: UnknownTaggedFields,
112}
113
114impl Default for DescribedBarrierGroup {
115    fn default() -> Self {
116        Self {
117            group: String::new(),
118            error_code: 0,
119            error_message: None,
120            topics: Vec::new(),
121            interval_ms: -1,
122            retained_cuts: 0,
123            last_epoch: -1,
124            coordinator_id: -1,
125            unknown_tagged_fields: UnknownTaggedFields::default(),
126        }
127    }
128}
129
130impl Encode for DescribedBarrierGroup {
131    fn encode<B: BufMut>(&self, buf: &mut B, _version: i16) -> Result<(), ProtocolError> {
132        put_compact_string(buf, &self.group)?;
133        put_i16(buf, self.error_code);
134        put_compact_nullable_string(buf, self.error_message.as_deref())?;
135        encode_string_array(buf, &self.topics)?;
136        put_i64(buf, self.interval_ms);
137        put_i32(buf, self.retained_cuts);
138        put_i64(buf, self.last_epoch);
139        put_i32(buf, self.coordinator_id);
140        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
141        Ok(())
142    }
143
144    fn encoded_len(&self, _version: i16) -> usize {
145        compact_string_len(&self.group)
146            + I16_LEN
147            + compact_nullable_string_len(self.error_message.as_deref())
148            + string_array_len(&self.topics)
149            + I64_LEN
150            + I32_LEN
151            + I64_LEN
152            + I32_LEN
153            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
154    }
155}
156
157impl Decode<'_> for DescribedBarrierGroup {
158    fn decode<B: Buf>(buf: &mut B, _version: i16) -> Result<Self, ProtocolError> {
159        Ok(Self {
160            group: get_compact_string_owned(buf)?,
161            error_code: get_i16(buf)?,
162            error_message: get_compact_nullable_string_owned(buf)?,
163            topics: decode_string_array(buf)?,
164            interval_ms: get_i64(buf)?,
165            retained_cuts: get_i32(buf)?,
166            last_epoch: get_i64(buf)?,
167            coordinator_id: get_i32(buf)?,
168            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
169        })
170    }
171}
172
173/// The definitions that a [`DescribeBarrierGroupsRequest`] asked for.
174#[derive(Debug, Clone, PartialEq, Eq, Default)]
175pub struct DescribeBarrierGroupsResponse {
176    /// Milliseconds that the broker held the request back for a quota.
177    pub throttle_time_ms: i32,
178    /// One entry for each group that the request named, or for every group when
179    /// the request named none.
180    pub groups: Vec<DescribedBarrierGroup>,
181    /// Tagged fields that this build does not know.
182    pub unknown_tagged_fields: UnknownTaggedFields,
183}
184
185impl Encode for DescribeBarrierGroupsResponse {
186    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
187        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
188        put_i32(buf, self.throttle_time_ms);
189        encode_array(buf, &self.groups, version)?;
190        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
191        Ok(())
192    }
193
194    fn encoded_len(&self, version: i16) -> usize {
195        I32_LEN
196            + array_len(&self.groups, version)
197            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
198    }
199}
200
201impl Decode<'_> for DescribeBarrierGroupsResponse {
202    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
203        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
204        Ok(Self {
205            throttle_time_ms: get_i32(buf)?,
206            groups: decode_array(buf, version)?,
207            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
208        })
209    }
210}
211
212#[cfg(test)]
213impl DescribeBarrierGroupsRequest {
214    /// Builds a request with every field set, for the round-trip tests.
215    fn populated() -> Self {
216        Self {
217            groups: vec!["orders".to_owned(), "audit".to_owned()],
218            unknown_tagged_fields: sample_tagged_fields(),
219        }
220    }
221}
222
223#[cfg(test)]
224impl DescribeBarrierGroupsResponse {
225    /// Builds a response with every field set, for the round-trip tests.
226    fn populated() -> Self {
227        Self {
228            throttle_time_ms: 5,
229            groups: vec![
230                DescribedBarrierGroup {
231                    group: "orders".to_owned(),
232                    error_code: 0,
233                    error_message: None,
234                    topics: vec!["orders".to_owned(), "payments".to_owned()],
235                    interval_ms: 30_000,
236                    retained_cuts: 64,
237                    last_epoch: 12,
238                    coordinator_id: 3,
239                    unknown_tagged_fields: sample_tagged_fields(),
240                },
241                DescribedBarrierGroup {
242                    group: "audit".to_owned(),
243                    error_code: 69,
244                    error_message: Some("barrier group not found".to_owned()),
245                    ..DescribedBarrierGroup::default()
246                },
247            ],
248            unknown_tagged_fields: UnknownTaggedFields::default(),
249        }
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use assert2::assert;
256
257    use super::*;
258    use crate::krabka::test_support::{roundtrip, unsupported_versions};
259
260    #[test]
261    fn default_roundtrips_all_versions() {
262        for version in MIN_VERSION..=MAX_VERSION {
263            roundtrip(&DescribeBarrierGroupsRequest::default(), version);
264            roundtrip(&DescribeBarrierGroupsResponse::default(), version);
265        }
266    }
267
268    #[test]
269    fn populated_roundtrips_all_versions() {
270        for version in MIN_VERSION..=MAX_VERSION {
271            roundtrip(&DescribeBarrierGroupsRequest::populated(), version);
272            roundtrip(&DescribeBarrierGroupsResponse::populated(), version);
273        }
274    }
275
276    #[test]
277    fn empty_group_list_asks_for_every_group() {
278        let request = DescribeBarrierGroupsRequest::default();
279        assert!(request.groups.is_empty());
280        roundtrip(&request, MIN_VERSION);
281    }
282
283    #[test]
284    fn rejects_unsupported_versions() {
285        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
286            let response = DescribeBarrierGroupsResponse::populated();
287            assert!(matches!(
288                response.encode(&mut Vec::<u8>::new(), version),
289                Err(ProtocolError::UnsupportedVersion {
290                    api_key: API_KEY,
291                    ..
292                })
293            ));
294        }
295    }
296}
krabka_protocol/krabka/barrier/describe_barrier_groups.rs

krabka_protocol/krabka/barrier/
describe_barrier_groups.rs
