krabka_client_producer/
lib.rs

1//! Idempotent producer client for Apache Kafka in Rust.
2//!
3//! It builds on [`krabka_client_core`] for transport, and adds full
4//! idempotent-producer semantics: `InitProducerId` on connect, per-batch
5//! `(producer_id, producer_epoch, base_sequence)`, and retries that re-frame
6//! the same `RecordBatch` so the broker's dedup catches them.
7//!
8//! It also supports transactional, exactly-once production:
9//! `init_transactions`, `begin_transaction`, which returns a [`Transaction`]
10//! guard whose `commit`/`abort` finishes it, and `send_offsets_to_transaction`
11//! for the consume-process-produce pattern of KIP-447.
12//!
13//! ## Quick start
14//!
15//! ```no_run
16//! use std::time::Duration;
17//!
18//! use bytes::Bytes;
19//! use krabka_client_producer::{Acks, Compression, Producer, ProducerRecord};
20//!
21//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
22//! let producer = Producer::builder()
23//!     .bootstrap("localhost:9092")
24//!     .compression(Compression::Lz4)
25//!     .acks(Acks::All)
26//!     .linger(Duration::from_millis(5))
27//!     .build()
28//!     .await?;
29//!
30//! let metadata = producer
31//!     .send(ProducerRecord {
32//!         topic: "my-topic".into(),
33//!         value: Some(Bytes::from("hello")),
34//!         ..Default::default()
35//!     })
36//!     .await?;
37//!
38//! producer.flush().await?;
39//! producer.close().await?;
40//! # Ok(())
41//! # }
42//! ```
43//!
44//! ## Pipelined delivery
45//!
46//! Use [`Producer::enqueue`] to queue records without waiting for each
47//! acknowledgement. Each [`DeliveryHandle`] returns one producer result.
48//!
49//! ```no_run
50//! # use krabka_client_producer::{Producer, ProducerError, ProducerRecord};
51//! # async fn run(producer: &Producer, records: Vec<ProducerRecord>) -> Result<(), ProducerError> {
52//! let mut deliveries = Vec::new();
53//! for record in records {
54//!     deliveries.push(producer.enqueue(record).await?);
55//! }
56//! for delivery in deliveries {
57//!     let metadata = delivery.await?;
58//! }
59//! # Ok(())
60//! # }
61//! ```
62//!
63//! ## Capabilities and boundaries
64//!
65//! This crate owns producer-facing semantics: batching, compression,
66//! idempotence, retries, per-record partition overrides, transactional RPCs,
67//! and `send_offsets_to_transaction` for consume-process-produce flows. The
68//! built-in partitioner follows Kafka's: keyed records hash to a partition,
69//! and keyless records use an adaptive sticky partition (KIP-480, KIP-794). Set
70//! `ProducerRecord::partition` to pin an individual record. Serialization is
71//! deliberately owned by the caller: `key` and `value` are raw `Bytes`, so a
72//! schema-registry or serde integration can sit on top without constraining the
73//! producer API.
74
75#![doc(html_root_url = "https://docs.rs/krabka-client-producer/0.7.1")]
76
77mod accumulator;
78mod buffer_pool;
79mod builder;
80#[cfg(test)]
81mod client_failover_model;
82mod compression;
83mod error;
84mod error_class;
85mod metadata_age;
86mod metadata_wait;
87mod partitioner;
88mod producer;
89mod record;
90mod sender;
91mod transactional;
92mod transport;
93mod txn_retry;
94
95pub use builder::{
96    DEFAULT_PRODUCER_ACKS, DEFAULT_PRODUCER_BATCH_BYTES, DEFAULT_PRODUCER_BUFFER_MEMORY,
97    DEFAULT_PRODUCER_COMPRESSION, DEFAULT_PRODUCER_DELIVERY_TIMEOUT,
98    DEFAULT_PRODUCER_FLUSH_TIMEOUT, DEFAULT_PRODUCER_INIT_RETRY_TIMEOUT, DEFAULT_PRODUCER_LINGER,
99    DEFAULT_PRODUCER_MAX_BLOCK, DEFAULT_PRODUCER_MAX_IN_FLIGHT, DEFAULT_PRODUCER_MAX_REQUEST_SIZE,
100    DEFAULT_PRODUCER_PARTITIONER_ADAPTIVE_PARTITIONING_ENABLE,
101    DEFAULT_PRODUCER_PARTITIONER_AVAILABILITY_TIMEOUT, DEFAULT_PRODUCER_PARTITIONER_IGNORE_KEYS,
102    DEFAULT_PRODUCER_PARTITIONER_RACK_AWARE, DEFAULT_PRODUCER_RECEIVE_BUFFER,
103    DEFAULT_PRODUCER_REQUEST_TIMEOUT, DEFAULT_PRODUCER_RETRIES, DEFAULT_PRODUCER_RETRY_BACKOFF,
104    DEFAULT_PRODUCER_RETRY_BACKOFF_MAX, DEFAULT_PRODUCER_SEND_BUFFER,
105    DEFAULT_PRODUCER_TRANSACTION_TIMEOUT, ProducerFlushTimeout, ProducerRetryPolicy,
106    ProducerThroughputPolicy,
107};
108pub use compression::{
109    Compression, CompressionLevels, DEFAULT_PRODUCER_COMPRESSION_GZIP_LEVEL,
110    DEFAULT_PRODUCER_COMPRESSION_LZ4_LEVEL, DEFAULT_PRODUCER_COMPRESSION_ZSTD_LEVEL,
111};
112pub use error::{ProducerError, RecordSizeLimit};
113pub use krabka_client_consumer::ConsumerGroupMetadata;
114pub use metadata_age::{
115    DEFAULT_PRODUCER_METADATA_MAX_AGE, DEFAULT_PRODUCER_METADATA_MAX_IDLE,
116    MIN_PRODUCER_METADATA_MAX_IDLE,
117};
118pub use partitioner::{PartitionInfo, Partitioner, partition_for_key};
119pub use producer::{Acks, Producer};
120pub use record::{DeliveryHandle, Header, ProducerRecord, RecordMetadata};
121pub use transactional::{
122    EndTransactionError, OwnedTransaction, PreparedTransactionState,
123    PreparedTransactionStateParseError, Transaction,
124};
krabka_client_producer/lib.rs

krabka_client_producer/
lib.rs
