librdkafka Documentation

Thread Safety

The rd_kafka_share_t handle is not thread-safe by design. A single handle must not be used concurrently from multiple threads. Concurrent use is detected on a best-effort basis and results in the error RD_KAFKA_RESP_ERR__CONFLICT. The recommended pattern is one handle per thread or serialized access.

Current Limitations (Preview)

• Record Limit: max.poll.records (default 500) is a soft bound; it does not strictly limit the number of records returned per poll.
• Acknowledgement Types: Limited to ACCEPT, RELEASE, and REJECT. There is no support for acquisition-lock renewal (KIP-1222).
• No Wakeup API: Blocking poll or commit calls can only be terminated by their specified timeout.
• Close Behavior: rd_kafka_share_consumer_close() does not accept a timeout; it is bounded by socket.timeout.ms.
• Metadata/Admin: No share-group admin operations (list/describe/delete) or client-metrics APIs are available in this preview.
• Auto-creation: allow.auto.create.topics has no effect for share consumers; topics are resolved by topic ID.
• Fetch Pattern: Each poll fetches from a single broker (round-robin over partitions).

The statistics output is a hierarchical JSON object. The structure follows this general pattern:

• Top-level fields: General client metrics (name, client_id, type, etc.).
• brokers: A dictionary where keys are broker names and values are objects containing per-broker metrics (connection state, throughput, latency).
  • brokers.toppars: Mapping of topic-partitions handled by a specific broker.
• topics: A dictionary where keys are topic names and values are objects containing topic-level metrics.
  • topics.partitions: A dictionary of partition-specific metrics (offsets, consumer lag, fetch/produce queues).
• cgrp (Optional): Consumer group metrics (state, rebalance counts).
• eos (Optional): Exactly-once semantics / Idempotent producer state and metrics.

{
 <Top-level fields>
 "brokers": {
    <brokers fields>,
    "toppars": { <toppars fields> }
 },
 "topics": {
   <topic fields>,
   "partitions": {
     <partitions fields>
   }
 }
[, "cgrp": { <cgrp fields> } ]
[, "eos": { <eos fields> } ]
}

Starting with librdkafka v2.12.0, the KIP-848 protocol is production-ready. This protocol shifts assignment logic from the client (Group Leader) to the broker (Group Coordinator).

Enabling the protocol

Set group.protocol=consumer. By default, the protocol is set to classic.

Key Differences from Classic Protocol

• Assignment: Calculated by the Broker (Group Coordinator) rather than the client.
• Configuration: Many client-side configs (like partition.assignment.strategy) are replaced by broker-side configs or the group.remote.assignor property.
• Regex Subscriptions: Matching is performed on the broker using the Google RE2/J engine. Unlike the libc engine used in the classic protocol, RE2/J requires regexes to match the complete topic name. For example, to match topic-1 and topic-2 using a prefix, use ^topic.* instead of ^topic.
• Rebalance Callbacks: The protocol is fully incremental. Inside callbacks, you must use incremental APIs:
  • rd_kafka_incremental_assign(rk, partitions)
  • rd_kafka_incremental_unassign(rk, partitions)
  • Do not use rd_kafka_assign().
  • The partitions list provided in the callback contains only the specific partitions being added or revoked, not the full set.
• Session Timeout: Enforced by the broker. If the Coordinator is unreachable, the consumer continues fetching but cannot commit offsets.

# Next-Gen Protocol Configuration
group.protocol=consumer
# Optional: select a remote assignor ('uniform' or 'range')
group.remote.assignor=uniform

Librdkafka emits a hierarchical JSON object containing performance metrics. The top-level structure includes:

• Client Metadata: name, client_id, type, ts (timestamp), time.
• Global Counters: msg_cnt, msg_size, tx, rx, txmsgs, rxmsgs, etc.
• brokers: A map of broker connection statistics (e.g., nodeid, state, txbytes, rxbytes, and latency/RTT distributions).
• topics: A map of topic-specific metrics, including batchsize and batchcnt distributions.
• topics.{topic_name}.partitions: A map of partition-level metrics (e.g., leader, consumer_lag, msgq_cnt, txmsgs).

