C/C++ Client for Apache Kafka

librdkafka is a C/C++ client library for Apache Kafka® that provides a producer and a consumer. librdkafka is licensed under the BSD 2-Clause License.

For a step-by-step guide on building a C/C++ client application for Kafka, see Getting Started with Apache Kafka and C/C++.

Install librdkafka

Install librdkafka from a prebuilt binary or from source. Most users should use the prebuilt binaries. The following installation methods are available:

• On Debian and Red Hat-based Linux distributions: prebuilt packages

• On macOS: the latest release from Homebrew

• On any platform: source code on GitHub

Linux packages are available for the x86_64, arm64, and s390x architectures. Linux s390x is the CPU architecture for IBM Z mainframes and LinuxONE servers. For release updates, see the changelog in the GitHub repo.

Install on Linux

To install librdkafka on Linux, set up the repositories for Debian or Red Hat distributions. Then use apt-get or yum to install the appropriate packages.

For example, to build a C application on a Red Hat-based distribution, install the librdkafka-devel package:

sudo yum install librdkafka-devel

On a Debian-based distribution, install the librdkafka-dev package:

sudo apt-get install librdkafka-dev

Install on macOS

To install the latest librdkafka release on macOS, use Homebrew:

brew install librdkafka

Get the source code

The librdkafka source code is available on GitHub.

Find librdkafka code examples

For a step-by-step tutorial with code samples for the producer and consumer, see Get Started with Apache Kafka and C/C++.

For more examples, see C client examples.

Produce messages with librdkafka

To produce messages with librdkafka, create a producer handle from a configuration, create a topic handle, then call rd_kafka_produce for each message. librdkafka writes messages asynchronously.

Initialize a producer

To initialize a producer, create a configuration object, set the properties, and create the producer handle with rd_kafka_new:

char hostname[128];
char errstr[512];

rd_kafka_conf_t *conf = rd_kafka_conf_new();

if (gethostname(hostname, sizeof(hostname))) {
  fprintf(stderr, "%% Failed to lookup hostname\n");
  exit(1);
}

