Struct KafkaProducer 

pub struct KafkaProducer<K, V> { /* private fields */ }

A Kafka client that publishes records to the Kafka cluster.

The producer is thread-safe and sharing a single producer instance across tasks will generally be faster than having multiple instances.

The producer consists of a pool of buffer space that holds records that haven’t yet been transmitted to the server, as well as a background I/O task that is responsible for turning these records into requests and transmitting them to the cluster. Failure to close the producer after use will leak these resources.

The send method is asynchronous. When called, it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.

Implementations

impl<K, V> KafkaProducer<K, V>

pub const NETWORK_THREAD_PREFIX: &str = "kafka-producer-network-thread"

Network thread name prefix.

pub const PRODUCER_METRIC_GROUP_NAME: &str = "producer-metrics"

Producer metric group name.

pub fn new( config: ProducerConfig, key_serializer: impl Serializer<K> + 'static, value_serializer: impl Serializer<V> + 'static, ) -> Result<Self, Error>

where K: 'static, V: 'static,

A producer is instantiated by providing a set of key-value pairs as configuration, a key and a value Serializer.

Note: after creating a KafkaProducer you must always close it to avoid resource leaks.

Arguments

• config - The producer configs
• key_serializer - The serializer for key that implements Serializer. The configure method won’t be called in the producer when the serializer is passed in directly.
• value_serializer - The serializer for value that implements Serializer. The configure method won’t be called in the producer when the serializer is passed in directly.

Errors

• Error::is_kafka_error if the producer cannot be constructed; the cause is its source.

Examples

use std::collections::HashMap;
use confluent_kafka::producer::KafkaProducer;
use confluent_kafka::producer::ProducerConfig;
use confluent_kafka::common::serialization::StringSerializer;

let props = HashMap::from([
    ("bootstrap.servers".to_string(), "localhost:9092".to_string()),
]);
let config = ProducerConfig::new(&props).expect("Invalid config");

let producer = KafkaProducer::<String, String>::new(
    config,
    StringSerializer::default(),
    StringSerializer::default(),
).expect("Failed to create producer");

Translation notes

Java’s constructors without serializers are not translated: Rust cannot instantiate the class named by key.serializer.

pub fn init_transactions( &self, ) -> impl Future<Output = Result<(), Error>> + Send

Needs to be called before any other methods when the transactional.id is set in the configuration. This method does the following:

1. Ensures any transactions initiated by previous instances of the producer with the same transactional.id are completed. If the previous instance had failed with a transaction in progress, it will be aborted. If the last transaction had begun completion, but not yet finished, this method awaits its completion.
2. Gets the internal producer id and epoch, used in all future transactional messages issued by the producer.

Note that this method will return a timeout error if the transactional state cannot be initialized before expiration of max.block.ms. Additionally, it will return an interrupt error if interrupted. It is safe to retry in either case, but once the transactional state has been successfully initialized, this method should no longer be used.

Errors

• Error::is_local_illegal_state_error if no transactional.id has been configured, or the producer has been closed
• Error::is_unsupported_version_error fatal error indicating the broker does not support transactions (i.e. if its version is lower than 0.11.0.0)
• Error::is_authorization_error error indicating that the configured transactional.id is not authorized, or the idempotent producer id is unavailable. See the error for more details. User may retry this function call after fixing the permission.
• Error::is_kafka_error if the producer has encountered a previous fatal error or for any other unexpected error
• Error::is_timeout_error if the time taken for initialize the transaction has surpassed max.block.ms.
• Error::is_interrupt_error if the task is interrupted while blocked

Translation notes

A Rust task is never interrupted: dropping the future cancels the wait.

pub fn begin_transaction(&self) -> Result<(), Error>

Should be called before the start of each new transaction. Note that prior to the first invocation of this method, you must invoke Self::init_transactions exactly one time.

Errors

