DOCSEARCHbyAlgolia
Submit Docs

rust-rdkafka

repository·master·Indexed 24 days ago

https://github.com/fede1024/rust-rdkafka

A fully asynchronous, futures-enabled Apache Kafka client library for Rust built on top of the high-performance C library librdkafka. It provides high-level clients like StreamConsumer and FutureProducer for Tokio integration, as well as low-level wrappers and an AdminClient for managing cluster resources such as topics, partitions, and groups. The library supports at-least-once and exactly-once delivery semantics and includes the rdkafka-sys crate for low-level FFI bindings.

Tokens
13.3K
Snippets
16
Records
81
Agent score
82%

What's inside rust-rdkafka

  1. Integrate rust-rdkafka with Tokio or other runtimes

    master

    The high-level clients (StreamConsumer and FutureProducer) are designed to work seamlessly with the Tokio runtime.

    Using Tokio

    Tokio integration is enabled by default. You can use these clients to write asynchronous message processing code within the Tokio ecosystem.

    Using other runtimes (smol, async-std)

    If you want to avoid the heavyweight Tokio dependency, you can disable default features:

    [dependencies]
rdkafka = { version = "0.25", default-features = false }

    To use a different asynchronous runtime like smol or async-std, you must provide a shim that implements the AsyncRuntime trait.

  2. Implement At-least-once delivery semantics

    master

    To achieve at-least-once delivery, your application must ensure that messages are processed before their offsets are committed.

    Warning: Committing offsets too early can lead to message loss. If a failure occurs after an early commit but before processing is complete, the consumer will resume from the next offset upon recovery, skipping the unprocessed message.

  3. Implement Exactly-once semantics (EOS)

    master

    Exactly-once semantics (EOS) can be achieved using transactional producers. This allows you to commit produced records and consumer offsets atomically.

    To ensure consumers only see these atomic operations, set their isolation.level configuration to read_committed. This is particularly useful in read-process-write scenarios.

  5. How client types work in rust-rdkafka

    master

    The library provides two levels of abstraction for interacting with Kafka: Low-level and High-level.

    Low-level Clients

    These are simple wrappers around librdkafka and require manual management of the polling loop.

    • BaseConsumer: Requires periodic calls to poll() to execute callbacks, handle rebalances, and receive messages.
    • BaseProducer: Requires periodic calls to poll() to execute delivery callbacks.
    • ThreadedProducer: A BaseProducer that manages its own dedicated polling thread.

    High-level Clients

    These are designed for easier integration with asynchronous workflows.

    • StreamConsumer: Provides a Stream of messages and handles polling automatically.
    • FutureProducer: Provides a Future that completes when a message is successfully delivered (or fails).

  6. Install rust-rdkafka via Cargo

    master

    To install rust-rdkafka, add it to your Cargo.toml. It is recommended to use the cmake-build feature, which compiles librdkafka from source and links it statically to your executable.

    Prerequisites

    To compile librdkafka using the cmake-build feature, you need:

    • GNU toolchain
    • GNU make
    • pthreads
    • libcurl-dev (e.g., libcurl4-openssl-dev on Ubuntu)
    • cmake (required for the cmake-build feature)
    • Optional dependencies: zlib (included by default via libz), libssl-dev (via ssl), libsasl2-dev (via gssapi), and libzstd-dev (via zstd-pkg-config).
    [dependencies]
rdkafka = { version = "0.25", features = ["cmake-build"] }

  7. Configure rdkafka-sys linking and build systems

    master

    You can control how rdkafka-sys links to librdkafka and which build system it uses via Cargo features:

    • dynamic-linking: Links to a locally installed version of librdkafka using pkg-config. The system version must exactly match the version bundled with this crate.
    • static-linking: Statically links against a locally built version of librdkafka. This requires setting the DEP_LIBRDKAFKA_STATIC_ROOT environment variable.
    • cmake-build: Uses the CMake build system instead of the default mklove system. This is required for Windows support and requires CMake to be installed on the build machine.

    By default, the crate uses a submodule containing librdkafka sources to compile and statically link the library.

  9. Overview of rust-rdkafka

    master

    rust-rdkafka is a high-performance Kafka client library for Rust. It is built on top of librdkafka (via the rdkafka-sys crate) and is designed to handle high throughput, capable of processing up to one million messages per second.

    Key capabilities include:

    • Asynchronous data processing: Native support for the Tokio runtime.
    • Delivery Guarantees: Supports both 'at-least-once' delivery and 'exactly-once' semantics.
    • High Performance: Optimized for high-volume workloads.

  10. Manage Kafka message headers

    master

    Kafka messages support headers, which are key-value pairs sent alongside the payload. The library provides two ways to handle headers:

    1. Reading Headers (Headers trait)

    Use the Headers trait (implemented by BorrowedHeaders and OwnedHeaders) to access message headers.

    • count(): Returns the number of headers.
    • get(idx): Returns a Header at the specified index. Panics if the index is out of bounds.
    • try_get(idx): Returns Option<Header> if the index is valid, avoiding panics.
    • get_as<V>(idx): Returns a Header with the value parsed into type V. Panics if the index is out of bounds.
    • iter(): Returns an iterator over all headers.

    2. Creating/Modifying Headers (OwnedHeaders)

    To create headers for a producer, use OwnedHeaders.

    • new(): Creates a new header collection with default capacity.
    • new_with_capacity(initial_capacity): Creates a new collection with a specific capacity.
    • insert(header): Inserts a Header into the collection. This follows a builder pattern and returns the OwnedHeaders instance.

    Header Structure

    A Header<'a, V> consists of a key: &'a str and a value: Option<V>.

  11. Identify Kafka resources with ResourceSpecifier

    master

    When performing administrative operations like describing or altering configurations, you must specify which Kafka resource you are targeting using ResourceSpecifier.

    Supported resource types:

    • Topic(&str): Identified by the topic name.
    • Group(&str): Identified by the consumer group ID.
    • Broker(i32): Identified by the broker ID.

    For operations that require ownership of the identifier (such as the results of an AlterConfigs operation), the library uses OwnedResourceSpecifier.

  12. Configure linking for librdkafka

    master

    You can control how librdkafka is linked to your project using Cargo features:

    1. Static Linking (Default): Compiles librdkafka from source and links it statically. Using the cmake-build feature is encouraged.
    2. Dynamic Linking: Use the dynamic-linking feature to link against the system's version of librdkafka instead of compiling it.
    3. Custom Static Linking: Use the static-linking feature and provide the DEP_LIBRDKAFKA_STATIC_ROOT environment variable pointing to your pre-built librdkafka directory.
    [dependencies]
rdkafka = { version = "0.25", features = ["dynamic-linking"]