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

krabka_protocol/krabka/barrier/
trigger_barrier.rs
