ruststream_rdkafka/topic.rs
1//! The subscription descriptor: one topic consumed through one consumer group.
2
3use ruststream::SubscriptionSource;
4
5use crate::broker::ConnectedKafkaBroker;
6use crate::error::KafkaError;
7use crate::retry::Retry;
8use crate::subscriber::KafkaSubscriber;
9
10/// Where a consumer group starts reading when it has no valid committed offset.
11///
12/// Kafka resumes from the group's committed position when a valid one exists; this choice (it
13/// maps to librdkafka's `auto.offset.reset`) applies when there is none - the group has never
14/// committed the partition, or the committed offset was deleted by retention / is out of
15/// range. The second case is why it matters for long-idle groups: with the librdkafka default
16/// (latest) an expired group skips to the end instead of reprocessing.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
18#[non_exhaustive]
19pub enum StartOffset {
20 /// Leave the choice to librdkafka (its default resets to the latest offset).
21 #[default]
22 Committed,
23 /// Start from the earliest retained offset.
24 Earliest,
25 /// Start from the latest offset (only messages published after the group formed).
26 Latest,
27}
28
29/// The partition assignment strategy for the consumer group (librdkafka's
30/// `partition.assignment.strategy`).
31///
32/// These are librdkafka's built-in strategies; the client offers no API for a custom group
33/// assignor (the rebalance callback only observes assignments). Cooperative and eager
34/// strategies cannot mix within one group - librdkafka rejects the join, and the error
35/// surfaces on the subscriber stream.
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37#[non_exhaustive]
38pub enum Assignment {
39 /// Co-partitioned ranges per topic (the Kafka default family).
40 Range,
41 /// Round-robin across all subscribed topics.
42 RoundRobin,
43 /// Incremental cooperative rebalancing: unaffected partitions keep flowing during a
44 /// rebalance instead of stopping the world.
45 CooperativeSticky,
46}
47
48impl Assignment {
49 pub(crate) fn as_config_value(self) -> &'static str {
50 match self {
51 Self::Range => "range",
52 Self::RoundRobin => "roundrobin",
53 Self::CooperativeSticky => "cooperative-sticky",
54 }
55 }
56}
57
58/// What drives keyed worker lanes (`workers(n, by_key)`) for this subscription.
59///
60/// The runtime lanes deliveries by [`IncomingMessage::partition_key`]
61/// (deliveries sharing a lane key process in order on one lane); this choice picks what that
62/// key is for Kafka.
63///
64/// [`IncomingMessage::partition_key`]: ruststream::IncomingMessage::partition_key
65#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
66#[non_exhaustive]
67pub enum LaneKey {
68 /// The source partition (the default): lanes mirror Kafka's own ordering unit, so
69 /// everything a partition delivers (keyless included) processes in order on one lane.
70 #[default]
71 Partition,
72 /// The native record key: per-key ordering, finer than a partition, so messages of one
73 /// partition may process concurrently when their keys differ. Keyless deliveries carry no
74 /// lane key and rotate across lanes, losing their partition order.
75 RecordKey,
76}
77
78/// How processed deliveries are committed back to the consumer group.
79#[derive(Debug, Clone, PartialEq, Eq, Default)]
80#[non_exhaustive]
81pub enum Commit {
82 /// librdkafka auto-commit, the librdkafka default: positions are stored as messages are
83 /// handed to the application and committed every `auto.commit.interval.ms`. `ack` and
84 /// `nack` are advisory no-ops; a crash can lose the tail of processed-but-uncommitted work
85 /// or skip unprocessed deliveries that were already stored.
86 #[default]
87 Auto,
88 /// Per-message acknowledgement: `enable.auto.offset.store` is switched off and an `ack`
89 /// advances the stored position to just below the lowest still-unsettled delivery (or to
90 /// the highest delivered offset once none are outstanding). At-least-once stays precise
91 /// with concurrent handler lanes, and offset gaps the consumer never receives (transaction
92 /// markers, compacted-away records) cannot block the position. Auto-commit still flushes
93 /// the stored position in the background and once more when the consumer closes.
94 Tracked,
95 /// Exactly-once: the consumer never commits its own offsets - the
96 /// [`EosPipeline`](crate::EosPipeline) whose transactional id matches this name commits
97 /// them through the producer transaction (`send_offsets_to_transaction`), so source
98 /// positions move atomically with the records the handlers publish. `enable.auto.commit`
99 /// and `enable.auto.offset.store` are switched off; `ack` advances the shared watermark
100 /// exactly like [`Tracked`](Self::Tracked), and the pipeline picks the watermark up at its
101 /// next window commit.
102 Transactional(String),
103}
104
105/// A subscription to one Kafka topic through one consumer group.
106///
107/// Everything except the topic name is optional; unset options fall back to the librdkafka
108/// defaults (this crate does not impose its own). The group can also come from
109/// [`KafkaBroker::default_group`](crate::KafkaBroker::default_group); a subscription that ends up with no group at all is a
110/// startup error, because Kafka cannot subscribe without one.
111///
112/// # Examples
113///
114/// ```
115/// use ruststream_rdkafka::{Assignment, Commit, KafkaTopic, StartOffset};
116///
117/// let topic = KafkaTopic::new("orders")
118/// .group("orders-svc")
119/// .start(StartOffset::Earliest)
120/// .commit(Commit::Tracked)
121/// .assignment(Assignment::CooperativeSticky)
122/// .config("fetch.min.bytes", "1024");
123/// assert_eq!(topic.topic(), "orders");
124/// ```
125#[derive(Debug, Clone)]
126pub struct KafkaTopic {
127 /// The subscribed names; librdkafka treats entries starting with `^` as regex patterns.
128 topics: Vec<String>,
129 /// The handler-metadata name: the subscribed names joined with `,`.
130 name: String,
131 /// Set by [`pattern`](Self::pattern), which promises a `^`-anchored regex.
132 requires_pattern: bool,
133 group: Option<String>,
134 start: StartOffset,
135 commit: Commit,
136 assignment: Option<Assignment>,
137 lane_key: LaneKey,
138 partitions: Vec<i32>,
139 retry: Option<Retry>,
140 max_deliveries: Option<u32>,
141 dead_letter: Option<String>,
142 config: Vec<(String, String)>,
143}
144
145impl KafkaTopic {
146 fn with_first(first: String, requires_pattern: bool) -> Self {
147 Self {
148 name: first.clone(),
149 topics: vec![first],
150 requires_pattern,
151 group: None,
152 start: StartOffset::default(),
153 commit: Commit::default(),
154 assignment: None,
155 lane_key: LaneKey::default(),
156 partitions: Vec::new(),
157 retry: None,
158 max_deliveries: None,
159 dead_letter: None,
160 config: Vec::new(),
161 }
162 }
163
164 /// Describes a subscription to `topic` with librdkafka defaults for everything else.
165 #[must_use]
166 pub fn new(topic: impl Into<String>) -> Self {
167 Self::with_first(topic.into(), false)
168 }
169
170 /// Describes a subscription to every existing topic matching `pattern`.
171 ///
172 /// The pattern is a librdkafka topic regex and must start with `^` (that anchor is how
173 /// librdkafka distinguishes a pattern from a literal name); subscribing fails with a clear
174 /// error otherwise. Topics created after the group formed are picked up on the next
175 /// metadata refresh.
176 ///
177 /// # Examples
178 ///
179 /// ```
180 /// use ruststream_rdkafka::KafkaTopic;
181 ///
182 /// let orders = KafkaTopic::pattern("^orders\\..*").group("orders-svc");
183 /// assert_eq!(orders.topic(), "^orders\\..*");
184 /// ```
185 #[must_use]
186 pub fn pattern(pattern: impl Into<String>) -> Self {
187 Self::with_first(pattern.into(), true)
188 }
189
190 /// Adds another topic to the same subscription: one consumer, one group, several topics.
191 ///
192 /// All matched topics share the handler (and therefore its payload type). Entries starting
193 /// with `^` are librdkafka regex patterns, exactly as in [`pattern`](Self::pattern).
194 ///
195 /// # Examples
196 ///
197 /// ```
198 /// use ruststream_rdkafka::KafkaTopic;
199 ///
200 /// let both = KafkaTopic::new("orders").and_topic("cancellations");
201 /// assert_eq!(both.topic(), "orders,cancellations");
202 /// ```
203 #[must_use]
204 pub fn and_topic(mut self, topic: impl Into<String>) -> Self {
205 let topic = topic.into();
206 self.name.push(',');
207 self.name.push_str(&topic);
208 self.topics.push(topic);
209 self
210 }
211
212 /// The consumer group for this subscription, overriding
213 /// [`KafkaBroker::default_group`](crate::KafkaBroker::default_group).
214 #[must_use]
215 pub fn group(mut self, group: impl Into<String>) -> Self {
216 self.group = Some(group.into());
217 self
218 }
219
220 /// Where the group starts when it has no committed offset (see [`StartOffset`]).
221 #[must_use]
222 pub fn start(mut self, start: StartOffset) -> Self {
223 self.start = start;
224 self
225 }
226
227 /// How processed deliveries are committed (see [`Commit`]).
228 #[must_use]
229 pub fn commit(mut self, commit: Commit) -> Self {
230 self.commit = commit;
231 self
232 }
233
234 /// The partition assignment strategy (see [`Assignment`]); unset means the librdkafka
235 /// default (`range,roundrobin`).
236 #[must_use]
237 pub fn assignment(mut self, assignment: Assignment) -> Self {
238 self.assignment = Some(assignment);
239 self
240 }
241
242 /// What drives keyed worker lanes for this subscription (see [`LaneKey`]); the default
243 /// lanes by the source partition, Kafka's native ordering unit.
244 ///
245 /// # Examples
246 ///
247 /// ```
248 /// use ruststream_rdkafka::{KafkaTopic, LaneKey};
249 ///
250 /// // Opt into finer, per-record-key lanes: one tenant never processes concurrently,
251 /// // different tenants in one partition do.
252 /// let topic = KafkaTopic::new("orders")
253 /// .group("orders-svc")
254 /// .lane_key(LaneKey::RecordKey);
255 /// # let _ = topic;
256 /// ```
257 #[must_use]
258 pub fn lane_key(mut self, lane_key: LaneKey) -> Self {
259 self.lane_key = lane_key;
260 self
261 }
262
263 /// Switches the subscription to manual partition assignment: the consumer `assign()`s
264 /// exactly these partitions of the topic - no group membership, no rebalancing.
265 ///
266 /// Deliveries start per [`start`](Self::start); with a group also named the consumer
267 /// commits into it without joining it (so `StartOffset::Committed` resumes from the
268 /// group's positions), and without one commits are off and the start offset must be
269 /// explicit. Does not combine with [`and_topic`](Self::and_topic) /
270 /// [`pattern`](Self::pattern) (manual assignment names exact partitions of one topic) or
271 /// with `Commit::Transactional`.
272 ///
273 /// # Examples
274 ///
275 /// ```
276 /// use ruststream_rdkafka::{KafkaTopic, StartOffset};
277 ///
278 /// // An inspection reader pinned to partition 0, no group side effects.
279 /// let topic = KafkaTopic::new("orders")
280 /// .partitions([0])
281 /// .start(StartOffset::Earliest);
282 /// # let _ = topic;
283 /// ```
284 #[must_use]
285 pub fn partitions(mut self, partitions: impl IntoIterator<Item = i32>) -> Self {
286 self.partitions = partitions.into_iter().collect();
287 self
288 }
289
290 /// What `nack(true)` does on this subscription (see [`Retry`]); unset keeps Kafka's native
291 /// behavior - the offset stays unsettled and redelivers on the next fetch of the partition.
292 ///
293 /// # Examples
294 ///
295 /// ```
296 /// use ruststream_rdkafka::{KafkaTopic, Retry};
297 ///
298 /// let topic = KafkaTopic::new("orders")
299 /// .group("orders-svc")
300 /// .retry(Retry::Topic("orders.retry".into()))
301 /// .max_deliveries(5)
302 /// .dead_letter("orders.dlq");
303 /// # let _ = topic;
304 /// ```
305 #[must_use]
306 pub fn retry(mut self, retry: Retry) -> Self {
307 self.retry = Some(retry);
308 self
309 }
310
311 /// The poison cap: how many times a message may be delivered before `nack(true)` takes the
312 /// drop path instead of retrying (the original delivery counts as one). Enforced by the
313 /// [`retry`](Self::retry) policy - through [`RETRY_COUNT_HEADER`](crate::RETRY_COUNT_HEADER)
314 /// for [`Retry::Topic`], and through an in-session counter for [`Retry::SeekBack`].
315 ///
316 /// The cap needs a [`retry`](Self::retry) policy or a [`dead_letter`](Self::dead_letter)
317 /// topic to count against; on its own it could never apply, so subscribing rejects that
318 /// combination with [`KafkaError::InvalidOptions`](crate::KafkaError::InvalidOptions).
319 #[must_use]
320 pub fn max_deliveries(mut self, max_deliveries: u32) -> Self {
321 self.max_deliveries = Some(max_deliveries);
322 self
323 }
324
325 /// The dead-letter topic for the drop path: `nack(false)` and an exhausted retry republish
326 /// the message there (stamped with the `kafka-dlq-source-*` headers), then settle. Without
327 /// it the drop path just settles.
328 #[must_use]
329 pub fn dead_letter(mut self, topic: impl Into<String>) -> Self {
330 self.dead_letter = Some(topic.into());
331 self
332 }
333
334 /// Raw librdkafka consumer property passthrough for anything not surfaced as a typed
335 /// option, applied last (it wins over the typed options and the broker-wide config).
336 #[must_use]
337 pub fn config(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
338 self.config.push((key.into(), value.into()));
339 self
340 }
341
342 /// The subscribed name(s), joined with `,` when there are several (also the handler
343 /// metadata name); a pattern subscription returns the pattern.
344 #[must_use]
345 pub fn topic(&self) -> &str {
346 &self.name
347 }
348
349 pub(crate) fn subscribed_topics(&self) -> &[String] {
350 &self.topics
351 }
352
353 pub(crate) fn validate(&self) -> Result<(), KafkaError> {
354 if self.requires_pattern && !self.topics[0].starts_with('^') {
355 return Err(KafkaError::InvalidOptions(format!(
356 "pattern {:?} must start with '^' (librdkafka's anchor for topic regexes); \
357 without it the name would be subscribed literally",
358 self.topics[0],
359 )));
360 }
361 if !self.partitions.is_empty() && (self.topics.len() > 1 || self.requires_pattern) {
362 return Err(KafkaError::InvalidOptions(
363 "manual partition assignment names exact partitions of one topic; it does \
364 not combine with `and_topic` or `pattern`"
365 .to_owned(),
366 ));
367 }
368 // Without a retry policy or a dead-letter topic no path ever reads the cap: nack(true)
369 // is Kafka's native re-consumption, which this crate cannot count. Accepting the cap
370 // would silently disarm a poison-message guard the user believes is in place.
371 if self.max_deliveries.is_some() && self.retry.is_none() && self.dead_letter.is_none() {
372 return Err(KafkaError::InvalidOptions(
373 "max_deliveries needs a retry(..) policy or a dead_letter(..) topic to count \
374 against; alone it would never apply"
375 .to_owned(),
376 ));
377 }
378 Ok(())
379 }
380
381 pub(crate) fn group_or<'a>(&'a self, fallback: Option<&'a str>) -> Option<&'a str> {
382 self.group.as_deref().or(fallback)
383 }
384
385 pub(crate) fn start_offset(&self) -> StartOffset {
386 self.start
387 }
388
389 pub(crate) fn commit_mode(&self) -> &Commit {
390 &self.commit
391 }
392
393 pub(crate) fn assignment_strategy(&self) -> Option<Assignment> {
394 self.assignment
395 }
396
397 pub(crate) fn lane_key_choice(&self) -> LaneKey {
398 self.lane_key
399 }
400
401 pub(crate) fn retry_policy(&self) -> Option<&Retry> {
402 self.retry.as_ref()
403 }
404
405 pub(crate) fn max_deliveries_cap(&self) -> Option<u32> {
406 self.max_deliveries
407 }
408
409 pub(crate) fn dead_letter_topic(&self) -> Option<&str> {
410 self.dead_letter.as_deref()
411 }
412
413 pub(crate) fn assigned_partitions(&self) -> &[i32] {
414 &self.partitions
415 }
416
417 pub(crate) fn config_entries(&self) -> &[(String, String)] {
418 &self.config
419 }
420}
421
422impl SubscriptionSource<ConnectedKafkaBroker> for KafkaTopic {
423 type Subscriber = KafkaSubscriber;
424
425 fn name(&self) -> &str {
426 &self.name
427 }
428
429 async fn subscribe(
430 self,
431 connected: &ConnectedKafkaBroker,
432 ) -> Result<Self::Subscriber, KafkaError> {
433 connected.subscribe_with(self).await
434 }
435}
436
437#[cfg(feature = "testing")]
438impl SubscriptionSource<crate::testing::ConnectedKafkaTestBroker> for KafkaTopic {
439 type Subscriber = crate::testing::KafkaTestSubscriber;
440
441 fn name(&self) -> &str {
442 &self.name
443 }
444
445 async fn subscribe(
446 self,
447 broker: &crate::testing::ConnectedKafkaTestBroker,
448 ) -> Result<Self::Subscriber, KafkaError> {
449 if !self.partitions.is_empty() {
450 return Err(KafkaError::InvalidOptions(
451 "the in-process test broker does not simulate partitions; manual partition \
452 assignment needs a real cluster"
453 .to_owned(),
454 ));
455 }
456 broker.subscribe_topics(&self.topics).await
457 }
458}
459
460#[cfg(test)]
461mod tests {
462 use super::*;
463
464 #[test]
465 fn max_deliveries_alone_is_rejected() {
466 let err = KafkaTopic::new("orders")
467 .max_deliveries(3)
468 .validate()
469 .expect_err("a cap with nothing to count against must be rejected");
470 assert!(matches!(err, KafkaError::InvalidOptions(_)));
471 }
472
473 #[test]
474 fn max_deliveries_with_retry_or_dead_letter_passes() {
475 KafkaTopic::new("orders")
476 .retry(Retry::SeekBack)
477 .max_deliveries(3)
478 .validate()
479 .expect("a cap with a retry policy is valid");
480 KafkaTopic::new("orders")
481 .dead_letter("orders.dlq")
482 .max_deliveries(3)
483 .validate()
484 .expect("a cap with a dead-letter topic is valid");
485 }
486}