Working with Apache Kafka in C++: librdkafka Library and Practical Examples

librdkafka Library: A Powerful Tool for Working with Kafka in C++

Apache Kafka is a distributed event streaming platform that has become the de facto standard for building high-load messaging systems. For integrating Kafka with C++, the most popular and mature library is librdkafka — a high-performance C library that provides a full set of capabilities for working with Kafka producers and consumers. It implements the Apache Kafka protocol for versions 0.8 and above, including the latest stable versions.

Key Features of librdkafka

• Support for all key operations: sending messages (producer) and reading (consumer) with offset management.
• Working with topics, partitions, and consumer groups.
• Native support for compression (gzip, snappy, lz4, zstd).
• SSL/TLS encryption and SASL authentication (PLAIN, SCRAM, GSSAPI, OAUTHBEARER).
• Idempotent producer and transactions (since Kafka 0.11).
• Asynchronous processing with callbacks and low latency.
• Fine-tuning through configuration properties (similar to the Java client).

Installation and Setup

You can install librdkafka via a package manager or build it from source. For Linux (Ubuntu/Debian), the command is:

sudo apt-get install librdkafka-dev

For macOS via Homebrew:

brew install librdkafka

After installation, include the header file and link the library during compilation:

g++ -o my_app my_app.cpp -lrdkafka -lpthread -lz

Example: Creating a Producer

The producer sends messages to a specified topic. Here is a minimal working example:

#include <iostream>#include <string>#include <librdkafka/rdkafkacpp.h>

int main() {    std::string errstr;    RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);

    // Configure broker    conf->set("bootstrap.servers", "localhost:9092", errstr);

    // Create producer    RdKafka::Producer *producer = RdKafka::Producer::create(conf, errstr);    if (!producer) {        std::cerr << "Failed to create producer: " << errstr << std::endl;        return 1;    }    delete conf;

    // Send message    std::string topic_name = "test-topic";    std::string payload = "Hello from C++ Kafka!";    RdKafka::ErrorCode resp = producer->produce(        topic_name,        RdKafka::Topic::PARTITION_UA,        RdKafka::Producer::RK_MSG_COPY,        const_cast<char*>(payload.c_str()),        payload.size(),        nullptr, 0, 0, nullptr    );

    if (resp != RdKafka::ERR_NO_ERROR) {        std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;    } else {        std::cout << "Message sent successfully" << std::endl;    }

    // Wait for delivery    producer->flush(10000);    delete producer;    return 0;}

In this example, we create a configuration, specify the broker address, create a Producer instance, and send a string message to the test-topic topic. The flush method ensures that all messages are sent before the program terminates.

Example: Creating a Consumer

The consumer reads messages from a topic as part of a consumer group. Example with manual offset management:

#include <iostream>#include <string>#include <librdkafka/rdkafkacpp.h>

class MyConsumerCb : public RdKafka::ConsumeCb {public:    void consume_cb(RdKafka::Message &msg, void *opaque) override {        if (msg.err()) {            if (msg.err() != RdKafka::ERR__TIMED_OUT) {                std::cerr << "Error: " << msg.errs