krabka_protocol/krabka/barrier/
trigger_barrier.rs

1//! `TriggerBarrier`, api key 1012.
2//!
3//! The request asks the coordinator to make a cut now. The coordinator picks the
4//! next epoch, freezes the target set, and injects a marker into every partition
5//! of the group. The response carries the offsets that those markers took.
6//!
7//! The coordinator consumes the epoch even when an injection misses a partition,
8//! and it never reuses an epoch. The response then reports
9//! [`CUT_STATUS_PARTIAL`](CUT_STATUS_PARTIAL) and names
10//! the partitions that got no marker.
11
12use bytes::{Buf, BufMut};
13
14#[cfg(test)]
15use crate::krabka::{barrier::CUT_STATUS_PARTIAL, test_support::sample_tagged_fields};
16use crate::{
17    Decode, Encode, ProtocolError, ProtocolRequest, UnknownTaggedFields,
18    krabka::{
19        barrier::cut::{BarrierCutTopic, BarrierMissingPartition},
20        common::{
21            I8_LEN, I16_LEN, I32_LEN, I64_LEN, array_len, check_version, decode_array,
22            encode_array, read_unknown_tagged_fields, unknown_tagged_fields_len,
23            write_unknown_tagged_fields,
24        },
25    },
26    primitives::{
27        fixed::{get_i8, get_i16, get_i32, get_i64, put_i8, put_i16, put_i32, put_i64},
28        string_bytes::{
29            compact_nullable_string_len, compact_string_len, get_compact_nullable_string_owned,
30            get_compact_string_owned, put_compact_nullable_string, put_compact_string,
31        },
32    },
33};
34
35/// Api key of `TriggerBarrier`.
36pub const API_KEY: i16 = 1012;
37/// Lowest version of `TriggerBarrier` that this build speaks.
38pub const MIN_VERSION: i16 = 0;
39/// Highest version of `TriggerBarrier` that this build speaks.
40pub const MAX_VERSION: i16 = 0;
41/// First version with flexible framing from KIP-482.
42pub const FLEXIBLE_MIN: i16 = 0;
43
44/// Reports whether `version` uses flexible framing from KIP-482.
45///
46/// The request header takes the same framing as the request body, so a caller
47/// that writes the header needs this answer.
48#[inline]
49#[must_use]
50pub fn is_flexible(version: i16) -> bool {
51    version >= FLEXIBLE_MIN
52}
53
54/// Asks the coordinator for a cut of one barrier group.
55#[derive(Debug, Clone, PartialEq, Eq, Default)]
56pub struct TriggerBarrierRequest {
57    /// Name of the barrier group to cut.
58    pub group: String,
59    /// Milliseconds that the coordinator waits for every partition to take a
60    /// marker. After the deadline it publishes a partial cut.
61    pub timeout_ms: i32,
62    /// Tagged fields that this build does not know.
63    pub unknown_tagged_fields: UnknownTaggedFields,
64}
65
66impl Encode for TriggerBarrierRequest {
67    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
68        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
69        put_compact_string(buf, &self.group)?;
70        put_i32(buf, self.timeout_ms);
71        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
72        Ok(())
73    }
74
75    fn encoded_len(&self, _version: i16) -> usize {
76        compact_string_len(&self.group)
77            + I32_LEN
78            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
79    }
80}
81
82impl Decode<'_> for TriggerBarrierRequest {
83    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
84        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
85        Ok(Self {
86            group: get_compact_string_owned(buf)?,
87            timeout_ms: get_i32(buf)?,
88            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
89        })
90    }
91}
92
93impl ProtocolRequest for TriggerBarrierRequest {
94    const API_KEY: i16 = API_KEY;
95    const MIN_VERSION: i16 = MIN_VERSION;
96    const MAX_VERSION: i16 = MAX_VERSION;
97    const LATEST_STABLE_VERSION: i16 = MAX_VERSION;
98    const FLEXIBLE_MIN: i16 = FLEXIBLE_MIN;
99
100    type Response = TriggerBarrierResponse;
101}
102
103/// The cut that a [`TriggerBarrierRequest`] made.
104#[derive(Debug, Clone, PartialEq, Eq, Default)]
105pub struct TriggerBarrierResponse {
106    /// Milliseconds that the broker held the request back for a quota.
107    pub throttle_time_ms: i32,
108    /// Kafka error code. `0` means the coordinator ran the injection, and the
109    /// cut fields below hold its result.
110    pub error_code: i16,
111    /// Text that explains a non-zero `error_code`. `None` when the coordinator
112    /// has nothing to add to the code.
113    pub error_message: Option<String>,
114    /// Epoch that the coordinator assigned to this cut. The coordinator never
115    /// reuses an epoch, and a partial cut consumes one as a complete cut does.
116    pub epoch: i64,
117    /// [`CUT_STATUS_COMPLETE`](crate::krabka::barrier::CUT_STATUS_COMPLETE)
118    /// when every partition took a marker, and
119    /// [`CUT_STATUS_PARTIAL`](CUT_STATUS_PARTIAL) when
120    /// at least one partition did not.
121    pub status: i8,
122    /// Cut offsets, grouped by topic.
123    pub topics: Vec<BarrierCutTopic>,
124    /// Partitions of the frozen target set that took no marker. This array is
125    /// empty for a complete cut.
126    pub missing: Vec<BarrierMissingPartition>,
127    /// Tagged fields that this build does not know.
128    pub unknown_tagged_fields: UnknownTaggedFields,
129}
130
131impl Encode for TriggerBarrierResponse {
132    fn encode<B: BufMut>(&self, buf: &mut B, version: i16) -> Result<(), ProtocolError> {
133        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
134        put_i32(buf, self.throttle_time_ms);
135        put_i16(buf, self.error_code);
136        put_compact_nullable_string(buf, self.error_message.as_deref())?;
137        put_i64(buf, self.epoch);
138        put_i8(buf, self.status);
139        encode_array(buf, &self.topics, version)?;
140        encode_array(buf, &self.missing, version)?;
141        write_unknown_tagged_fields(buf, &self.unknown_tagged_fields)?;
142        Ok(())
143    }
144
145    fn encoded_len(&self, version: i16) -> usize {
146        I32_LEN
147            + I16_LEN
148            + compact_nullable_string_len(self.error_message.as_deref())
149            + I64_LEN
150            + I8_LEN
151            + array_len(&self.topics, version)
152            + array_len(&self.missing, version)
153            + unknown_tagged_fields_len(&self.unknown_tagged_fields)
154    }
155}
156
157impl Decode<'_> for TriggerBarrierResponse {
158    fn decode<B: Buf>(buf: &mut B, version: i16) -> Result<Self, ProtocolError> {
159        check_version(API_KEY, version, MIN_VERSION, MAX_VERSION)?;
160        Ok(Self {
161            throttle_time_ms: get_i32(buf)?,
162            error_code: get_i16(buf)?,
163            error_message: get_compact_nullable_string_owned(buf)?,
164            epoch: get_i64(buf)?,
165            status: get_i8(buf)?,
166            topics: decode_array(buf, version)?,
167            missing: decode_array(buf, version)?,
168            unknown_tagged_fields: read_unknown_tagged_fields(buf)?,
169        })
170    }
171}
172
173#[cfg(test)]
174impl TriggerBarrierRequest {
175    /// Builds a request with every field set, for the round-trip tests.
176    fn populated() -> Self {
177        Self {
178            group: "orders".to_owned(),
179            timeout_ms: 30_000,
180            unknown_tagged_fields: sample_tagged_fields(),
181        }
182    }
183}
184
185#[cfg(test)]
186impl TriggerBarrierResponse {
187    /// Builds a partial-cut response with every field set, for the round-trip
188    /// tests.
189    fn populated() -> Self {
190        Self {
191            throttle_time_ms: 3,
192            error_code: 0,
193            error_message: Some("one partition had no leader".to_owned()),
194            epoch: 41,
195            status: CUT_STATUS_PARTIAL,
196            topics: super::test_support::sample_cut_topics(),
197            missing: super::test_support::sample_missing_partitions(),
198            unknown_tagged_fields: sample_tagged_fields(),
199        }
200    }
201}
202
203#[cfg(test)]
204mod tests {
205    use assert2::assert;
206
207    use super::*;
208    use crate::krabka::{
209        barrier::{CUT_STATUS_COMPLETE, CUT_STATUS_PARTIAL, test_support::sample_cut_topics},
210        test_support::{roundtrip, unsupported_versions},
211    };
212
213    #[test]
214    fn default_roundtrips_all_versions() {
215        for version in MIN_VERSION..=MAX_VERSION {
216            roundtrip(&TriggerBarrierRequest::default(), version);
217            roundtrip(&TriggerBarrierResponse::default(), version);
218        }
219    }
220
221    #[test]
222    fn populated_roundtrips_all_versions() {
223        for version in MIN_VERSION..=MAX_VERSION {
224            roundtrip(&TriggerBarrierRequest::populated(), version);
225            roundtrip(&TriggerBarrierResponse::populated(), version);
226        }
227    }
228
229    #[test]
230    fn complete_cut_carries_no_missing_partitions() {
231        let response = TriggerBarrierResponse {
232            epoch: 7,
233            status: CUT_STATUS_COMPLETE,
234            topics: sample_cut_topics(),
235            ..TriggerBarrierResponse::default()
236        };
237        roundtrip(&response, MIN_VERSION);
238        assert!(response.missing.is_empty());
239    }
240
241    #[test]
242    fn status_codes_are_distinct() {
243        assert!(CUT_STATUS_COMPLETE != CUT_STATUS_PARTIAL);
244    }
245
246    #[test]
247    fn rejects_unsupported_versions() {
248        for version in unsupported_versions(MIN_VERSION, MAX_VERSION) {
249            let request = TriggerBarrierRequest::populated();
250            assert!(matches!(
251                request.encode(&mut Vec::<u8>::new(), version),
252                Err(ProtocolError::UnsupportedVersion {
253                    api_key: API_KEY,
254                    ..
255                })
256            ));
257        }
258    }
259}
krabka_protocol/krabka/barrier/trigger_barrier.rs

krabka_protocol/krabka/barrier/
trigger_barrier.rs
