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

krabka_protocol/krabka/barrier/
describe_barrier_groups.rs
