kafka_client/consumer/config.rs
1//! Consumer configuration types.
2
3use std::time::Duration;
4
5// ---------------------------------------------------------------------------
6// Configuration & enums
7// ---------------------------------------------------------------------------
8
9/// Kafka consumer configuration.
10///
11/// # Construction
12///
13/// Use `ConsumerConfig::new()` for a direct-mode (non-group) consumer, or
14/// chain `.with_group_id(...)` to enable consumer group coordination:
15///
16/// ```
17/// use kafka_client::{ConsumerConfig, AutoOffsetReset};
18/// use std::time::Duration;
19///
20/// // Direct mode — no consumer group
21/// let config = ConsumerConfig::new();
22///
23/// // Group mode — with consumer group coordination
24/// let config = ConsumerConfig::new()
25/// .with_group_id("my-group")
26/// .with_earliest()
27/// .with_max_poll_records(1000);
28/// ```
29#[derive(Debug, Clone)]
30pub struct ConsumerConfig {
31 /// Consumer group ID. `None` means direct (non-group) consumer mode.
32 /// `Some(...)` enables consumer group coordination (join/sync/heartbeat/commit).
33 pub group_id: Option<String>,
34 /// If true, offsets are committed automatically at `auto_commit_interval`.
35 pub auto_commit: bool,
36 /// Interval between automatic offset commits.
37 pub auto_commit_interval: Duration,
38 /// What to do when there is no committed offset or the offset is out of range.
39 pub auto_offset_reset: AutoOffsetReset,
40 /// Minimum bytes of data to return in a single Fetch response.
41 /// The broker waits until at least this much data is available (up to `max_wait`).
42 pub min_bytes: i32,
43 /// Maximum bytes of data to return in a single Fetch response.
44 pub max_bytes: i32,
45 /// Maximum bytes per partition in a single Fetch response.
46 pub partition_max_bytes: i32,
47 /// Maximum time the broker will wait to satisfy `min_bytes`.
48 pub max_wait: Duration,
49 /// Maximum number of records returned by a single `poll()` call.
50 pub max_poll_records: usize,
51 /// Group session timeout. If the coordinator receives no heartbeat within this
52 /// window, the member is considered dead and its partitions are reassigned.
53 pub session_timeout: Duration,
54 /// Maximum time a rebalance is allowed to complete.
55 pub rebalance_timeout: Duration,
56 /// Interval between heartbeats sent to the group coordinator.
57 pub heartbeat_interval: Duration,
58 /// Partition assignment strategy used during group rebalance.
59 pub partition_assignment_strategy: PartitionAssignmentStrategy,
60 /// If true, committed offsets track the last record **delivered to `poll()`**
61 /// rather than the last record **fetched from the broker**.
62 ///
63 /// This provides **at-least-once** delivery semantics: on restart,
64 /// the consumer resumes from the last delivered position, which may
65 /// cause some messages to be re-processed.
66 ///
67 /// Default: `false` (at-most-once, same as before).
68 pub enable_at_least_once: bool,
69 /// Maximum number of retries for retriable offset commit failures.
70 pub retries: i32,
71 /// Initial backoff delay between retries (doubles each attempt).
72 pub retry_backoff: Duration,
73}
74
75impl ConsumerConfig {
76 /// Create a new consumer config with sensible defaults in **direct mode**
77 /// (no consumer group coordination).
78 ///
79 /// Call [`with_group_id`](Self::with_group_id) to enable consumer group mode.
80 pub fn new() -> Self {
81 Self {
82 group_id: None,
83 auto_commit: true,
84 auto_commit_interval: Duration::from_secs(5),
85 auto_offset_reset: AutoOffsetReset::Latest,
86 min_bytes: 1,
87 max_bytes: 50 * 1024 * 1024,
88 partition_max_bytes: 1024 * 1024,
89 max_wait: Duration::from_millis(500),
90 max_poll_records: 500,
91 session_timeout: Duration::from_secs(10),
92 rebalance_timeout: Duration::from_secs(30),
93 heartbeat_interval: Duration::from_secs(3),
94 partition_assignment_strategy: PartitionAssignmentStrategy::Range,
95 enable_at_least_once: false,
96 retries: 3,
97 retry_backoff: Duration::from_millis(200),
98 }
99 }
100
101 // ------------------------------------------------------------------
102 // Builder methods
103 // ------------------------------------------------------------------
104
105 /// Set the consumer group ID, enabling group coordination.
106 pub fn with_group_id(mut self, group_id: impl Into<String>) -> Self {
107 self.group_id = Some(group_id.into());
108 self
109 }
110
111 /// Enable or disable automatic offset commit.
112 pub fn with_auto_commit(mut self, auto_commit: bool) -> Self {
113 self.auto_commit = auto_commit;
114 self
115 }
116
117 /// Set the interval between automatic offset commits.
118 pub fn with_auto_commit_interval(mut self, interval: Duration) -> Self {
119 self.auto_commit_interval = interval;
120 self
121 }
122
123 /// Set what to do when there is no committed offset or the offset is out of range.
124 pub fn with_auto_offset_reset(mut self, reset: AutoOffsetReset) -> Self {
125 self.auto_offset_reset = reset;
126 self
127 }
128
129 /// Convenience: reset offset to the earliest available (beginning of partition).
130 pub fn with_earliest(mut self) -> Self {
131 self.auto_offset_reset = AutoOffsetReset::Earliest;
132 self
133 }
134
135 /// Convenience: reset offset to the latest available (end of partition).
136 pub fn with_latest(mut self) -> Self {
137 self.auto_offset_reset = AutoOffsetReset::Latest;
138 self
139 }
140
141 /// Set minimum bytes of data to return in a single Fetch response.
142 pub fn with_min_bytes(mut self, min_bytes: i32) -> Self {
143 self.min_bytes = min_bytes;
144 self
145 }
146
147 /// Set maximum bytes of data to return in a single Fetch response.
148 pub fn with_max_bytes(mut self, max_bytes: i32) -> Self {
149 self.max_bytes = max_bytes;
150 self
151 }
152
153 /// Set maximum bytes per partition in a single Fetch response.
154 pub fn with_partition_max_bytes(mut self, partition_max_bytes: i32) -> Self {
155 self.partition_max_bytes = partition_max_bytes;
156 self
157 }
158
159 /// Set the maximum time the broker will wait to satisfy `min_bytes`.
160 pub fn with_max_wait(mut self, max_wait: Duration) -> Self {
161 self.max_wait = max_wait;
162 self
163 }
164
165 /// Set the maximum number of records returned by a single `poll()` call.
166 pub fn with_max_poll_records(mut self, max_poll_records: usize) -> Self {
167 self.max_poll_records = max_poll_records;
168 self
169 }
170
171 /// Set group session timeout.
172 pub fn with_session_timeout(mut self, timeout: Duration) -> Self {
173 self.session_timeout = timeout;
174 self
175 }
176
177 /// Set maximum time a rebalance is allowed to complete.
178 pub fn with_rebalance_timeout(mut self, timeout: Duration) -> Self {
179 self.rebalance_timeout = timeout;
180 self
181 }
182
183 /// Set the interval between heartbeats sent to the group coordinator.
184 pub fn with_heartbeat_interval(mut self, interval: Duration) -> Self {
185 self.heartbeat_interval = interval;
186 self
187 }
188
189 /// Set the partition assignment strategy.
190 pub fn with_assignment_strategy(mut self, strategy: PartitionAssignmentStrategy) -> Self {
191 self.partition_assignment_strategy = strategy;
192 self
193 }
194
195 /// Enable at-least-once delivery semantics.
196 ///
197 /// When enabled, committed offsets track the last record delivered to
198 /// `poll()` rather than the last record fetched from the broker.
199 /// On restart, the consumer resumes from the last delivered position.
200 pub fn with_at_least_once(mut self) -> Self {
201 self.enable_at_least_once = true;
202 self
203 }
204
205 /// Set the maximum number of retries for retriable offset commit failures.
206 pub fn with_retries(mut self, retries: i32) -> Self {
207 self.retries = retries;
208 self
209 }
210
211 /// Set the initial backoff delay between retries (doubles each attempt).
212 pub fn with_retry_backoff(mut self, backoff: Duration) -> Self {
213 self.retry_backoff = backoff;
214 self
215 }
216}
217
218impl Default for ConsumerConfig {
219 fn default() -> Self {
220 Self::new()
221 }
222}
223
224#[derive(Debug, Clone, Copy, PartialEq, Eq)]
225pub enum AutoOffsetReset {
226 Earliest,
227 Latest,
228 None,
229}
230
231#[derive(Debug, Clone, Copy, PartialEq, Eq)]
232pub enum PartitionAssignmentStrategy {
233 /// Range assignment — each topic's partitions are divided contiguously
234 /// among members. May cause uneven distribution when topics have
235 /// different partition counts.
236 Range,
237 /// Round-robin assignment — partitions are distributed evenly across
238 /// members in cyclic order.
239 RoundRobin,
240 /// Cooperative sticky assignment (simplified).
241 ///
242 /// This is a **simplified** implementation that distributes partitions
243 /// using a round-robin-like algorithm. A full CooperativeSticky
244 /// implementation would need to track each member's previous assignment
245 /// across rebalances and minimize partition movement.
246 CooperativeSticky,
247}