krabka_protocol/krabka/barrier/
alter_barrier_groups.rs

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

krabka_protocol/krabka/barrier/
alter_barrier_groups.rs
