datum-mq 0.11.2

#![forbid(unsafe_code)]
//! Compile-checked snippets backing docs/guides/datum-mq.md.
//!
//! Construction/blueprint-level only: these build `KafkaSource`/`KafkaSink`
//! blueprints and never open a broker connection, so they compile and run in CI
//! without Kafka. The consumer/producer clients are created at materialization,
//! not at construction. The end-to-end committable-flow example in the guide is
//! a fenced block adapted from `tests/kafka_integration.rs` and labelled as
//! broker-gated.

use datum_mq::{
    CommitPolicy, KafkaConsumerSettings, KafkaProducerSettings, KafkaSink, KafkaSource,
    Subscription,
};

#[test]
fn kafka_committable_source_blueprint() {
    // #region mq-committable
    use datum::SourceWithContext;
    use datum_mq::{ConsumerRecord, KafkaControl, KafkaOffset};

    // At-least-once consumer: each record carries a committable KafkaOffset as
    // context. Datum-managed manual commits are the default (CommitPolicy::Manual).
    let settings = KafkaConsumerSettings::new("127.0.0.1:9092", "orders-consumer")
        .with("auto.offset.reset", "earliest")
        .with_commit_policy(CommitPolicy::Manual)
        .with_backpressure(2_048, 4_096);

    let source: SourceWithContext<ConsumerRecord, KafkaOffset, KafkaControl> =
        KafkaSource::committable(settings, Subscription::topics(["orders"]));

    // Building the blueprint connects nothing; materialization starts the
    // consumer group and its poll loop.
    let _ = source;
    // #endregion mq-committable
}

#[test]
fn kafka_sink_blueprint() {
    // #region mq-sink
    use datum::Sink;
    use datum_mq::{KafkaProducerControl, ProducerRecord};

    // Idempotent producer (enable.idempotence + acks=all) is the default from
    // KafkaProducerSettings::new. The record carries its own target topic.
    let sink: Sink<ProducerRecord, KafkaProducerControl> =
        KafkaSink::plain(KafkaProducerSettings::new("127.0.0.1:9092"));

    let _ = sink;
    // #endregion mq-sink
}

#[test]
fn kafka_native_producer_backend_blueprint() {
    use datum::Sink;
    use datum_mq::{
        KafkaProducerBackend, KafkaProducerControl, KafkaProducerSettings, KafkaSink,
        ProducerRecord,
    };

    // Native is rustls/SASL-capable, supports gzip/Snappy/LZ4/Zstd, and defaults to
    // idempotent retries (acks=all) so response-loss retries do not duplicate.
    // Native transactions / consume-transform-produce EOS are still not claimed.
    let settings = KafkaProducerSettings::new("127.0.0.1:9092")
        .with_producer_backend(KafkaProducerBackend::Native);
    let sink: Sink<ProducerRecord, KafkaProducerControl> = KafkaSink::plain(settings);

    // Construction is still a pure blueprint; no broker connection is opened.
    let _ = sink;
}

#[test]
fn kafka_native_consumer_backend_blueprint() {
    // #region mq-native-backend
    use datum_mq::{KafkaConsumerBackend, KafkaConsumerSettings, KafkaSource, Subscription};

    let settings = KafkaConsumerSettings::new("127.0.0.1:9092", "orders-consumer")
        .with_consumer_backend(KafkaConsumerBackend::Native)
        .with("auto.offset.reset", "earliest");

    let source =
        KafkaSource::committable_payload_batches(settings, Subscription::topics(["orders"]));

    let _ = source;
    // #endregion mq-native-backend
}

