datum-mq 0.10.7

#![allow(dead_code)]

//! Datum-native Kafka consumer and producer engine.
//!
//! This module is private to `datum-mq`; users configure it through the public
//! Kafka consumer and producer settings.
//! Both native backends are the defaults; `KafkaProducerBackend` selects the
//! optional rdkafka compatibility producer when that feature is enabled.
//! Both native paths support plaintext, rustls TLS, SASL PLAIN/SCRAM, and
//! record-batch v2 LZ4/Zstd. Native produce enables idempotent retries by
//! default (no transactional EOS): `InitProducerId` negotiates a producer
//! id/epoch and stamps a per-`(topic, partition)`
//! sequence into each batch, so a retried batch is deduplicated by the broker
//! (`DUPLICATE_SEQUENCE_NUMBER` is success) and `UNKNOWN_PRODUCER_ID` /
//! `OUT_OF_ORDER_SEQUENCE_NUMBER` re-initialize the producer id. Produce
//! requests are bounded per broker connection and never overlap for the same
//! topic-partition; failures halt and settle the issued pipeline before
//! retrying in original order. Users can explicitly opt out with
//! `enable.idempotence=false`.

mod client;
mod connection;
mod error;
mod model;
mod producer;
pub(crate) mod profile;
pub(crate) mod protocol;
mod security;
mod source;

pub(crate) use client::{
    AutoOffsetReset, FetchTuning, GroupAssignor, NativeCommitPolicy, NativeKafkaConsumerConfig,
};
pub(crate) use error::{KafkaClientError, KafkaClientResult};
pub(crate) use model::{KafkaPayloadBatch, KafkaTimestamp, StartOffset, TopicPartitionAssignment};
pub(crate) use producer::{NativeKafkaProducerControl, NativeKafkaProducerHandle};
pub(crate) use security::{NativeSecurityConfig, native_security_config};
pub(crate) use source::{NativeKafkaControl, NativeKafkaSource};

pub(crate) const VERSION: &str = env!("CARGO_PKG_VERSION");

/// Benchmark-only shim exposing the private record-batch encoder.
///
/// Encodes one produce record batch either with the idempotent header
/// (`InitProducerId` producer id/epoch plus a base sequence) or with the
/// non-idempotent `-1` sentinels, returning the wire bytes. The criterion
/// harness is a separate crate that cannot reach the private protocol encoder,
/// so the KC-11 overhead micro-benchmark calls through
/// `datum_mq::bench_support`. Gated behind the internal `bench-internals`
/// feature; not part of the supported API.
#[cfg(feature = "bench-internals")]
#[doc(hidden)]
#[must_use]
pub fn bench_encode_produce_batch(records: &[crate::ProducerRecord], idempotent: bool) -> Vec<u8> {
    let refs = records.iter().collect::<Vec<_>>();
    let identity = idempotent.then_some(protocol::BatchIdentity {
        producer_id: 4242,
        producer_epoch: 0,
        base_sequence: 0,
    });
    let mut zstd_context =
        protocol::ZstdEncoderContext::new().expect("benchmark zstd encoder context");
    protocol::encode_produce_record_batch(
        &refs,
        protocol::CompressionCodec::None,
        &mut zstd_context,
        identity,
    )
    .expect("benchmark record batch encodes")
}

#[cfg(all(test, feature = "rdkafka"))]
pub(crate) use client::NativeKafkaConsumer;
#[cfg(test)]
pub(crate) use model::TopicPartition;
#[cfg(all(test, feature = "rdkafka"))]
mod tests;