krabka_protocol/krabka/barrier/
write_barrier_markers.rs

1//! `WriteBarrierMarkers`, api key 1014.
2//!
3//! The coordinator sends this request to a broker that leads a partition of the
4//! group. That broker appends one barrier marker to each named partition and
5//! returns the offset that each marker took.
6//!
7//! This is inter-broker traffic. A client never sends it. The fan-out follows
8//! the transaction-marker path, and it collects the returned offsets, which the
9//! transaction path does not need.
10
11use bytes::{Buf, BufMut};
12
13#[cfg(test)]
14use crate::krabka::test_support::sample_tagged_fields;
15use crate::{
16    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
17    krabka::common::{
18        I16_LEN, I32_LEN, I64_LEN, array_len, check_version, decode_array, encode_array,
19        read_unknown_tagged_fields, unknown_tagged_fields_len, write_unknown_tagged_fields,
20    },
21    primitives::{
22        fixed::{get_i16, get_i32, get_i64, put_i16, put_i32, put_i64},
23        string_bytes::{compact_string_len, get_compact_string_owned, put_compact_string},
24    },
25};
26
27/// Api key of `WriteBarrierMarkers`.
28pub const API_KEY: i16 = 1014;
29/// Lowest version of `WriteBarrierMarkers` that this build speaks.
30pub const MIN_VERSION: i16 = 0;
31/// Highest version of `WriteBarrierMarkers` that this build speaks.
32pub const MAX_VERSION: i16 = 0;
33/// First version with flexible framing from KIP-482.
34pub const FLEXIBLE_MIN: i16 = 0;
35
36/// Reports whether `version` uses flexible framing from KIP-482.
37///
38/// The request header takes the same framing as the request body, so a caller
39/// that writes the header needs this answer.
40#[inline]
41#[must_use]
42pub fn is_flexible(version: i16) -> bool {
43    version >= FLEXIBLE_MIN
44}
45
46/// One partition for the target broker to mark, and its expected leader epoch.
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct WritableBarrierPartition {
49    /// Partition index within the topic.
50    pub partition: i32,
51    /// Leader epoch that the coordinator read for this partition when it froze
52    /// the target set. The target broker compares this epoch against its own
53    /// leader epoch, and it rejects a mismatch with `FENCED_LEADER_EPOCH`. `-1`
54    /// means the coordinator had no epoch for this partition, and the target
55    /// broker does not fence on it.
56    pub expected_leader_epoch: i32,
57    /// Tagged fields that this build does not know.
58    pub unknown_tagged_fields: UnknownTaggedFields,
59}
60
61impl Default for WritableBarrierPartition {
62    fn default() -> Self {
63        Self {
64            partition: 0,
65            expected_leader_epoch: -1,
66            unknown_tagged_fields: UnknownTaggedFields::default(),
67        }
68    }
69}
70
71impl Encode for WritableBarrierPartition {
72    fn encode<B: BufMut>(&self, buf: &mut B, _version: i16) -> Result<(), ProtocolError> {
73        put_i32(buf, self.partition);
74        put_i32(buf, self.expected_leader_epoch);
75        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
76        Ok(())
77    }
78
79    fn encoded_len(&self, _version: i16) -> usize {
80        I32_LEN + I32_LEN + unknown_tagged_fields_len(&self.unknown_tagged_fields)
81    }
82}
83
84impl Decode<'_> for WritableBarrierPartition {
85    fn decode<B: Buf>(buf: &mut B, _version: i16) -> Result<Self, ProtocolError> {
86        Ok(Self {
87            partition: get_i32(buf)?,
88            expected_leader_epoch: get_i32(buf)?,
89            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
90        })
91    }
92}
93
94/// The partitions of one topic for the target broker to mark.
95#[derive(Debug, Clone, PartialEq, Eq, Default)]
96pub struct WritableBarrierTopic {
97    /// Topic name.
98    pub topic: String,
99    /// One entry for each partition that the target broker leads.
100    pub partitions: Vec<WritableBarrierPartition>,
101    /// Tagged fields that this build does not know.
102    pub unknown_tagged_fields: UnknownTaggedFields,
103}
104
105impl Encode for WritableBarrierTopic {
106    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
107        put_compact_string(buf, &self.topic)?;
108        encode_array(buf, &self.partitions, version)?;
109        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
110        Ok(())
111    }
112
113    fn encoded_len(&self, version: i16) -> usize {
114        compact_string_len(&self.topic)
115            + array_len(&self.partitions, version)
116            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
117    }
118}
119
120impl Decode<'_> for WritableBarrierTopic {
121    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
122        Ok(Self {
123            topic: get_compact_string_owned(buf)?,
124            partitions: decode_array(buf, version)?,
125            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
126        })
127    }
128}
129
130/// Asks a broker to append one barrier marker to each named partition.
131#[derive(Debug, Clone, PartialEq, Eq, Default)]
132pub struct WriteBarrierMarkersRequest {
133    /// Name of the barrier group that the epoch belongs to.
134    pub group: String,
135    /// Epoch that the coordinator assigned to this cut. The marker carries it.
136    pub epoch: i64,
137    /// Milliseconds since the Unix epoch, at the moment the coordinator started
138    /// the injection. The marker carries it.
139    pub triggered_at: i64,
140    /// Target partitions, grouped by topic.
141    pub topics: Vec<WritableBarrierTopic>,
142    /// Tagged fields that this build does not know.
143    pub unknown_tagged_fields: UnknownTaggedFields,
144}
145
146impl Encode for WriteBarrierMarkersRequest {
147    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
148        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
149        put_compact_string(buf, &self.group)?;
150        put_i64(buf, self.epoch);
151        put_i64(buf, self.triggered_at);
152        encode_array(buf, &self.topics, version)?;
153        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
154        Ok(())
155    }
156
157    fn encoded_len(&self, version: i16) -> usize {
158        compact_string_len(&self.group)
159            + I64_LEN
160            + I64_LEN
161            + array_len(&self.topics, version)
162            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
163    }
164}
165
166impl Decode<'_> for WriteBarrierMarkersRequest {
167    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
168        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
169        Ok(Self {
170            group: get_compact_string_owned(buf)?,
171            epoch: get_i64(buf)?,
172            triggered_at: get_i64(buf)?,
173            topics: decode_array(buf, version)?,
174            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
175        })
176    }
177}
178
179impl ProtocolRequest for WriteBarrierMarkersRequest {
180    const API_KEY: i16 = API_KEY;
181    const MIN_VERSION: i16 = MIN_VERSION;
182    const MAX_VERSION: i16 = MAX_VERSION;
183    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
184    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
185
186    type Response = WriteBarrierMarkersResponse;
187}
188
189/// The outcome for one partition that the target broker tried to mark.
190#[derive(Debug, Clone, PartialEq, Eq, Default)]
191pub struct WrittenBarrierPartition {
192    /// Partition index within the topic.
193    pub partition: i32,
194    /// Kafka error code. `0` means the append succeeded and `offset` holds the
195    /// position of the marker.
196    pub error_code: i16,
197    /// Offset that the barrier marker took. The coordinator writes this offset
198    /// into the cut for this partition.
199    pub offset: i64,
200    /// Tagged fields that this build does not know.
201    pub unknown_tagged_fields: UnknownTaggedFields,
202}
203
204impl Encode for WrittenBarrierPartition {
205    fn encode<B: BufMut>(&self, buf: &mut B, _version: i16) -> Result<(), ProtocolError> {
206        put_i32(buf, self.partition);
207        put_i16(buf, self.error_code);
208        put_i64(buf, self.offset);
209        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
210        Ok(())
211    }
212
213    fn encoded_len(&self, _version: i16) -> usize {
214        I32_LEN + I16_LEN + I64_LEN + unknown_tagged_fields_len(&self.unknown_tagged_fields)
215    }
216}
217
218impl Decode<'_> for WrittenBarrierPartition {
219    fn decode<B: Buf>(buf: &mut B, _version: i16) -> Result<Self, ProtocolError> {
220        Ok(Self {
221            partition: get_i32(buf)?,
222            error_code: get_i16(buf)?,
223            offset: get_i64(buf)?,
224            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
225        })
226    }
227}
228
229/// The outcomes for the partitions of one topic.
230#[derive(Debug, Clone, PartialEq, Eq, Default)]
231pub struct WrittenBarrierTopic {
232    /// Topic name.
233    pub topic: String,
234    /// One outcome for each partition that the request named for this topic.
235    pub partitions: Vec<WrittenBarrierPartition>,
236    /// Tagged fields that this build does not know.
237    pub unknown_tagged_fields: UnknownTaggedFields,
238}
239
240impl Encode for WrittenBarrierTopic {
241    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
242        put_compact_string(buf, &self.topic)?;
243        encode_array(buf, &self.partitions, version)?;
244        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
245        Ok(())
246    }
247
248    fn encoded_len(&self, version: i16) -> usize {
249        compact_string_len(&self.topic)
250            + array_len(&self.partitions, version)
251            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
252    }
253}
254
255impl Decode<'_> for WrittenBarrierTopic {
256    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
257        Ok(Self {
258            topic: get_compact_string_owned(buf)?,
259            partitions: decode_array(buf, version)?,
260            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
261        })
262    }
263}
264
265/// The offset that each barrier marker took.
266///
267/// This response carries no throttle time. It is inter-broker traffic, and the
268/// broker applies no client quota to it.
269#[derive(Debug, Clone, PartialEq, Eq, Default)]
270pub struct WriteBarrierMarkersResponse {
271    /// One entry for each topic that the request named, in request order.
272    pub topics: Vec<WrittenBarrierTopic>,
273    /// Tagged fields that this build does not know.
274    pub unknown_tagged_fields: UnknownTaggedFields,
275}
276
277impl Encode for WriteBarrierMarkersResponse {
278    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
279        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
280        encode_array(buf, &self.topics, version)?;
281        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
282        Ok(())
283    }
284
285    fn encoded_len(&self, version: i16) -> usize {
286        array_len(&self.topics, version) + unknown_tagged_fields_len(&self.unknown_tagged_fields)
287    }
288}
289
290impl Decode<'_> for WriteBarrierMarkersResponse {
291    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
292        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
293        Ok(Self {
294            topics: decode_array(buf, version)?,
295            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
296        })
297    }
298}
299
300#[cfg(test)]
301impl WriteBarrierMarkersRequest {
302    /// Builds a request with every field set, for the round-trip tests.
303    fn populated() -> Self {
304        Self {
305            group: "orders".to_owned(),
306            epoch: 41,
307            triggered_at: 1_724_500_000_000,
308            topics: vec![
309                WritableBarrierTopic {
310                    topic: "orders".to_owned(),
311                    partitions: vec![
312                        WritableBarrierPartition {
313                            partition: 0,
314                            expected_leader_epoch: 4,
315                            unknown_tagged_fields: UnknownTaggedFields::default(),
316                        },
317                        WritableBarrierPartition {
318                            partition: 3,
319                            expected_leader_epoch: -1,
320                            unknown_tagged_fields: sample_tagged_fields(),
321                        },
322                        WritableBarrierPartition {
323                            partition: 7,
324                            expected_leader_epoch: 17,
325                            unknown_tagged_fields: UnknownTaggedFields::default(),
326                        },
327                    ],
328                    unknown_tagged_fields: sample_tagged_fields(),
329                },
330                WritableBarrierTopic {
331                    topic: "payments".to_owned(),
332                    partitions: Vec::new(),
333                    unknown_tagged_fields: UnknownTaggedFields::default(),
334                },
335            ],
336            unknown_tagged_fields: sample_tagged_fields(),
337        }
338    }
339}
340
341#[cfg(test)]
342impl WriteBarrierMarkersResponse {
343    /// Builds a response with every field set, for the round-trip tests.
344    fn populated() -> Self {
345        Self {
346            topics: vec![WrittenBarrierTopic {
347                topic: "orders".to_owned(),
348                partitions: vec![
349                    WrittenBarrierPartition {
350                        partition: 0,
351                        error_code: 0,
352                        offset: 9_412,
353                        unknown_tagged_fields: UnknownTaggedFields::default(),
354                    },
355                    WrittenBarrierPartition {
356                        partition: 3,
357                        error_code: 6,
358                        offset: -1,
359                        unknown_tagged_fields: sample_tagged_fields(),
360                    },
361                ],
362                unknown_tagged_fields: UnknownTaggedFields::default(),
363            }],
364            unknown_tagged_fields: sample_tagged_fields(),
365        }
366    }
367}
368
369#[cfg(test)]
370mod tests {
371    use assert2::assert;
372
373    use super::*;
374    use crate::krabka::test_support::{reject_truncated, roundtrip, unsupported_versions};
375
376    #[test]
377    fn default_roundtrips_all_versions() {
378        for version in MIN_VERSION..=MAX_VERSION {
379            roundtrip(&WriteBarrierMarkersRequest::default(), version);
380            roundtrip(&WriteBarrierMarkersResponse::default(), version);
381        }
382    }
383
384    #[test]
385    fn populated_roundtrips_all_versions() {
386        for version in MIN_VERSION..=MAX_VERSION {
387            roundtrip(&WriteBarrierMarkersRequest::populated(), version);
388            roundtrip(&WriteBarrierMarkersResponse::populated(), version);
389        }
390    }
391
392    #[test]
393    fn no_expected_leader_epoch_is_the_partition_default() {
394        assert!(
395            WritableBarrierPartition::default()
396                == WritableBarrierPartition {
397                    partition: 0,
398                    expected_leader_epoch: -1,
399                    unknown_tagged_fields: UnknownTaggedFields::default(),
400                }
401        );
402    }
403
404    #[test]
405    fn truncated_input_is_rejected() {
406        reject_truncated(&WriteBarrierMarkersRequest::populated(), MIN_VERSION);
407        reject_truncated(&WriteBarrierMarkersResponse::populated(), MIN_VERSION);
408    }
409
410    #[test]
411    fn rejects_unsupported_versions() {
412        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
413            let request = WriteBarrierMarkersRequest::populated();
414            assert!(matches!(
415                request.encode(&mut Vec::<u8>::new(), version),
416                Err(ProtocolError::UnsupportedVersion {
417                    api_key: API_KEY,
418                    ..
419                })
420            ));
421        }
422    }
423}
krabka_protocol/krabka/barrier/write_barrier_markers.rs

krabka_protocol/krabka/barrier/
write_barrier_markers.rs
