krabka_protocol/krabka/barrier/
cut.rs

1//! Cut types that more than one barrier message uses.
2//!
3//! A cut is the set of offsets that one epoch's barrier markers took.
4//! [`BarrierCutTopic`] holds those offsets for one topic, and
5//! [`BarrierMissingPartition`] names a partition that got no marker.
6//! [`TriggerBarrierResponse`](super::trigger_barrier::TriggerBarrierResponse)
7//! returns the cut it made. [`ListBarrierCutsResponse`](super::list_barrier_cuts::ListBarrierCutsResponse)
8//! returns the cuts the coordinator published before.
9
10use bytes::{Buf, BufMut};
11
12use crate::{
13    Decode, Encode, ProtocolError, UnknownTaggedFields,
14    krabka::common::{
15        I32_LEN, I64_LEN, array_len, decode_array, encode_array, read_unknown_tagged_fields,
16        unknown_tagged_fields_len, write_unknown_tagged_fields,
17    },
18    primitives::{
19        fixed::{get_i32, get_i64, put_i32, put_i64},
20        string_bytes::{compact_string_len, get_compact_string_owned, put_compact_string},
21    },
22};
23
24/// Status of a cut that reached every partition of the group.
25pub const CUT_STATUS_COMPLETE: i8 = 0;
26
27/// Status of a cut that missed at least one partition. The `missing` array of
28/// the response names those partitions.
29pub const CUT_STATUS_PARTIAL: i8 = 1;
30
31/// One partition of a cut, and the offset that its barrier marker took.
32///
33/// Records below `offset` are before the cut. Records at `offset` and above are
34/// after it.
35#[derive(Debug, Clone, PartialEq, Eq, Default)]
36pub struct BarrierCutPartition {
37    /// Partition index within the topic.
38    pub partition: i32,
39    /// Offset of the epoch's barrier marker in this partition.
40    pub offset: i64,
41    /// Tagged fields that this build does not know.
42    pub unknown_tagged_fields: UnknownTaggedFields,
43}
44
45impl Encode for BarrierCutPartition {
46    fn encode<B: BufMut>(&self, buf: &mut B, _version: i16) -> Result<(), ProtocolError> {
47        put_i32(buf, self.partition);
48        put_i64(buf, self.offset);
49        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
50        Ok(())
51    }
52
53    fn encoded_len(&self, _version: i16) -> usize {
54        I32_LEN + I64_LEN + unknown_tagged_fields_len(&self.unknown_tagged_fields)
55    }
56}
57
58impl Decode<'_> for BarrierCutPartition {
59    fn decode<B: Buf>(buf: &mut B, _version: i16) -> Result<Self, ProtocolError> {
60        Ok(Self {
61            partition: get_i32(buf)?,
62            offset: get_i64(buf)?,
63            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
64        })
65    }
66}
67
68/// The partitions of one topic that a cut covers.
69#[derive(Debug, Clone, PartialEq, Eq, Default)]
70pub struct BarrierCutTopic {
71    /// Topic name.
72    pub topic: String,
73    /// Cut offset for each partition of the topic that got a marker.
74    pub partitions: Vec<BarrierCutPartition>,
75    /// Tagged fields that this build does not know.
76    pub unknown_tagged_fields: UnknownTaggedFields,
77}
78
79impl Encode for BarrierCutTopic {
80    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
81        put_compact_string(buf, &self.topic);
82        encode_array(buf, &self.partitions, version)?;
83        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
84        Ok(())
85    }
86
87    fn encoded_len(&self, version: i16) -> usize {
88        compact_string_len(&self.topic)
89            + array_len(&self.partitions, version)
90            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
91    }
92}
93
94impl Decode<'_> for BarrierCutTopic {
95    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
96        Ok(Self {
97            topic: get_compact_string_owned(buf)?,
98            partitions: decode_array(buf, version)?,
99            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
100        })
101    }
102}
103
104/// A partition of the frozen target set that got no barrier marker.
105///
106/// A partial cut names every such partition, so one record tells a reader which
107/// partitions to skip for that epoch.
108#[derive(Debug, Clone, PartialEq, Eq, Default)]
109pub struct BarrierMissingPartition {
110    /// Topic name.
111    pub topic: String,
112    /// Partition index within the topic.
113    pub partition: i32,
114    /// Tagged fields that this build does not know.
115    pub unknown_tagged_fields: UnknownTaggedFields,
116}
117
118impl Encode for BarrierMissingPartition {
119    fn encode<B: BufMut>(&self, buf: &mut B, _version: i16) -> Result<(), ProtocolError> {
120        put_compact_string(buf, &self.topic);
121        put_i32(buf, self.partition);
122        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields);
123        Ok(())
124    }
125
126    fn encoded_len(&self, _version: i16) -> usize {
127        compact_string_len(&self.topic)
128            + I32_LEN
129            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
130    }
131}
132
133impl Decode<'_> for BarrierMissingPartition {
134    fn decode<B: Buf>(buf: &mut B, _version: i16) -> Result<Self, ProtocolError> {
135        Ok(Self {
136            topic: get_compact_string_owned(buf)?,
137            partition: get_i32(buf)?,
138            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
139        })
140    }
141}
krabka_protocol/krabka/barrier/cut.rs

krabka_protocol/krabka/barrier/
cut.rs
