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

krabka_protocol/krabka/barrier/
list_barrier_cuts.rs
