LibrdKafka Unleash
Real-Time Data at Lightning Speed

				
					#include <stdio.h>
#include <signal.h>
#include <librdkafka/rdkafka.h>

static volatile int running = 1;

static void stop(int sig) {
    running = 0;
}

int main() {
    rd_kafka_t *consumer;
    rd_kafka_conf_t *conf;
    rd_kafka_topic_partition_list_t *topics;
    char errstr[512];

    signal(SIGINT, stop);
    signal(SIGTERM, stop);

    conf = rd_kafka_conf_new();

    rd_kafka_conf_set(conf, "bootstrap.servers",
                      "localhost:9092", NULL, 0);

    rd_kafka_conf_set(conf, "group.id",
                      "my-consumer-group", NULL, 0);

    rd_kafka_conf_set(conf, "auto.offset.reset",
                      "earliest", NULL, 0);

    consumer = rd_kafka_new(RD_KAFKA_CONSUMER,
                            conf,
                            errstr, sizeof(errstr));

    if (!consumer) {
        fprintf(stderr, "Failed to create consumer: %s\n", errstr);
        return 1;
    }

    rd_kafka_poll_set_consumer(consumer);

    topics = rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_list_add(topics, "my-topic", -1);
    rd_kafka_subscribe(consumer, topics);

    while (running) {
        rd_kafka_message_t *msg;

        msg = rd_kafka_consumer_poll(consumer, 1000);
        if (!msg)
            continue;

        if (msg->err) {
            if (msg->err != RD_KAFKA_RESP_ERR__PARTITION_EOF) {
                fprintf(stderr, "Consumer error: %s\n",
                        rd_kafka_message_errstr(msg));
            }
        } else {
            printf("Received: %.*s\n",
                   (int)msg->len,
                   (char *)msg->payload);
        }

        rd_kafka_message_destroy(msg);
    }

    rd_kafka_consumer_close(consumer);
    rd_kafka_topic_partition_list_destroy(topics);
    rd_kafka_destroy(consumer);

    return 0;
}