Trabajo con Apache Kafka en C++: librería librdkafka y ejemplos prácticos

Librería librdkafka: una potente herramienta para trabajar con Kafka en C++

Apache Kafka es una plataforma distribuida para transmisión de eventos que se ha convertido en el estándar de facto para la construcción de sistemas de intercambio de mensajes de alta carga. Para la integración de Kafka con C++, la librería más popular y madura es librdkafka, una librería C de alto rendimiento que proporciona un conjunto completo de funcionalidades para trabajar con productores y consumidores de Kafka. Implementa el protocolo Apache Kafka desde la versión 0.8 en adelante, incluyendo las últimas versiones estables.

Funcionalidades principales de librdkafka

• Soporte de todas las operaciones clave: envío de mensajes (producer) y lectura (consumer) con gestión de offsets.
• Trabajo con tópicos, particiones y grupos de consumidores.
• Soporte nativo de compresión (gzip, snappy, lz4, zstd).
• Cifrado SSL/TLS y autenticación SASL (PLAIN, SCRAM, GSSAPI, OAUTHBEARER).
• Productor idempotente y transacciones (desde Kafka 0.11).
• Procesamiento asíncrono con callbacks y baja latencia.
• Ajuste fino mediante propiedades de configuración (similar al cliente Java).

Instalación y configuración

Puede instalar librdkafka a través del gestor de paquetes o compilarlo desde el código fuente. Para Linux (Ubuntu/Debian), el comando es:

sudo apt-get install librdkafka-dev

Para macOS mediante Homebrew:

brew install librdkafka

Después de la instalación, incluya el archivo de cabecera y la librería al compilar:

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

Ejemplo: creación de un productor (Producer)

El productor envía mensajes al tópico especificado. Veamos un ejemplo mínimo funcional:

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

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

    // Configuración del broker    conf->set("bootstrap.servers", "localhost:9092", errstr);

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

    // Envío del mensaje    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;    }

    // Espera de entrega    producer->flush(10000);    delete producer;    return 0;}

En este ejemplo creamos una configuración, especificamos la dirección del broker, creamos una instancia de Producer y enviamos un mensaje de texto al tópico test-topic. El método flush garantiza que todos los mensajes se envíen antes de que finalice el programa.

Ejemplo: creación de un consumidor (Consumer)

El consumidor lee mensajes de un tópico como parte de un grupo de consumidores. Ejemplo con gestión manual de offsets:

#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