krabka_client_consumer/share/
types.rs

1//! Public value types for the share-group consumer.
2
3use bytes::Bytes;
4
5use crate::consumer::TimestampType;
6
7/// One record delivered by [`ShareConsumer::poll`](super::ShareConsumer).
8///
9/// Unlike a classic `ConsumerRecord`, a share record carries a
10/// `delivery_count`. The broker increments it each time it re-acquires the
11/// record after a prior delivery was released or after its acquisition lock
12/// expired, per KIP-932. A first delivery has `delivery_count == 1`.
13#[derive(Debug, Clone, PartialEq, Eq)]
14pub struct ShareConsumerRecord {
15    pub topic: String,
16    pub partition: i32,
17    pub offset: i64,
18    /// The create time, or for a `LogAppendTime` batch the broker append
19    /// time (the batch `max_timestamp`).
20    pub timestamp: i64,
21    /// The timestamp type from the attributes of the record batch.
22    pub timestamp_type: TimestampType,
23    pub key: Option<Bytes>,
24    pub value: Option<Bytes>,
25    /// Kafka record headers, kept lossless and in wire order.
26    pub headers: Vec<(String, Option<Bytes>)>,
27    pub delivery_count: i16,
28}
29
30/// How acquired records are acknowledged.
31///
32/// In [`Implicit`](ShareAckMode::Implicit) mode the next `poll()`, and
33/// `close()`, implicitly `Accept` every record that the previous `poll()`
34/// returned. In [`Explicit`](ShareAckMode::Explicit) mode the application must
35/// call `acknowledge()` for each record. The broker releases un-acknowledged
36/// records back to the queue.
37#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
38pub enum ShareAckMode {
39    /// Auto-accept the previous batch on the next poll. The default, matching
40    /// the JVM `share.acknowledgement.mode=implicit`.
41    #[default]
42    Implicit,
43    /// The application acknowledges each record explicitly.
44    Explicit,
45}
46
47/// How a `ShareFetch` applies its maximum-record limit.
48#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
49pub enum ShareAcquireMode {
50    /// Acquire complete record batches, which can exceed the requested record
51    /// limit. This keeps the existing Kafka-compatible behavior.
52    #[default]
53    BatchOptimized,
54    /// Stop acquiring when the requested record limit is reached.
55    RecordLimit,
56}
57
58impl ShareAcquireMode {
59    /// The `i8` wire value carried in `ShareFetch` version 2 and later.
60    pub(crate) fn wire(self) -> i8 {
61        match self {
62            Self::BatchOptimized => 0,
63            Self::RecordLimit => 1,
64        }
65    }
66}
67
68/// The disposition of an acknowledged record, with the KIP-932 wire codes.
69#[derive(Debug, Clone, Copy, PartialEq, Eq)]
70pub enum ShareAckType {
71    /// Successfully processed. Advance the share-partition start offset past it.
72    Accept,
73    /// Return to the queue for re-delivery. This increments `delivery_count`.
74    Release,
75    /// Permanently skip. Do not re-deliver.
76    Reject,
77}
78
79impl ShareAckType {
80    /// The `i8` wire value carried in `ShareFetch` / `ShareAcknowledge`
81    /// acknowledgement batches.
82    pub(crate) fn wire(self) -> i8 {
83        match self {
84            Self::Accept => 1,
85            Self::Release => 2,
86            Self::Reject => 3,
87        }
88    }
89}
90
91#[cfg(test)]
92mod tests {
93
94    use super::*;
95
96    #[test]
97    fn ack_wire_codes() {
98        for (_name, ack, expected) in [
99            ("accept", ShareAckType::Accept, 1),
100            ("release", ShareAckType::Release, 2),
101            ("reject", ShareAckType::Reject, 3),
102        ] {
103            assert2::assert!(ack.wire() == expected);
104        }
105    }
106
107    #[test]
108    fn ack_mode_default_is_implicit() {
109        assert2::assert!(ShareAckMode::default() == ShareAckMode::Implicit);
110    }
111
112    #[test]
113    fn acquire_mode_default_is_batch_optimized() {
114        assert2::assert!(ShareAcquireMode::default() == ShareAcquireMode::BatchOptimized);
115    }
116
117    #[test]
118    fn acquire_mode_wire_codes_match_kafka() {
119        for (mode, expected) in [
120            (ShareAcquireMode::BatchOptimized, 0),
121            (ShareAcquireMode::RecordLimit, 1),
122        ] {
123            assert2::assert!(mode.wire() == expected);
124        }
125    }
126}
krabka_client_consumer/share/types.rs

krabka_client_consumer/share/
types.rs