• Error::is_local_illegal_state_error if no transactional.id has been configured or if Self::init_transactions has not yet been invoked, or the producer has been closed
• Error::is_producer_fenced_error if another producer with the same transactional.id is active
• Error::is_invalid_producer_epoch_error if the producer has attempted to produce with an old epoch to the partition leader. See the error for more details
• Error::is_unsupported_version_error fatal error indicating the broker does not support transactions (i.e. if its version is lower than 0.11.0.0)
• Error::is_authorization_error fatal error indicating that the configured transactional.id is not authorized. See the error for more details
• Error::is_kafka_error if the producer has encountered a previous fatal error or for any other unexpected error

Translation notes

Synchronous, as Java’s beginTransaction never blocks.

pub fn send_offsets_to_transaction( &self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, group_metadata: &dyn ConsumerGroupMetadata, ) -> impl Future<Output = Result<(), Error>> + Send

Sends a list of specified offsets to the consumer group coordinator, and also marks those offsets as part of the current transaction. These offsets will be considered committed only if the transaction is committed successfully. The committed offset should be the next message your application will consume, i.e. next_record_to_be_processed.offset() (or ConsumerRecords::next_offsets). You should also add the leader epoch as commit metadata, which can be obtained from ConsumerRecord::leader_epoch or ConsumerRecords::next_offsets.

This method should be used when you need to batch consumed and produced messages together, typically in a consume-transform-produce pattern. Thus, the specified group_metadata should be extracted from the used consumer via KafkaConsumer::group_metadata to leverage consumer group metadata. This will provide stronger fencing than just supplying the consumer group id, however note that the full set of consumer group metadata returned by KafkaConsumer::group_metadata requires the brokers to be on version 2.5 or newer to understand.

This method is a blocking call that waits until the request has been received and acknowledged by the consumer group coordinator; but the offsets are not considered as committed until the transaction itself is successfully committed later (via the Self::commit_transaction call).

Note, that the consumer should have enable.auto.commit=false and should also not commit offsets manually (via sync or async commits). This method will return a timeout error if the producer cannot resolve the metadata for the topics in offsets and send the offsets before expiration of max.block.ms. Additionally, it will return an interrupt error if interrupted.

Arguments

• offsets - the offsets to commit, per partition; an empty map does nothing
• group_metadata - the consumer group metadata, from KafkaConsumer::group_metadata

Errors

• Error::is_local_illegal_argument_error if group_metadata has a generation id greater than zero but an unknown member id
• Error::is_local_illegal_state_error if no transactional.id has been configured or no transaction has been started, or the producer has been closed.
• Error::is_producer_fenced_error fatal error indicating another producer with the same transactional.id is active
• Error::is_unsupported_version_error fatal error indicating the broker does not support transactions (i.e. if its version is lower than 0.11.0.0) or the broker doesn’t support the latest version of transactional API with all consumer group metadata (i.e. if its version is lower than 2.5.0).
• Error::is_unsupported_for_message_format_error fatal error indicating the message format used for the offsets topic on the broker does not support transactions
• Error::is_authorization_error fatal error indicating that the configured transactional.id is not authorized, or the consumer group id is not authorized.
• Error::is_consumer_commit_failed_error if the commit failed and cannot be retried (e.g. if the consumer has been kicked out of the group). Users should handle this by aborting the transaction.
• Error::is_fenced_instance_id_error if this producer instance gets fenced by broker due to a mis-configured consumer instance id within group metadata.
• Error::is_invalid_producer_epoch_error if the producer has attempted to produce with an old epoch to the partition leader. See the error for more details
• Error::is_kafka_error if the producer has encountered a previous fatal or abortable error, or for any other unexpected error
• Error::is_timeout_error if the combined time taken for resolving topic metadata and sending the offsets has surpassed max.block.ms.
• Error::is_interrupt_error if the task is interrupted while blocked
• Error::is_bootstrap_resolution_error if DNS resolution of the bootstrap servers fails within bootstrap.resolve.timeout.ms

Translation notes

A Rust task is never interrupted: dropping the future cancels the wait. offsets is taken by value, as the request carries it to the coordinator; group_metadata is borrowed and copied.