{
  "name": "rdkafka#producer-1",
  "client_id": "rdkafka",
  "type": "producer",
  "ts": 5016483227792,
  "brokers": {
    "localhost:9092/2": {
      "nodeid": 2,
      "state": "UP",
      "txbytes": 84283332
    }
  },
  "topics": {
    "test": {
      "partitions": {
        "0": {
          "partition": 0,
          "consumer_lag": -1
        }
      }
    }
  }
}

Librdkafka uses a poll-based API to trigger callbacks. You must call rd_kafka_poll() at regular intervals to process events and trigger these callbacks.

Poll-triggered Callbacks

• dr_msg_cb: Message delivery report (signals success or failure of a message).
• error_cb: Signals an error. If the error code is RD_KAFKA_RESP_ERR__FATAL, the application should retrieve the reason via rd_kafka_fatal_error() and terminate.
• stats_cb: Emits JSON metrics (if statistics.interval.ms is set).
• throttle_cb: Triggered when a broker throttles a request.

Spontaneous Callbacks (Not triggered by poll)

• log_cb: For application-level logging of librdkafka messages.
• partitioner_cb: Custom logic for determining message partitions.
  • Constraints: Must NOT call any rd_kafka_* functions, must NOT block, and must return a value between 0 and partition_cnt-1 (or RD_KAFKA_PARTITION_UA).

Share groups (KIP-932) provide queue-like semantics to Kafka. Unlike standard consumer groups where one partition is assigned to one member, a share group allows multiple members to consume from the same partitions cooperatively.

Key Concepts:

• Unit of Progress: Individual records, not committed offsets.
• Lifecycle of a Record: available $\rightarrow$ acquired (under a time-limited lock) $\rightarrow$ acknowledged (or archived if it fails/exceeds limits).
• Scaling: Allows scaling consumers beyond the number of partitions.
• Broker Requirement: Requires Kafka 4.2.0+ with share groups enabled.
• API Status: Currently a preview feature available via the C API only (rd_kafka_share_* functions in rdkafka.h). There is no C++ wrapper yet.

Unknown Topics

If a consumer subscribes to a non-existent topic, it will receive an error: RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART.

Topic Propagation: Because topic creation is asynchronous, a new topic might not be immediately visible. librdkafka will not flag a topic as non-existent until a propagation time has elapsed (default 30 seconds, configurable via topic.metadata.propagation.max.ms).

Unauthorized Topics

If a consumer has Describe (ACL) permissions but lacks Read permissions, the Fetch requests will fail with RD_KAFKA_RESP_ERR_TOPIC_AUTHORIZATION_FAILED. This error is raised once per partition, and the fetcher will back off for fetch.error.backoff.ms (minimum 1 second) before retrying. It is recommended to adjust subscriptions to exclude unauthorized topics.

Topic Auto-Creation

Librdkafka supports auto-creation if the broker has auto.create.topics.enable=true.

• Producers: Can trigger auto-creation by producing to a non-existent topic.
• Consumers: By default, consumers prevent triggering auto-creation via the allow.auto.create.topics configuration (set to false). Set this to true to allow consumers to trigger topic creation on the broker.

The linger.ms (also known as queue.buffering.max.ms) configuration property is the primary lever for balancing throughput and latency. It defines how long the producer waits for batch.num.messages or batch.size to fill up in the local per-partition queue before sending the batch to the broker.

• For High Throughput: Increase linger.ms. This allows larger batches to accumulate, amortizing messaging overhead and reducing the number of requests. For maximum throughput, it is recommended to set this to >50ms (often between 100ms and 1000ms).
• For Low Latency: Decrease linger.ms. Setting it to 0 or 0.1 ensures messages are sent as soon as possible. Note that very low values result in smaller batches, increasing CPU, memory, and network overhead.
• General Purpose: A value of 5ms is a good starting point for balanced workloads.

These settings are configured globally via rd_kafka_conf_t but are applied on a per-topic+partition basis.

# Example configurations
linger.ms=5       # General purpose
linger.ms=100     # High throughput
linger.ms=0       # Low latency