krabka_protocol/krabka/barrier/
list_barrier_cuts.rs

1//! `ListBarrierCuts`, api key 1013.
2//!
3//! The request reads cuts that the coordinator published before. It is the RPC
4//! path to the manifest. Any Kafka consumer can read the same cuts from the
5//! `__barrier_state` topic, which is the portable path for a client in another
6//! language.
7
8use bytes::{Buf, BufMut};
9
10use crate::{
11    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
12    krabka::{
13        barrier::cut::{BarrierCutTopic, BarrierMissingPartition},
14        common::{
15            I8_LEN, I16_LEN, I32_LEN, I64_LEN, array_len, check_version, decode_array,
16            encode_array, read_unknown_tagged_fields, unknown_tagged_fields_len,
17            write_unknown_tagged_fields,
18        },
19    },
20    primitives::{
21        fixed::{get_i8, get_i16, get_i32, get_i64, put_i8, put_i16, put_i32, put_i64},
22        string_bytes::{
23            compact_nullable_string_len, compact_string_len, get_compact_nullable_string_owned,
24            get_compact_string_owned, put_compact_nullable_string, put_compact_string,
25        },
26    },
27};
28
29/// Api key of `ListBarrierCuts`.
30pub const API_KEY: i16 = 1013;
31/// Lowest version of `ListBarrierCuts` that this build speaks.
32pub const MIN_VERSION: i16 = 0;
33/// Highest version of `ListBarrierCuts` that this build speaks.
34pub const MAX_VERSION: i16 = 0;
35/// First version with flexible framing from KIP-482.
36pub const FLEXIBLE_MIN: i16 = 0;
37
38/// Reports whether `version` uses flexible framing from KIP-482.
39///
40/// The request header takes the same framing as the request body, so a caller
41/// that writes the header needs this answer.
42#[inline]
43#[must_use]
44pub fn is_flexible(version: i16) -> bool {
45    version >= FLEXIBLE_MIN
46}
47
48/// Reads the published cuts of one barrier group.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct ListBarrierCutsRequest {
51    /// Name of the barrier group to read.
52    pub group: String,
53    /// Lowest epoch to return. The coordinator returns cuts in epoch order from
54    /// this epoch upward, and it includes this epoch.
55    pub from_epoch: i64,
56    /// Largest number of cuts to return. `-1` asks for every retained cut, and
57    /// `0` asks for none.
58    pub max_results: i32,
59    /// Tagged fields that this build does not know.
60    pub unknown_tagged_fields: UnknownTaggedFields,
61}
62
63impl Default for ListBarrierCutsRequest {
64    fn default() -> Self {
65        Self {
66            group: String::new(),
67            from_epoch: 0,
68            max_results: -1,
69            unknown_tagged_fields: UnknownTaggedFields::default(),
70        }
71    }
72}
73
74impl Encode for ListBarrierCutsRequest {
75    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
76        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
77        put_compact_string(buf, &self.group);
78        put_i64(buf, self.from_epoch);
79        put_i32(buf, self.max_results);
80        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
81        Ok(())
82    }
83
84    fn encoded_len(&self, _version: i16) -> usize {
85        compact_string_len(&self.group)
86            + I64_LEN
87            + I32_LEN
88            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
89    }
90}
91
92impl Decode<'_> for ListBarrierCutsRequest {
93    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
94        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
95        Ok(Self {
96            group: get_compact_string_owned(buf)?,
97            from_epoch: get_i64(buf)?,
98            max_results: get_i32(buf)?,
99            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
100        })
101    }
102}
103
104impl ProtocolRequest for ListBarrierCutsRequest {
105    const API_KEY: i16 = API_KEY;
106    const MIN_VERSION: i16 = MIN_VERSION;
107    const MAX_VERSION: i16 = MAX_VERSION;
108    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
109    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
110
111    type Response = ListBarrierCutsResponse;
112}
113
114/// One published cut of a barrier group.
115#[derive(Debug, Clone, PartialEq, Eq, Default)]
116pub struct BarrierCut {
117    /// Epoch of this cut. The coordinator never reuses an epoch.
118    pub epoch: i64,
119    /// Milliseconds since the Unix epoch, at the moment the coordinator started
120    /// the injection.
121    pub triggered_at: i64,
122    /// Milliseconds since the Unix epoch, at the moment the coordinator
123    /// published the cut.
124    pub completed_at: i64,
125    /// [`CUT_STATUS_COMPLETE`](crate::krabka::barrier::CUT_STATUS_COMPLETE)
126    /// when every partition took a marker, and
127    /// [`CUT_STATUS_PARTIAL`](crate::krabka::barrier::CUT_STATUS_PARTIAL) when
128    /// at least one partition did not.
129    pub status: i8,
130    /// Cut offsets, grouped by topic.
131    pub topics: Vec<BarrierCutTopic>,
132    /// Partitions of the frozen target set that took no marker. This array is
133    /// empty for a complete cut.
134    pub missing: Vec<BarrierMissingPartition>,
135    /// Tagged fields that this build does not know.
136    pub unknown_tagged_fields: UnknownTaggedFields,
137}
138
139impl Encode for BarrierCut {
140    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
141        put_i64(buf, self.epoch);
142        put_i64(buf, self.triggered_at);
143        put_i64(buf, self.completed_at);
144        put_i8(buf, self.status);
145        encode_array(buf, &self.topics, version)?;
146        encode_array(buf, &self.missing, version)?;
147        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
148        Ok(())
149    }
150
151    fn encoded_len(&self, version: i16) -> usize {
152        I64_LEN
153            + I64_LEN
154            + I64_LEN
155            + I8_LEN
156            + array_len(&self.topics, version)
157            + array_len(&self.missing, version)
158            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
159    }
160}
161
162impl Decode<'_> for BarrierCut {
163    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
164        Ok(Self {
165            epoch: get_i64(buf)?,
166            triggered_at: get_i64(buf)?,
167            completed_at: get_i64(buf)?,
168            status: get_i8(buf)?,
169            topics: decode_array(buf, version)?,
170            missing: decode_array(buf, version)?,
171            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
172        })
173    }
174}
175
176/// The cuts that a [`ListBarrierCutsRequest`] asked for.
177#[derive(Debug, Clone, PartialEq, Eq, Default)]
178pub struct ListBarrierCutsResponse {
179    /// Milliseconds that the broker held the request back for a quota.
180    pub throttle_time_ms: i32,
181    /// Kafka error code. `0` means `cuts` holds the answer.
182    pub error_code: i16,
183    /// Text that explains a non-zero `error_code`. `None` when the coordinator
184    /// has nothing to add to the code.
185    pub error_message: Option<String>,
186    /// Cuts in epoch order, from the oldest epoch that the request asked for.
187    pub cuts: Vec<BarrierCut>,
188    /// Tagged fields that this build does not know.
189    pub unknown_tagged_fields: UnknownTaggedFields,
190}
191
192impl Encode for ListBarrierCutsResponse {
193    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
194        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
195        put_i32(buf, self.throttle_time_ms);
196        put_i16(buf, self.error_code);
197        put_compact_nullable_string(buf, self.error_message.as_deref());
198        encode_array(buf, &self.cuts, version)?;
199        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
200        Ok(())
201    }
202
203    fn encoded_len(&self, version: i16) -> usize {
204        I32_LEN
205            + I16_LEN
206            + compact_nullable_string_len(self.error_message.as_deref())
207            + array_len(&self.cuts, version)
208            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
209    }
210}
211
212impl Decode<'_> for ListBarrierCutsResponse {
213    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
214        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
215        Ok(Self {
216            throttle_time_ms: get_i32(buf)?,
217            error_code: get_i16(buf)?,
218            error_message: get_compact_nullable_string_owned(buf)?,
219            cuts: decode_array(buf, version)?,
220            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
221        })
222    }
223}
224
225#[cfg(test)]
226impl ListBarrierCutsRequest {
227    /// Builds a request with every field set, for the round-trip tests.
228    fn populated() -> Self {
229        Self {
230            group: "orders".to_string(),
231            from_epoch: 12,
232            max_results: 100,
233            unknown_tagged_fields: crate::krabka::test_support::sample_tagged_fields(),
234        }
235    }
236}
237
238#[cfg(test)]
239impl ListBarrierCutsResponse {
240    /// Builds a response with every field set, for the round-trip tests.
241    fn populated() -> Self {
242        Self {
243            throttle_time_ms: 9,
244            error_code: 0,
245            error_message: None,
246            cuts: vec![
247                BarrierCut {
248                    epoch: 12,
249                    triggered_at: 1_724_500_000_000,
250                    completed_at: 1_724_500_000_120,
251                    status: crate::krabka::barrier::CUT_STATUS_COMPLETE,
252                    topics: super::test_support::sample_cut_topics(),
253                    missing: Vec::new(),
254                    unknown_tagged_fields: UnknownTaggedFields::default(),
255                },
256                BarrierCut {
257                    epoch: 13,
258                    triggered_at: 1_724_500_060_000,
259                    completed_at: 1_724_500_090_000,
260                    status: crate::krabka::barrier::CUT_STATUS_PARTIAL,
261                    topics: super::test_support::sample_cut_topics(),
262                    missing: super::test_support::sample_missing_partitions(),
263                    unknown_tagged_fields: crate::krabka::test_support::sample_tagged_fields(),
264                },
265            ],
266            unknown_tagged_fields: crate::krabka::test_support::sample_tagged_fields(),
267        }
268    }
269}
270
271#[cfg(test)]
272mod tests {
273    use assert2::assert;
274
275    use super::*;
276    use crate::krabka::test_support::{reject_truncated, roundtrip, unsupported_versions};
277
278    #[test]
279    fn default_roundtrips_all_versions() {
280        for version in MIN_VERSION..=MAX_VERSION {
281            roundtrip(&ListBarrierCutsRequest::default(), version);
282            roundtrip(&ListBarrierCutsResponse::default(), version);
283        }
284    }
285
286    #[test]
287    fn populated_roundtrips_all_versions() {
288        for version in MIN_VERSION..=MAX_VERSION {
289            roundtrip(&ListBarrierCutsRequest::populated(), version);
290            roundtrip(&ListBarrierCutsResponse::populated(), version);
291        }
292    }
293
294    #[test]
295    fn unlimited_is_the_max_results_default() {
296        assert!(ListBarrierCutsRequest::default().max_results == -1);
297    }
298
299    #[test]
300    fn truncated_input_is_rejected() {
301        reject_truncated(&ListBarrierCutsResponse::populated(), MIN_VERSION);
302    }
303
304    #[test]
305    fn rejects_unsupported_versions() {
306        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
307            let response = ListBarrierCutsResponse::populated();
308            assert!(matches!(
309                response.encode(&mut Vec::<u8>::new(), version),
310                Err(ProtocolError::UnsupportedVersion {
311                    api_key: API_KEY,
312                    ..
313                })
314            ));
315        }
316    }
317}
krabka_protocol/krabka/barrier/list_barrier_cuts.rs

krabka_protocol/krabka/barrier/
list_barrier_cuts.rs