pub fn commit_transaction( &self, ) -> impl Future<Output = Result<(), Error>> + Send

Commits the ongoing transaction. This method will flush any unsent records before actually committing the transaction.

Further, if any of the send calls which were part of the transaction hit irrecoverable errors, this method will return the last received error immediately and the transaction will not be committed. So all send calls in a transaction must succeed in order for this method to succeed.

If the transaction is committed successfully and this method returns without an error, it is guaranteed that all callbacks for records in the transaction will have been invoked and completed. Note that panics raised by callbacks are ignored; the producer proceeds to commit the transaction in any case.

Note that this method will return a timeout error if the transaction cannot be committed before expiration of max.block.ms, but this does not mean the request did not actually reach the broker. In fact, it only indicates that we cannot get the acknowledgement response in time, so it’s up to the application’s logic to decide how to handle timeouts. Additionally, it will return an interrupt error if interrupted. It is safe to retry in either case, but it is not possible to attempt a different operation (such as Self::abort_transaction) since the commit may already be in the progress of completing. If not retrying, the only option is to close the producer.

Errors

• Error::is_local_illegal_state_error if no transactional.id has been configured or no transaction has been started, or the producer has been closed
• Error::is_producer_fenced_error fatal error indicating another producer with the same transactional.id is active
• Error::is_unsupported_version_error fatal error indicating the broker does not support transactions (i.e. if its version is lower than 0.11.0.0)
• Error::is_authorization_error fatal error indicating that the configured transactional.id is not authorized. See the error for more details
• Error::is_invalid_producer_epoch_error if the producer has attempted to produce with an old epoch to the partition leader. See the error for more details
• Error::is_kafka_error if the producer has encountered a previous fatal or abortable error, or for any other unexpected error
• Error::is_timeout_error if the time taken for committing the transaction has surpassed max.block.ms.
• Error::is_interrupt_error if the task is interrupted while blocked

Translation notes

A Rust task is never interrupted: dropping the future cancels the wait.

pub fn abort_transaction( &self, ) -> impl Future<Output = Result<(), Error>> + Send

Aborts the ongoing transaction. Any unflushed produce messages will be aborted when this call is made. This call will return an error immediately if any prior send calls failed with a producer-fenced error or an authorization error.

Note that this method will return a timeout error if the transaction cannot be aborted before expiration of max.block.ms, but this does not mean the request did not actually reach the broker. In fact, it only indicates that we cannot get the acknowledgement response in time, so it’s up to the application’s logic to decide how to handle timeouts. Additionally, it will return an interrupt error if interrupted. It is safe to retry in either case, but it is not possible to attempt a different operation (such as Self::commit_transaction) since the abort may already be in the progress of completing. If not retrying, the only option is to close the producer.

Errors

• Error::is_local_illegal_state_error if no transactional.id has been configured or no transaction has been started, or the producer has been closed
• Error::is_producer_fenced_error fatal error indicating another producer with the same transactional.id is active
• Error::is_invalid_producer_epoch_error if the producer has attempted to produce with an old epoch to the partition leader. See the error for more details
• Error::is_unsupported_version_error fatal error indicating the broker does not support transactions (i.e. if its version is lower than 0.11.0.0)
• Error::is_authorization_error fatal error indicating that the configured transactional.id is not authorized. See the error for more details
• Error::is_kafka_error if the producer has encountered a previous fatal error or for any other unexpected error
• Error::is_timeout_error if the time taken for aborting the transaction has surpassed max.block.ms.
• Error::is_interrupt_error if the task is interrupted while blocked

Translation notes

A Rust task is never interrupted: dropping the future cancels the wait.

impl KafkaProducer<Vec<u8>, Vec<u8>>

pub fn send( &self, record: ProducerRecord<&[u8], &[u8]>, callback: Option<Box<dyn Callback>>, ) -> impl Future<Output = Result<KafkaFuture<RecordMetadata>, Error>> + Send

Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged. See Producer::send_with_callback for details.

Arguments