if (rd_kafka_conf_set(conf, "client.id", hostname,
                      errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
  fprintf(stderr, "%% %s\n", errstr);
  exit(1);
}

if (rd_kafka_conf_set(conf, "bootstrap.servers", "host1:9092,host2:9092",
                      errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
  fprintf(stderr, "%% %s\n", errstr);
  exit(1);
}

rd_kafka_topic_conf_t *topic_conf = rd_kafka_topic_conf_new();

if (rd_kafka_topic_conf_set(topic_conf, "acks", "all",
                      errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
  fprintf(stderr, "%% %s\n", errstr);
  exit(1);
}

/* Create Kafka producer handle */
rd_kafka_t *rk;
if (!(rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf,
                        errstr, sizeof(errstr)))) {
  fprintf(stderr, "%% Failed to create new producer: %s\n", errstr);
  exit(1);
}

librdkafka handles the errors for each setting directly.

Write messages asynchronously

To write a message, create a rd_kafka_topic_t handle for the topic, then use rd_kafka_produce to send messages to it. For example:

rd_kafka_topic_t *rkt = rd_kafka_topic_new(rk, topic, topic_conf);

if (rd_kafka_produce(rkt, RD_KAFKA_PARTITION_UA,
                     RD_KAFKA_MSG_F_COPY,
         payload, payload_len,
                     key, key_len,
   NULL) == -1) {
  fprintf(stderr, "%% Failed to produce to topic %s: %s\n",
 topic, rd_kafka_err2str(rd_kafka_errno2err(errno)));
}

The arguments work as follows:

• The third argument to rd_kafka_topic_new is the topic-specific configuration. The example passes topic_conf, which sets acknowledgments. Passing NULL makes the producer use the default configuration.

• The second argument to rd_kafka_produce sets the partition for the message. RD_KAFKA_PARTITION_UA makes librdkafka use the default partitioner to select the partition.

• The RD_KAFKA_MSG_F_COPY flag makes librdkafka copy the payload and key, so you can free them after the call returns.

Warning

The Java producer and the librdkafka producer use different default partitioners. A message produced by each client with the same partition key can map to different partitions:

• The Java producer uses the murmur2 hash function.

• The librdkafka producer uses the crc32 hash function.

Handle delivery reports

To run code after a write completes, configure a delivery callback during initialization:

static void on_delivery(rd_kafka_t *rk,
                        const rd_kafka_message_t *rkmessage,
                        void *opaque) {
  if (rkmessage->err)
    fprintf(stderr, "%% Message delivery failed: %s\n",
            rd_kafka_message_errstr(rkmessage));
}

void init_rd_kafka() {
  rd_kafka_conf_t *conf = rd_kafka_conf_new();
  rd_kafka_conf_set_dr_msg_cb(conf, on_delivery);

  // initialization omitted
}

librdkafka invokes the delivery callback on your thread when you call rd_kafka_poll. A common pattern is to call rd_kafka_poll after every call to the produce API. This might not deliver reports regularly if the message produce rate isn’t steady. The API doesn’t provide a direct way to block for the result of a specific message delivery. To do that, see Write messages synchronously.

Write messages synchronously

Synchronous writes are typically a bad idea because they reduce throughput, so librdkafka is asynchronous by nature. Synchronous writes make sense in some cases, but the implementation isn’t trivial. For a full example, see the Sync producer example in the librdkafka wiki.

Consume messages with librdkafka

To consume messages with librdkafka, create a consumer handle from a configuration, subscribe to topics, and poll for messages.

Initialize a consumer

char hostname[128];
char errstr[512];

rd_kafka_conf_t *conf = rd_kafka_conf_new();

if (gethostname(hostname, sizeof(hostname))) {
 fprintf(stderr, "%% Failed to lookup hostname\n");
 exit(1);
}

if (rd_kafka_conf_set(conf, "client.id", hostname,
                     errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
 fprintf(stderr, "%% %s\n", errstr);
 exit(1);
}

if (rd_kafka_conf_set(conf, "group.id", "foo",
                     errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
 fprintf(stderr, "%% %s\n", errstr);
 exit(1);
}

if (rd_kafka_conf_set(conf, "bootstrap.servers", "host1:9092,host2:9092",
                     errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
 fprintf(stderr, "%% %s\n", errstr);
 exit(1);
}

/* Create Kafka consumer handle */
rd_kafka_t *rk;
if (!(rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf,
                       errstr, sizeof(errstr)))) {
 fprintf(stderr, "%% Failed to create new consumer: %s\n", errstr);
 exit(1);
}

librdkafka handles configuration errors directly when you set properties.

How the consumer works

librdkafka uses a multi-threaded approach to Kafka consumption. The API returns only a single message or event at a time:

while (running) {
 rd_kafka_message_t *rkmessage = rd_kafka_consumer_poll(rk, 500);
 if (rkmessage) {
   msg_process(rkmessage);
   rd_kafka_message_destroy(rkmessage);

   if ((++msg_count % MIN_COMMIT_COUNT) == 0)
     rd_kafka_commit(rk, NULL, 0);
 }
}

librdkafka does all fetching and coordinator communication in background threads, so the session timeout doesn’t depend on your processing time. However, you must call rd_kafka_consumer_poll at least once every max.poll.interval.ms, or the consumer leaves the group and its partitions are reassigned. Close the client when you’re done, so that a stalled process doesn’t hold its assigned partitions.

Important

Partition rebalances also happen in a background thread. You must still handle potential commit failures, because the consumer might no longer have the same partition assignment when the commit begins. If you enable autocommit, you don’t need to handle commit failures, because librdkafka ignores them silently. In that case, you can’t roll back processing.

while (running) {
  rd_kafka_message_t *rkmessage = rd_kafka_consumer_poll(rk, 1000);
  if (!rkmessage)
    continue; // timeout: no message

  msg_process(rkmessage); // application-specific processing
  rd_kafka_message_destroy(rkmessage);

  if ((++msg_count % MIN_COMMIT_COUNT) == 0) {
    rd_kafka_resp_err_t err = rd_kafka_commit(rk, NULL, 0);
    if (err) {
      // application-specific rollback of processed records
    }
  }
}

Poll for messages in a loop

The consumer API centers on the rd_kafka_consumer_poll method, which retrieves records from the brokers. The rd_kafka_subscribe method controls which topics rd_kafka_consumer_poll fetches. Typically, you call rd_kafka_subscribe once to set the topics of interest, then loop on rd_kafka_consumer_poll until the application shuts down.

static int shutdown = 0;
static void msg_process(rd_kafka_message_t message);

void basic_consume_loop(rd_kafka_t *rk,
                        rd_kafka_topic_partition_list_t *topics) {
  rd_kafka_resp_err_t err;

  if ((err = rd_kafka_subscribe(rk, topics))) {
    fprintf(stderr, "%% Failed to start consuming topics: %s\n", rd_kafka_err2str(err));
    exit(1);
  }

  while (running) {
    rd_kafka_message_t *rkmessage = rd_kafka_consumer_poll(rk, 500);
    if (rkmessage) {
      msg_process(rkmessage);
      rd_kafka_message_destroy(rkmessage);
    }
  }

  err = rd_kafka_consumer_close(rk);
  if (err)
    fprintf(stderr, "%% Failed to close consumer: %s\n", rd_kafka_err2str(err));
  else
    fprintf(stderr, "%% Consumer closed\n");
}

The poll timeout is hard-coded to 500 milliseconds. If the consumer receives no message before the timeout expires, rd_kafka_consumer_poll returns NULL.

Close the consumer

Always call rd_kafka_consumer_close after you finish using the consumer. This call closes active sockets, cleans up internal state, and immediately triggers a group rebalance that reassigns the partitions the consumer owned to other members of the group. If you don’t close the consumer, the broker triggers the rebalance only after the session timeout expires.

Commit offsets synchronously

The simplest and most reliable way to manually commit offsets is rd_kafka_commit, which supports both synchronous and asynchronous commits. rd_kafka_consumer_poll returns single messages, not batches, so keep that in mind when you commit.

void consume_loop(rd_kafka_t *rk,
                  rd_kafka_topic_partition_list_t *topics) {
  static const int MIN_COMMIT_COUNT = 1000;

  int msg_count = 0;
  rd_kafka_resp_err_t err;

  if ((err = rd_kafka_subscribe(rk, topics))) {
    fprintf(stderr, "%% Failed to start consuming topics: %s\n", rd_kafka_err2str(err));
    exit(1);
  }

  while (running) {
    rd_kafka_message_t *rkmessage = rd_kafka_consumer_poll(rk, 500);
    if (rkmessage) {
      msg_process(rkmessage);
      rd_kafka_message_destroy(rkmessage);

      if ((++msg_count % MIN_COMMIT_COUNT) == 0)
        rd_kafka_commit(rk, NULL, 0);
    }
 }

  err = rd_kafka_consumer_close(rk);
  if (err)
    fprintf(stderr, "%% Failed to close consumer: %s\n", rd_kafka_err2str(err));
  else
    fprintf(stderr, "%% Consumer closed\n");
}

In this example, the consumer commits synchronously every 1,000 messages. The second argument to rd_kafka_commit is the list of offsets to commit. If you set it to NULL, librdkafka commits the latest offsets for the assigned positions. The third argument is a flag that controls whether the call is asynchronous. You can also trigger the commit when a timeout expires, to ensure the committed position updates regularly.

Choose a delivery guarantee

Committing after processing a message gives you at-least-once delivery. Committing before processing gives you at-most-once delivery, but you must handle commit failures carefully. The synchronous commit example gives at-least-once delivery because the commit follows message processing.

The following example commits before processing to get at-most-once delivery:

void consume_loop(rd_kafka_t *rk,
                  rd_kafka_topic_partition_list_t *topics) {
  rd_kafka_resp_err_t err;

  if ((err = rd_kafka_subscribe(rk, topics))) {
    fprintf(stderr, "%% Failed to start consuming topics: %s\n", rd_kafka_err2str(err));
    exit(1);
  }

  while (running) {
    rd_kafka_message_t *rkmessage = rd_kafka_consumer_poll(rk, 500);
    if (rkmessage && !rd_kafka_commit_message(rk, rkmessage, 0)) {
      msg_process(rkmessage);
      rd_kafka_message_destroy(rkmessage);
    }
  }

  err = rd_kafka_consumer_close(rk);
  if (err)
    fprintf(stderr, "%% Failed to close consumer: %s\n", rd_kafka_err2str(err));
  else
    fprintf(stderr, "%% Consumer closed\n");
}

For simplicity, this example calls rd_kafka_commit_message before it processes the message. Committing on every message produces a lot of overhead in practice. A better approach is to collect a batch of messages, run the synchronous commit, and then process the messages only if the commit succeeds.

Commit offsets asynchronously

To commit asynchronously, enable the asynchronous flag in the call to rd_kafka_commit. The consumer sends the request and returns immediately.

void consume_loop(rd_kafka_t *rk,
                  rd_kafka_topic_partition_list_t *topics) {
  static const int MIN_COMMIT_COUNT = 1000;

  int msg_count = 0;
  rd_kafka_resp_err_t err;

  if ((err = rd_kafka_subscribe(rk, topics))) {
    fprintf(stderr, "%% Failed to start consuming topics: %s\n", rd_kafka_err2str(err));
    exit(1);
  }

  while (running) {
    rd_kafka_message_t *rkmessage = rd_kafka_consumer_poll(rk, 500);
    if (rkmessage) {
      msg_process(rkmessage);
      rd_kafka_message_destroy(rkmessage);

      if ((++msg_count % MIN_COMMIT_COUNT) == 0)
        rd_kafka_commit(rk, NULL, 1);
    }
  }

  err = rd_kafka_consumer_close(rk);
  if (err)
    fprintf(stderr, "%% Failed to close consumer: %s\n", rd_kafka_err2str(err));
  else
    fprintf(stderr, "%% Consumer closed\n");
}

librdkafka gives you a callback that it invokes when the commit succeeds or fails. Configure the callback during initialization:

static void on_commit(rd_kafka_t *rk,
                      rd_kafka_resp_err_t err,
                      rd_kafka_topic_partition_list_t *offsets,
                      void *opaque) {
  if (err)
    fprintf(stderr, "%% Failed to commit offsets: %s\n", rd_kafka_err2str(err));
}

void init_rd_kafka() {
  rd_kafka_conf_t *conf = rd_kafka_conf_new();
  rd_kafka_conf_set_offset_commit_cb(conf, on_commit);

  // initialization omitted
}

librdkafka API documentation

For the full C/C++ Client client API, see the librdkafka API documentation.

Related content

• Getting Started with Apache Kafka and C/C++ (Confluent Developer)

• Apache Kafka 101 (Confluent Developer)

• Debuting a Modern C++ API for Apache Kafka (Confluent Blog)