datum_mq/
source.rs

1use std::{
2    sync::{
3        Arc, Condvar, Mutex,
4        atomic::{AtomicBool, Ordering},
5    },
6    time::{Duration, Instant},
7};
8
9use crate::native::{
10    AutoOffsetReset, FetchTuning, GroupAssignor, NativeCommitPolicy, NativeKafkaConsumerConfig,
11    NativeKafkaSource, StartOffset, TopicPartitionAssignment,
12};
13use crate::offset::{OffsetCommitter, OffsetPosition};
14use crate::{
15    CommitPolicy, ConsumerRecord, KafkaBatchOffset, KafkaConsumerSettings, KafkaMetrics,
16    KafkaOffset, KafkaTimestamp, MqError, MqResult, Offset, Subscription, TopicPartition,
17};
18use bytes::Bytes;
19use datum::{Source, SourceWithContext, StreamError};
20
21/// Kafka source entry points.
22pub struct KafkaSource;
23
24#[derive(Debug, Clone)]
25struct KafkaEnvelope {
26    record: ConsumerRecord,
27    offset: KafkaOffset,
28}
29
30/// Payload-only Kafka record emitted by [`KafkaSource::committable_payload_batches`].
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct KafkaPayloadRecord {
33    pub partition: i32,
34    pub offset: i64,
35    pub timestamp: KafkaTimestamp,
36    payload_start: usize,
37    payload_end: usize,
38}
39
40impl KafkaPayloadRecord {
41    #[must_use]
42    pub fn payload<'a>(&self, batch: &'a KafkaPayloadBatch) -> &'a [u8] {
43        &batch.payloads[self.payload_start..self.payload_end]
44    }
45}
46
47/// A payload batch plus one committable Kafka watermark per touched partition.
48#[derive(Debug, Clone)]
49pub struct KafkaPayloadBatch {
50    records: Vec<KafkaPayloadRecord>,
51    payloads: Bytes,
52    offset: KafkaBatchOffset,
53    active_partitions: Vec<TopicPartition>,
54}
55
56impl KafkaPayloadBatch {
57    #[must_use]
58    pub fn records(&self) -> &[KafkaPayloadRecord] {
59        &self.records
60    }
61
62    #[must_use]
63    pub fn len(&self) -> usize {
64        self.records.len()
65    }
66
67    #[must_use]
68    pub fn is_empty(&self) -> bool {
69        self.records.is_empty()
70    }
71
72    #[must_use]
73    pub fn payload(&self, record: &KafkaPayloadRecord) -> &[u8] {
74        record.payload(self)
75    }
76
77    #[must_use]
78    pub fn offset(&self) -> &KafkaBatchOffset {
79        &self.offset
80    }
81
82    /// Topic/partitions assigned to the consumer when this batch was emitted.
83    ///
84    /// The list includes partitions that did not contribute a record to this
85    /// batch. Event-time consumers use it to min-reduce watermarks across idle
86    /// or slower Kafka partitions before exposing a merged stream.
87    #[must_use]
88    pub fn active_partitions(&self) -> &[TopicPartition] {
89        &self.active_partitions
90    }
91
92    /// Commits this batch after the downstream batch checkpoint succeeds.
93    pub fn commit(&self) -> MqResult<()> {
94        self.offset.commit()
95    }
96}
97
98impl KafkaSource {
99    /// Creates a plain source of owned Kafka records.
100    ///
101    /// Use [`KafkaSource::committable`] when records should advance consumer-group
102    /// offsets after downstream checkpoints.
103    #[must_use]
104    pub fn plain(
105        settings: KafkaConsumerSettings,
106        subscription: Subscription,
107    ) -> Source<ConsumerRecord, KafkaControl> {
108        native_envelope_source(settings, subscription).map(|envelope| envelope.record)
109    }
110
111    /// Creates a source whose records carry committable Kafka offsets as context.
112    ///
113    /// Call [`KafkaOffset::commit`](crate::KafkaOffset::commit) only after the
114    /// corresponding downstream checkpoint or side effect succeeds.
115    #[must_use]
116    pub fn committable(
117        settings: KafkaConsumerSettings,
118        subscription: Subscription,
119    ) -> SourceWithContext<ConsumerRecord, KafkaOffset, KafkaControl> {
120        native_envelope_source(settings, subscription)
121            .as_source_with_context(|envelope| envelope.offset.clone())
122            .map(|envelope| envelope.record)
123    }
124
125    /// Creates a payload-only batched source with committable partition watermarks.
126    ///
127    /// This is the throughput-oriented counterpart to [`KafkaSource::committable`].
128    /// It emits one Datum element per Kafka poll batch, copies payload bytes into a
129    /// batch buffer, and exposes a single [`KafkaBatchOffset`] for the highest
130    /// offset in each touched topic/partition. Downstream code should call
131    /// [`KafkaPayloadBatch::commit`] only after the whole batch checkpoint succeeds.
132    #[must_use]
133    pub fn committable_payload_batches(
134        settings: KafkaConsumerSettings,
135        subscription: Subscription,
136    ) -> Source<KafkaPayloadBatch, KafkaControl> {
137        native_payload_batch_source(settings, subscription)
138    }
139}
140
141fn failed_controlled_source<Out: Send + 'static>(
142    control: KafkaControl,
143    message: String,
144) -> Source<Out, KafkaControl> {
145    Source::unfold_resource(
146        move || -> datum::StreamResult<()> { Err(StreamError::Failed(message.clone())) },
147        |_| Ok(None),
148        |_| Ok(()),
149    )
150    .map_materialized_value(move |_| control.clone())
151}
152
153fn native_payload_batch_source(
154    settings: KafkaConsumerSettings,
155    subscription: Subscription,
156) -> Source<KafkaPayloadBatch, KafkaControl> {
157    let metrics = KafkaMetrics::default();
158    let control = KafkaControl::new(metrics.clone());
159    let config = match native_config_from_settings(&settings, &subscription) {
160        Ok(config) => config,
161        Err(error) => {
162            return failed_controlled_source(control, error.to_string());
163        }
164    };
165    NativeKafkaSource::payload_batches(config)
166        .map(native_batch_to_mq)
167        .map_materialized_value({
168            let control = control.clone();
169            move |native_control| {
170                let delegate = Arc::new(NativeControlDelegate { native_control });
171                delegate.refresh_metrics(&control.state.metrics);
172                control.register_delegate(delegate);
173                control.clone()
174            }
175        })
176}
177
178fn native_envelope_source(
179    settings: KafkaConsumerSettings,
180    subscription: Subscription,
181) -> Source<KafkaEnvelope, KafkaControl> {
182    let metrics = KafkaMetrics::default();
183    let control = KafkaControl::new(metrics.clone());
184    let config = match native_config_from_settings(&settings, &subscription) {
185        Ok(config) => config,
186        Err(error) => {
187            return failed_controlled_source(control, error.to_string());
188        }
189    };
190    NativeKafkaSource::payload_batches(config)
191        .map_concat(native_batch_to_envelopes)
192        .map_materialized_value({
193            let control = control.clone();
194            move |native_control| {
195                let delegate = Arc::new(NativeControlDelegate { native_control });
196                delegate.refresh_metrics(&control.state.metrics);
197                control.register_delegate(delegate);
198                control.clone()
199            }
200        })
201}
202
203/// Materialized control for a Kafka source.
204#[derive(Debug, Clone)]
205pub struct KafkaControl {
206    state: Arc<SourceControlState>,
207}
208
209struct SourceControlState {
210    draining: AtomicBool,
211    shutdown_now: AtomicBool,
212    metrics: KafkaMetrics,
213    delegate: Mutex<Option<Arc<dyn KafkaControlDelegate>>>,
214    wait_lock: Mutex<()>,
215    wait_cv: Condvar,
216}
217
218impl std::fmt::Debug for SourceControlState {
219    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
220        f.debug_struct("SourceControlState")
221            .field("draining", &self.draining.load(Ordering::Relaxed))
222            .field("shutdown_now", &self.shutdown_now.load(Ordering::Relaxed))
223            .field("metrics", &self.metrics.snapshot())
224            .finish_non_exhaustive()
225    }
226}
227
228trait KafkaControlDelegate: Send + Sync {
229    fn drain_and_shutdown(&self, timeout: Duration) -> MqResult<()>;
230    fn shutdown_now(&self);
231    fn refresh_metrics(&self, metrics: &KafkaMetrics);
232}
233
234impl KafkaControl {
235    fn new(metrics: KafkaMetrics) -> Self {
236        Self {
237            state: Arc::new(SourceControlState {
238                draining: AtomicBool::new(false),
239                shutdown_now: AtomicBool::new(false),
240                metrics,
241                delegate: Mutex::new(None),
242                wait_lock: Mutex::new(()),
243                wait_cv: Condvar::new(),
244            }),
245        }
246    }
247
248    #[must_use]
249    pub fn metrics(&self) -> KafkaMetrics {
250        if let Some(delegate) = self.delegate() {
251            delegate.refresh_metrics(&self.state.metrics);
252        }
253        self.state.metrics.clone()
254    }
255
256    #[must_use]
257    pub fn is_draining(&self) -> bool {
258        self.state.draining.load(Ordering::SeqCst)
259    }
260
261    /// Stops new emission and waits for already emitted offsets to be committed.
262    ///
263    /// The source iterator completes on its next pull once outstanding offsets are
264    /// drained. This is the Datum equivalent of the Alpakka `DrainingControl`
265    /// hook; `datum-agent` jobs can call this before stopping the materialized
266    /// graph.
267    pub fn drain_and_shutdown(&self, timeout: Duration) -> MqResult<()> {
268        self.state.draining.store(true, Ordering::SeqCst);
269        self.state.wait_cv.notify_all();
270        if let Some(delegate) = self.delegate() {
271            let result = delegate.drain_and_shutdown(timeout);
272            delegate.refresh_metrics(&self.state.metrics);
273            return result;
274        }
275        let deadline = Instant::now() + timeout;
276        let mut guard = self
277            .state
278            .wait_lock
279            .lock()
280            .map_err(|_| MqError::Failed("Kafka source drain lock poisoned".to_owned()))?;
281        loop {
282            if self.metrics().snapshot().outstanding == 0 {
283                return Ok(());
284            }
285            let now = Instant::now();
286            if now >= deadline {
287                return Err(MqError::DrainTimeout);
288            }
289            let remaining = deadline.saturating_duration_since(now);
290            let (next_guard, result) = self
291                .state
292                .wait_cv
293                .wait_timeout(guard, remaining)
294                .map_err(|_| MqError::Failed("Kafka source drain wait poisoned".to_owned()))?;
295            guard = next_guard;
296            if result.timed_out() && self.metrics().snapshot().outstanding != 0 {
297                return Err(MqError::DrainTimeout);
298            }
299        }
300    }
301
302    /// Stops the source without claiming uncommitted offsets.
303    pub fn shutdown_now(&self) {
304        self.state.shutdown_now.store(true, Ordering::SeqCst);
305        self.state.wait_cv.notify_all();
306        if let Some(delegate) = self.delegate() {
307            delegate.shutdown_now();
308            delegate.refresh_metrics(&self.state.metrics);
309        }
310    }
311
312    fn register_delegate(&self, delegate: Arc<dyn KafkaControlDelegate>) {
313        if let Ok(mut current) = self.state.delegate.lock() {
314            *current = Some(delegate);
315        }
316    }
317
318    fn delegate(&self) -> Option<Arc<dyn KafkaControlDelegate>> {
319        self.state
320            .delegate
321            .lock()
322            .ok()
323            .and_then(|delegate| delegate.clone())
324    }
325}
326
327struct NativeControlDelegate {
328    native_control: crate::native::NativeKafkaControl,
329}
330
331impl KafkaControlDelegate for NativeControlDelegate {
332    fn drain_and_shutdown(&self, timeout: Duration) -> MqResult<()> {
333        self.native_control
334            .drain_and_shutdown(timeout)
335            .map_err(native_error)
336    }
337
338    fn shutdown_now(&self) {
339        self.native_control.shutdown_now();
340    }
341
342    fn refresh_metrics(&self, metrics: &KafkaMetrics) {
343        let snapshot = self.native_control.metrics();
344        metrics.replace_from(crate::KafkaMetricsSnapshot {
345            assigned_partitions: snapshot.assigned_partitions,
346            revoked_partitions: snapshot.revoked_partitions,
347            lost_partitions: snapshot.lost_partitions,
348            rebalances: snapshot.rebalances,
349            emitted: snapshot.emitted,
350            committed_offsets: snapshot.committed_offsets,
351            commit_failures: snapshot.broker_commit_failures,
352            paused: false,
353            outstanding: snapshot.outstanding,
354            producer_in_flight: 0,
355            produced: 0,
356            delivery_failures: 0,
357            queue_full: 0,
358            high_watermark: u64_to_i64(snapshot.high_watermark),
359            committed_watermark: u64_to_i64(snapshot.committed_watermark),
360        });
361    }
362}
363
364fn u64_to_i64(value: u64) -> i64 {
365    i64::try_from(value).unwrap_or(i64::MAX)
366}
367
368fn native_batch_to_mq(batch: crate::native::KafkaPayloadBatch) -> KafkaPayloadBatch {
369    let commit_batch = batch.clone();
370    let parts = batch.into_parts();
371    let records = parts
372        .records
373        .into_iter()
374        .map(|record| {
375            let range = record.payload_range();
376            KafkaPayloadRecord {
377                partition: record.partition,
378                offset: record.offset,
379                timestamp: native_timestamp(record.timestamp),
380                payload_start: range.start,
381                payload_end: range.end,
382            }
383        })
384        .collect::<Vec<_>>();
385    let positions = parts
386        .watermarks
387        .iter()
388        .map(|watermark| {
389            OffsetPosition::new(
390                watermark.topic.clone(),
391                watermark.partition,
392                watermark.offset - 1,
393            )
394        })
395        .collect::<Vec<_>>();
396    let active_partitions = parts
397        .active_partitions
398        .into_iter()
399        .map(|partition| TopicPartition::new(partition.topic, partition.partition))
400        .collect();
401    KafkaPayloadBatch {
402        records,
403        payloads: parts.payloads,
404        offset: KafkaBatchOffset::custom(positions, move || {
405            commit_batch.commit().map_err(native_error)
406        }),
407        active_partitions,
408    }
409}
410
411fn native_batch_to_envelopes(batch: crate::native::KafkaPayloadBatch) -> Vec<KafkaEnvelope> {
412    let batch = Arc::new(batch);
413    batch
414        .records()
415        .iter()
416        .map(|record| {
417            let topic = record.topic(&batch).to_owned();
418            let position = OffsetPosition::new(topic.clone(), record.partition, record.offset);
419            let offset_committer: Arc<dyn OffsetCommitter> = Arc::new(NativeRecordCommitter {
420                batch: Arc::clone(&batch),
421            });
422            KafkaEnvelope {
423                record: ConsumerRecord {
424                    topic,
425                    partition: record.partition,
426                    offset: record.offset,
427                    timestamp: native_timestamp(record.timestamp),
428                    key: None,
429                    payload: Some(Bytes::copy_from_slice(batch.payload(record))),
430                    headers: Vec::new(),
431                },
432                offset: KafkaOffset::new(position, offset_committer),
433            }
434        })
435        .collect()
436}
437
438struct NativeRecordCommitter {
439    batch: Arc<crate::native::KafkaPayloadBatch>,
440}
441
442impl OffsetCommitter for NativeRecordCommitter {
443    fn stage_offset(&self, position: OffsetPosition) -> MqResult<()> {
444        self.commit_position(position)
445    }
446
447    fn commit_offset(&self, position: OffsetPosition) -> MqResult<()> {
448        self.commit_position(position)
449    }
450}
451
452impl NativeRecordCommitter {
453    fn commit_position(&self, position: OffsetPosition) -> MqResult<()> {
454        self.batch
455            .commit_offset(position.topic, position.partition, position.offset)
456            .map_err(native_error)
457    }
458}
459
460fn native_config_from_settings(
461    settings: &KafkaConsumerSettings,
462    subscription: &Subscription,
463) -> MqResult<NativeKafkaConsumerConfig> {
464    subscription.validate()?;
465    validate_native_boundary(settings)?;
466    let bootstrap = settings.config.get("bootstrap.servers").ok_or_else(|| {
467        MqError::InvalidConfig(
468            "native Kafka consumer backend requires bootstrap.servers".to_owned(),
469        )
470    })?;
471    let mut config = match subscription {
472        Subscription::Assignment(partitions) => NativeKafkaConsumerConfig::new(
473            bootstrap.to_owned(),
474            partitions
475                .iter()
476                .map(native_assignment)
477                .collect::<MqResult<Vec<_>>>()?,
478        ),
479        Subscription::Topics(topics) => {
480            NativeKafkaConsumerConfig::subscribe(bootstrap.to_owned(), topics.clone())
481        }
482        Subscription::Pattern(_) => {
483            return Err(unsupported_native(
484                "pattern subscriptions are not supported by the native backend",
485            ));
486        }
487    };
488    config.group_id = settings.group_id.clone();
489    config.commit_policy = match settings.commit_policy {
490        CommitPolicy::Manual => NativeCommitPolicy::Manual,
491        CommitPolicy::External => NativeCommitPolicy::External,
492        CommitPolicy::AutoKafka => {
493            return Err(unsupported_native(
494                "Kafka auto-commit is not supported by the native backend",
495            ));
496        }
497    };
498    config.commit_batch_size = settings.commit_batch_size;
499    config.commit_interval = settings.commit_interval;
500    config.fetch = native_fetch_tuning(settings)?;
501    config.auto_offset_reset = native_auto_offset_reset(settings)?;
502    config.client.security = crate::native::native_security_config(&settings.config)?;
503    if let Some(client_id) = settings.config.get("client.id") {
504        config.client.client_id = client_id.to_owned();
505    }
506    if let Some(request_timeout_ms) = parse_u64_config(settings, "request.timeout.ms")? {
507        config.client.request_timeout = Duration::from_millis(request_timeout_ms);
508    }
509    if let Some(group_instance_id) = settings.config.get("group.instance.id") {
510        config.group_instance_id = Some(group_instance_id.to_owned());
511    }
512    if let Some(assignor) = settings.config.get("partition.assignment.strategy") {
513        config.group_assignor = native_group_assignor(assignor)?;
514    }
515    if let Some(session_timeout_ms) = parse_u64_config(settings, "session.timeout.ms")? {
516        config.session_timeout = Duration::from_millis(session_timeout_ms);
517    }
518    if let Some(rebalance_timeout_ms) = parse_u64_config(settings, "max.poll.interval.ms")? {
519        config.rebalance_timeout = Duration::from_millis(rebalance_timeout_ms);
520    }
521    if let Some(heartbeat_interval_ms) = parse_u64_config(settings, "heartbeat.interval.ms")? {
522        config.heartbeat_interval = Duration::from_millis(heartbeat_interval_ms);
523    }
524    Ok(config)
525}
526
527fn validate_native_boundary(settings: &KafkaConsumerSettings) -> MqResult<()> {
528    for (key, value) in settings.config.properties() {
529        let key_lower = key.to_ascii_lowercase();
530        let value_lower = value.trim().to_ascii_lowercase();
531        if matches!(key_lower.as_str(), "transactional.id") || key_lower.starts_with("transaction.")
532        {
533            return Err(unsupported_native(
534                "transactions are not supported by the native backend",
535            ));
536        }
537        if matches!(
538            key_lower.as_str(),
539            "compression.type" | "compression.codec" | "topic.compression.codec"
540        ) {
541            match value_lower.as_str() {
542                "" | "none" | "uncompressed" | "gzip" | "snappy" | "lz4" | "zstd" => {}
543                other => {
544                    return Err(MqError::InvalidConfig(format!(
545                        "native Kafka consumer compression setting must be none, gzip, snappy, lz4, or zstd; got {other}"
546                    )));
547                }
548            }
549        }
550        if key_lower == "enable.auto.commit" && raw_bool_true(&value_lower) {
551            return Err(unsupported_native(
552                "Kafka auto-commit is not supported by the native backend; use CommitPolicy::Manual or External",
553            ));
554        }
555        if key_lower == "enable.auto.offset.store" && raw_bool_true(&value_lower) {
556            return Err(unsupported_native(
557                "Kafka auto offset store is not supported by the native backend; use CommitPolicy::Manual or External",
558            ));
559        }
560        if key_lower == "api.version.request" && !raw_bool_true(&value_lower) {
561            return Err(unsupported_native(
562                "broker/version fallback is outside the native support matrix; native requires ApiVersions negotiation against Kafka 2.8+",
563            ));
564        }
565        if key_lower == "broker.version.fallback" {
566            return Err(unsupported_native(
567                "broker.version.fallback is outside the native support matrix; native requires Kafka 2.8+ ApiVersions negotiation",
568            ));
569        }
570    }
571    if settings
572        .config
573        .get("isolation.level")
574        .is_some_and(|value| value.eq_ignore_ascii_case("read_committed"))
575    {
576        return Err(unsupported_native(
577            "read_committed isolation and transactions are not supported by the native backend",
578        ));
579    }
580    if let Some(group_protocol) = settings.config.get("group.protocol") {
581        match group_protocol.trim().to_ascii_lowercase().as_str() {
582            "" | "classic" => {}
583            "consumer" => {
584                return Err(unsupported_native(
585                    "KIP-848 group.protocol=consumer is not supported by the native backend; use classic groups",
586                ));
587            }
588            other => {
589                return Err(unsupported_native(format!(
590                    "group.protocol={other} is not supported by the native backend; use classic groups"
591                )));
592            }
593        }
594    }
595    Ok(())
596}
597
598fn native_assignment(
599    partition: &crate::TopicPartitionOffset,
600) -> MqResult<TopicPartitionAssignment> {
601    let start_offset = match partition.offset {
602        Offset::Beginning => StartOffset::Beginning,
603        Offset::End => StartOffset::End,
604        Offset::Stored => StartOffset::Committed,
605        Offset::Offset(offset) => StartOffset::Offset(offset),
606        Offset::OffsetTail(_) => {
607            return Err(unsupported_native(
608                "tail-relative assignment offsets are not supported by the native backend",
609            ));
610        }
611        Offset::Invalid => {
612            return Err(MqError::InvalidConfig(format!(
613                "native Kafka assignment for {}:{} has invalid start offset",
614                partition.topic, partition.partition
615            )));
616        }
617    };
618    Ok(TopicPartitionAssignment {
619        topic: partition.topic.clone(),
620        partition: partition.partition,
621        start_offset,
622    })
623}
624
625fn native_fetch_tuning(settings: &KafkaConsumerSettings) -> MqResult<FetchTuning> {
626    let mut tuning = FetchTuning {
627        output_records: settings.poll_batch_size,
628        output_payload_bytes: settings.poll_batch_size.saturating_mul(256).max(1),
629        include_timestamps: settings.include_payload_timestamps,
630        high_watermark: settings.high_watermark,
631        low_watermark: settings.low_watermark,
632        ..FetchTuning::default()
633    };
634    if let Some(value) = parse_i32_config(settings, "fetch.wait.max.ms")? {
635        tuning.max_wait_ms = value;
636    }
637    if let Some(value) = parse_i32_config(settings, "fetch.min.bytes")? {
638        tuning.min_bytes = value;
639    }
640    if let Some(value) = parse_i32_config(settings, "fetch.max.bytes")? {
641        tuning.max_bytes = value;
642    }
643    if let Some(value) = parse_i32_config(settings, "max.partition.fetch.bytes")?
644        .or(parse_i32_config(settings, "fetch.message.max.bytes")?)
645    {
646        tuning.partition_max_bytes = value;
647    }
648    Ok(tuning)
649}
650
651fn native_auto_offset_reset(settings: &KafkaConsumerSettings) -> MqResult<AutoOffsetReset> {
652    match settings
653        .config
654        .get("auto.offset.reset")
655        .unwrap_or("latest")
656        .trim()
657        .to_ascii_lowercase()
658        .as_str()
659    {
660        "earliest" | "smallest" => Ok(AutoOffsetReset::Earliest),
661        "latest" | "largest" => Ok(AutoOffsetReset::Latest),
662        "error" | "none" => Ok(AutoOffsetReset::Error),
663        other => Err(MqError::InvalidConfig(format!(
664            "native Kafka consumer backend does not recognize auto.offset.reset={other}"
665        ))),
666    }
667}
668
669fn native_group_assignor(value: &str) -> MqResult<GroupAssignor> {
670    if value.contains(',') {
671        return Err(unsupported_native(
672            "native Kafka consumer backend supports one classic assignor at a time (cooperative-sticky or range), not an assignor list",
673        ));
674    }
675    let normalized = value.trim().to_ascii_lowercase();
676    if normalized == "cooperative-sticky" || normalized.contains("cooperativestickyassignor") {
677        Ok(GroupAssignor::CooperativeSticky)
678    } else if normalized == "range" || normalized.contains("rangeassignor") {
679        Ok(GroupAssignor::Range)
680    } else {
681        Err(unsupported_native(format!(
682            "partition.assignment.strategy={value} is not supported by the native backend"
683        )))
684    }
685}
686
687fn parse_i32_config(settings: &KafkaConsumerSettings, key: &str) -> MqResult<Option<i32>> {
688    settings
689        .config
690        .get(key)
691        .map(|value| {
692            value.parse::<i32>().map_err(|_| {
693                MqError::InvalidConfig(format!(
694                    "native Kafka consumer backend expected integer {key}, got {value}"
695                ))
696            })
697        })
698        .transpose()
699}
700
701fn parse_u64_config(settings: &KafkaConsumerSettings, key: &str) -> MqResult<Option<u64>> {
702    settings
703        .config
704        .get(key)
705        .map(|value| {
706            value.parse::<u64>().map_err(|_| {
707                MqError::InvalidConfig(format!(
708                    "native Kafka consumer backend expected integer millisecond {key}, got {value}"
709                ))
710            })
711        })
712        .transpose()
713}
714
715fn raw_bool_true(value: &str) -> bool {
716    matches!(value.trim(), "1" | "true" | "TRUE" | "True" | "yes" | "YES")
717}
718
719fn native_timestamp(timestamp: crate::native::KafkaTimestamp) -> KafkaTimestamp {
720    match timestamp {
721        crate::native::KafkaTimestamp::NotAvailable => KafkaTimestamp::NotAvailable,
722        crate::native::KafkaTimestamp::CreateTime(value) => KafkaTimestamp::CreateTime(value),
723        crate::native::KafkaTimestamp::LogAppendTime(value) => KafkaTimestamp::LogAppendTime(value),
724    }
725}
726
727fn native_error(error: crate::native::KafkaClientError) -> MqError {
728    match error {
729        crate::native::KafkaClientError::InvalidConfig(message) => {
730            MqError::InvalidConfig(format!("native Kafka consumer backend: {message}"))
731        }
732        crate::native::KafkaClientError::Unsupported(message) => MqError::InvalidConfig(format!(
733            "native Kafka consumer backend: {message}; {}",
734            crate::consumer_backend_hint()
735        )),
736        crate::native::KafkaClientError::AssignmentLost { topic, partition } => {
737            MqError::AssignmentLost { topic, partition }
738        }
739        crate::native::KafkaClientError::Tls { broker, message } => {
740            MqError::NativeTls { broker, message }
741        }
742        crate::native::KafkaClientError::Sasl {
743            broker,
744            mechanism,
745            message,
746        } => MqError::NativeSasl {
747            broker,
748            mechanism,
749            message,
750        },
751        other => MqError::Failed(format!("native Kafka consumer backend failed: {other}")),
752    }
753}
754
755fn unsupported_native(message: impl Into<String>) -> MqError {
756    MqError::InvalidConfig(format!(
757        "native Kafka consumer backend unsupported boundary: {}; {}",
758        message.into(),
759        crate::consumer_backend_hint()
760    ))
761}
762
763#[cfg(test)]
764mod native_tests {
765    use super::*;
766    use crate::KafkaConsumerBackend;
767
768    fn native_settings() -> KafkaConsumerSettings {
769        KafkaConsumerSettings::new("127.0.0.1:9092", "group")
770            .with_consumer_backend(KafkaConsumerBackend::Native)
771    }
772
773    fn native_error_for(settings: KafkaConsumerSettings) -> String {
774        native_config_from_settings(&settings, &Subscription::topics(["topic"]))
775            .expect_err("native config should reject unsupported boundary")
776            .to_string()
777    }
778
779    #[test]
780    fn native_backend_rejects_unsupported_boundaries() {
781        let cases = [
782            (
783                native_settings()
784                    .with("security.protocol", "ssl")
785                    .with("ssl.endpoint.identification.algorithm", ""),
786                "verification",
787            ),
788            (
789                native_settings()
790                    .with("security.protocol", "sasl_ssl")
791                    .with("sasl.mechanism", "GSSAPI")
792                    .with("sasl.username", "user")
793                    .with("sasl.password", "secret"),
794                "GSSAPI",
795            ),
796            (
797                native_settings().with("isolation.level", "read_committed"),
798                "read_committed",
799            ),
800            (
801                native_settings().with("transactional.id", "tx"),
802                "transactions",
803            ),
804            (
805                native_settings().with("group.protocol", "consumer"),
806                "KIP-848",
807            ),
808            (
809                native_settings().with("api.version.request", "false"),
810                "support matrix",
811            ),
812        ];
813
814        for (settings, expected) in cases {
815            let error = native_error_for(settings);
816            assert!(
817                error.contains(expected),
818                "expected '{error}' to name '{expected}'"
819            );
820            assert!(
821                error.contains("only the native Kafka consumer backend"),
822                "unsupported boundary must identify the native-only backend: {error}"
823            );
824        }
825    }
826
827    #[test]
828    fn native_backend_accepts_tls_and_ratified_sasl_mechanisms() {
829        for codec in ["none", "gzip", "snappy", "lz4", "zstd"] {
830            assert!(
831                native_config_from_settings(
832                    &native_settings().with("compression.type", codec),
833                    &Subscription::topics(["topic"]),
834                )
835                .is_ok(),
836                "codec={codec}"
837            );
838        }
839        assert!(
840            native_config_from_settings(
841                &native_settings().with("security.protocol", "ssl"),
842                &Subscription::topics(["topic"]),
843            )
844            .is_ok()
845        );
846        for mechanism in ["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"] {
847            let settings = native_settings()
848                .with("security.protocol", "sasl_ssl")
849                .with("sasl.mechanism", mechanism)
850                .with("sasl.username", "user")
851                .with("sasl.password", "secret");
852            assert!(
853                native_config_from_settings(&settings, &Subscription::topics(["topic"])).is_ok(),
854                "mechanism={mechanism}"
855            );
856        }
857    }
858
859    #[test]
860    fn native_backend_rejects_auto_commit_and_pattern_subscriptions() {
861        let error = native_config_from_settings(
862            &native_settings().with_commit_policy(CommitPolicy::AutoKafka),
863            &Subscription::topics(["topic"]),
864        )
865        .expect_err("native auto commit should be rejected")
866        .to_string();
867        assert!(error.contains("auto-commit"));
868
869        let error =
870            native_config_from_settings(&native_settings(), &Subscription::pattern("topic-.*"))
871                .expect_err("native pattern subscription should be rejected")
872                .to_string();
873        assert!(error.contains("pattern subscriptions"));
874    }
875}
datum_mq/source.rs

datum_mq/
source.rs