• record - The record to send. If the topic or the partition specified in it cannot be found in metadata within max.block.ms, the returned future will time out when retrieved.
• callback - A user-supplied callback to execute when the record has been acknowledged by the server (None indicates no callback)

Returns

A future which will eventually contain the response information.

Errors

• Error::is_local_illegal_state_error if a transactional.id has been configured and no transaction has been started, or when send is invoked after producer has been closed.
• Error::is_interrupt_error If the task is interrupted while blocked
• Error::is_serialization_error If the key or value are not valid objects given the configured serializers
• Error::is_bootstrap_resolution_error if DNS resolution of the bootstrap servers fails within bootstrap.resolve.timeout.ms
• Error::is_kafka_error If a Kafka related error occurs that does not belong to the public API errors.

Translation notes

The byte-slice key and value bypass serialization, so no serialization error is returned: they are written straight into the batch buffer, the zero-copy path for callers that already hold &[u8] data (the C FFI). A Rust task is never interrupted: dropping the future cancels the wait.

Trait Implementations

impl<K, V> Debug for KafkaProducer<K, V>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<K, V> Drop for KafkaProducer<K, V>

Starts a graceful close of a producer that was not closed, so its sender task still sends the records it holds and then stops.

Java has no counterpart: its I/O thread is a daemon (built in the KafkaProducer constructor), which keeps sending the records of a producer nothing references any more. The sender task here does the same for as long as its runtime runs.

fn drop(&mut self)

Executes the destructor for this type.

impl<K, V> Producer<K, V> for KafkaProducer<K, V>

where K: Send + Sync, V: Send + Sync,

async fn init_transactions(&self) -> Result<(), Error>

Needs to be called before any other methods when the transactional.id is set in the configuration.

fn begin_transaction(&self) -> Result<(), Error>

Should be called before the start of each new transaction. Note that prior to the first invocation of this method, you must invoke init_transactions exactly one time.

async fn send_offsets_to_transaction( &self, offsets: HashMap<TopicPartition, OffsetAndMetadata>, group_metadata: &dyn ConsumerGroupMetadata, ) -> Result<(), Error>

Sends a list of specified offsets to the consumer group coordinator, and also marks those offsets as part of the current transaction. These offsets will be considered committed only if the transaction is committed successfully.

async fn commit_transaction(&self) -> Result<(), Error>

Commits the ongoing transaction. This method will flush any unsent records before actually committing the transaction.

async fn abort_transaction(&self) -> Result<(), Error>

Aborts the ongoing transaction. Any unflushed produce messages will be aborted when this call is made.

async fn send( &self, record: ProducerRecord<K, V>, ) -> Result<KafkaFuture<RecordMetadata>, Error>

Asynchronously send a record to a topic. Equivalent to send_with_callback(record, None). See send_with_callback for details.

async fn send_with_callback( &self, record: ProducerRecord<K, V>, callback: Option<Box<dyn Callback>>, ) -> Result<KafkaFuture<RecordMetadata>, Error>

Asynchronously send a record to a topic and invoke the provided callback when the send has been acknowledged.

async fn flush(&self) -> Result<(), Error>

Invoking this method makes all buffered records immediately available to send (even if linger.ms is greater than 0) and blocks on the completion of the requests associated with these records. The post-condition of flush() is that any previously sent record will have completed (e.g. its future is done and callbacks passed to Self::send_with_callback have been called). A request is considered completed when it is successfully acknowledged according to the acks configuration you have specified or else it results in an error.

fn metrics(&self) -> MetricsView

Get the full set of internal metrics maintained by the producer.

async fn partitions_for(&self, topic: &str) -> Result<Vec<PartitionInfo>, Error>

Get the partition metadata for the given topic. This can be used for custom partitioning.

async fn close(&self) -> Result<(), Error>

Close this producer. This method blocks until all previously sent requests complete. This method is equivalent to close_with_timeout(Duration::MAX).

async fn close_with_timeout(&self, timeout: Duration) -> Result<(), Error>

This method waits up to timeout for the producer to complete the sending of all incomplete requests.