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

krabka_protocol/krabka/barrier/
alter_barrier_groups.rs