#[test]
fn cookbook_native_committable_at_least_once() {
    // #region cookbook-kafka-native-committable
    use datum::{Sink, StreamError};
    use datum_mq::{
        CommitPolicy, ConsumerRecord, KafkaConsumerBackend, KafkaConsumerSettings, KafkaOffset,
        KafkaSource, Subscription,
    };

    // Broker-gated: set these when running the recipe against a local Kafka topic
    // that already contains at least three order records.
    let Some(topic) = std::env::var("DATUM_DOCS_ORDERS_TOPIC").ok() else {
        return;
    };
    let bootstrap =
        std::env::var("MQ_BOOTSTRAP_SERVERS").unwrap_or_else(|_| "127.0.0.1:9092".to_owned());
    let group = format!("datum-docs-native-{}", std::process::id());

    let settings = KafkaConsumerSettings::new(bootstrap, group)
        .with_consumer_backend(KafkaConsumerBackend::Native)
        .with_commit_policy(CommitPolicy::Manual)
        .with("auto.offset.reset", "earliest")
        .with("partition.assignment.strategy", "cooperative-sticky")
        .with_backpressure(512, 1_024)
        .with_poll_batch_size(64);

    let processed = KafkaSource::committable(settings, Subscription::topics([topic]))
        .as_source()
        .take(3)
        .run_with(Sink::fold_result(
            0_u64,
            |count, (record, offset): (ConsumerRecord, KafkaOffset)| {
                // Do durable, idempotent work first. Use topic/partition/offset
                // from the record or offset as your de-duplication key.
                let _dedupe_key = (&record.topic, record.partition, record.offset);

                // Commit only after the side effect/checkpoint succeeds.
                offset.commit().map_err(StreamError::from)?;
                Ok(count + 1)
            },
        ))
        .expect("native Kafka consumer materializes")
        .wait()
        .expect("native Kafka consumer completes");

    println!("processed {processed} orders");
    assert_eq!(processed, 3);
    // #endregion cookbook-kafka-native-committable
}

#[test]
fn cookbook_native_consumer_group_drain() {
    // #region cookbook-kafka-native-consumer-group
    use std::time::Duration;

    use datum::{Keep, Sink, StreamCompletion, StreamError};
    use datum_mq::{
        KafkaConsumerBackend, KafkaConsumerSettings, KafkaControl, KafkaPayloadBatch, KafkaSource,
        MqError, Subscription,
    };

    // Broker-gated: the topic should exist and receive records while this
    // process is running.
    let Some(topic) = std::env::var("DATUM_DOCS_GROUP_TOPIC").ok() else {
        return;
    };
    let bootstrap =
        std::env::var("MQ_BOOTSTRAP_SERVERS").unwrap_or_else(|_| "127.0.0.1:9092".to_owned());

    let settings = KafkaConsumerSettings::new(bootstrap, "orders-workers")
        .with_consumer_backend(KafkaConsumerBackend::Native)
        .with("auto.offset.reset", "earliest")
        // The native backend accepts one classic assignor. Cooperative-sticky is
        // the default; set it explicitly when you want rollout parity.
        .with("partition.assignment.strategy", "cooperative-sticky")
        .with_backpressure(256, 512)
        .with_commit_batch_size(128)
        .with_commit_interval(Duration::from_millis(100));

    let (control, completion): (KafkaControl, StreamCompletion<datum::NotUsed>) =
        KafkaSource::committable_payload_batches(settings, Subscription::topics([topic]))
            .to_mat(
                Sink::foreach_result(|batch: KafkaPayloadBatch| {
                    for record in batch.records() {
                        let _payload = batch.payload(record);
                        // Process each payload here.
                    }

                    match batch.commit() {
                        Ok(()) => Ok(()),
                        Err(error @ MqError::AssignmentLost { .. }) => {
                            // The group no longer owns at least one partition in
                            // this batch. Do not retry this commit; fail the
                            // stream and let the group replay from the last
                            // committed offset.
                            Err(StreamError::from(error))
                        }
                        Err(error) => Err(StreamError::from(error)),
                    }
                }),
                Keep::both,
            )
            .run()
            .expect("native Kafka group consumer materializes");

    // On SIGTERM or deploy shutdown, stop taking new records and wait for
    // already emitted batches to commit.
    control
        .drain_and_shutdown(Duration::from_secs(30))
        .expect("consumer drains outstanding commits");
    completion.wait().expect("consumer completes after drain");

    let metrics = control.metrics().snapshot();
    println!(
        "drained; committed={} rebalances={} lost={}",
        metrics.committed_offsets, metrics.rebalances, metrics.lost_partitions
    );
    // #endregion cookbook-kafka-native-consumer-group
}