librdkafka consumer is not receiving message from broker

0
votes

I have written kafka consumer and producer in C using librdkafka library. Kafka broker version is kafka_2.12-2.3.0. Producer is successfully producing message and dr_msg_cb function confirms the successful delivery. However, the consumer is not receiving message from broker. Can someone please help to debug further?

I can see that TCP connection from consumer to broker is ESTABLISHED. But TCPdump shows that broker is not sending any data to consumer. I enabled debug on the consumer code and below are the messages.

[2019 Nov  8 19:18:09.458553135:155:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-SSL: rdkafka#consumer-2: [thrd:app]: Loading CA certificate(s) from file /mnt/ifc/cfg/d
ata/securedata/clientcerts/kafka/ApicCa.crt
[2019 Nov  8 19:18:09.458880860:156:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-SSL: rdkafka#consumer-2: [thrd:app]: Loading certificate from file /mnt/ifc/cfg/data/se
curedata/clientcerts/kafka/KafkaClient.crt
[2019 Nov  8 19:18:09.459151178:157:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-SSL: rdkafka#consumer-2: [thrd:app]: Loading private key file from /mnt/ifc/cfg/data/se
curedata/clientcerts/kafka/KafkaClient8.key
[2019 Nov  8 19:18:09.459583515:158:E:logger:1741] TID 06 : [LOG_TRACE]:RDKAFKA-7-BRKMAIN: rdkafka#consumer-2: [thrd::0/internal]: :0/internal: Enter main broker thread
[2019 Nov  8 19:18:09.459589163:159:E:logger:1741] TID 06 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd::0/internal]: :0/internal: Broker changed state INIT -
> UP
[2019 Nov  8 19:18:09.459593374:160:E:logger:1741] TID 06 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd::0/internal]: Broadcasting state change
[2019 Nov  8 19:18:09.459608395:161:E:logger:1741] TID 07 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:main]: Broadcasting state change
[2019 Nov  8 19:18:09.459708091:162:E:logger:1741] TID 05 : [LOG_TRACE]:RDKAFKA-7-BROKER: rdkafka#consumer-2: [thrd:app]: ssl://10.0.0.1:9092/bootstrap: Added new broker with NodeId -1
[2019 Nov  8 19:18:09.459718029:163:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BRKMAIN: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Enter main broker thread
[2019 Nov  8 19:18:09.459723538:164:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECT: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: broker in state INIT connecting
[2019 Nov  8 19:18:09.459918518:165:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECT: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Connecting to ipv4#10.0.0.1:9092 (ssl) with socket 34
[2019 Nov  8 19:18:09.460017515:166:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker changed state INIT -> CONNECT
[2019 Nov  8 19:18:09.460021977:167:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.460228677:168:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECT: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Connected to ipv4#10.0.0.1:9092
[2019 Nov  8 19:18:09.790145695:169:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-SSLVERIFY: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker SSL certificate verified
[2019 Nov  8 19:18:09.790151895:170:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-CONNECTED: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Connected (#1)
[2019 Nov  8 19:18:09.790168266:171:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-FEATURE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Updated enabled protocol features +ApiVersion to ApiVersion
[2019 Nov  8 19:18:09.790172810:172:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker changed state CONNECT -> APIVERSION_QUERY
[2019 Nov  8 19:18:09.790176880:173:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.790888559:174:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-FEATURE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Updated enabled protocol features to MsgVer1,ApiVersion,BrokerBalancedConsumer,ThrottleTime,Sasl,SaslHandshake,BrokerGroupCoordinator,LZ4,OffsetTime,MsgVer2
[2019 Nov  8 19:18:09.790893525:175:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: Broker changed state APIVERSION_QUERY -> UP
[2019 Nov  8 19:18:09.790897645:176:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.791643149:177:E:logger:1741] TID 07 : [LOG_TRACE]:RDKAFKA-7-CLUSTERID: rdkafka#consumer-2: [thrd:main]: ssl://10.0.0.1:9092/bootstrap: ClusterId update "" -> "r7Us-jYGQRq34re8owKyJA"
[2019 Nov  8 19:18:09.791654890:178:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-UPDATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/bootstrap: NodeId changed from -1 to 1
[2019 Nov  8 19:18:09.791663562:179:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-UPDATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Name changed from ssl://10.0.0.1:9092/bootstrap to ssl://10.0.0.1:9092/1
[2019 Nov  8 19:18:09.791668295:180:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-LEADER: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Mapped 0 partition(s) to broker
[2019 Nov  8 19:18:09.791671709:181:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Broker changed state UP -> UPDATE
[2019 Nov  8 19:18:09.791675360:182:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change
[2019 Nov  8 19:18:09.791692544:183:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-STATE: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Broker changed state UPDATE -> UP
[2019 Nov  8 19:18:09.791696027:184:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-BROADCAST: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: Broadcasting state change

Consumer code is show below.

static void
msg_consume(rd_kafka_message_t *rkmessage, void *opaque)
{
    if (rkmessage == NULL) {
        return;
    }

    if (rkmessage->err) {
        if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
            DEBUG_PRINT(DBG_TRACE,
                        "%% Consumer reached end of %s [%"PRId32"] "
                        "message queue at offset %"PRId64"\n",
                         (rkmessage->rkt) ? 
                          rd_kafka_topic_name(rkmessage->rkt) : "NULL",
                          rkmessage->partition, rkmessage->offset);
            return;
        }

        if (rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION ||
            rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC) {
            return;
        }

        return;
    }

    if (rkmessage->key_len) {
        DEBUG_PRINT(DBG_TRACE, "Key: %.*s\n", (int)rkmessage->key_len, 
                    (char *)rkmessage->key);
    }

    DEBUG_PRINT(DBG_TRACE, "%.*s\n", (int)rkmessage->len, 
                (char *)rkmessage->payload);
}

syserr_t
kafka_consumer_create()
{
    rd_kafka_topic_conf_t *consTopicCfg;
    rd_kafka_conf_t *conf = NULL;
    rd_kafka_t *rk = NULL;
    char errstr[512];
    rd_kafka_resp_err_t errCode;

    conf = rd_kafka_conf_new();

    if (!conf) {
        return ~SUCCESS;
    }

    rd_kafka_conf_set_log_cb(conf, logger);

    if (rd_kafka_conf_set(conf, "debug", 
                          "generic,broker,topic,security,msg,fetch",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
    }

    if (rd_kafka_conf_set(conf, "bootstrap.servers", broker_list[0],
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "security.protocol", "SSL",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.certificate.location", kafka_clnt_cert,
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.key.location", kafka_clnt_key,
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.ca.location", kafka_apic_cert,
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "auto.commit.enable", "true",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "auto.commit.interval.ms", "500",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "group.id", "consumerGroup",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    consTopicCfg = rd_kafka_topic_conf_new();

    if (RD_KAFKA_CONF_OK != rd_kafka_topic_conf_set(consTopicCfg, 
                                                    "auto.offset.reset",
                                                    "latest" ,errstr,
                                                    sizeof(errstr))) {
        return ~SUCCESS;
    }

    rd_kafka_conf_set_default_topic_conf(conf, consT opicCfg);
    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (!rk) {
        return ~SUCCESS;
    }

//    conf = NULL; // Disown conf as rd_kafka_new() has ownership now.

    const char *ep_topic="eprecords";
    rd_kafka_topic_partition_list_t *tp_list = 
                            rd_kafka_topic_partition_list_new(1);
    rd_kafka_topic_partition_t* tpObj = 
                            rd_kafka_topic_partition_list_add(tp_list,
                                                              ep_topic, 
                                                      RD_KAFKA_PARTITION_UA);

    if (NULL == tpObj) {
        return ~SUCCESS;
    }

    errCode = rd_kafka_subscribe(rk, tp_list);
    if (errCode  != RD_KAFKA_RESP_ERR_NO_ERROR) {
        return ~SUCCESS;
    }

    rd_kafka_topic_partition_list_destroy(tp_list);

    while(1) {
        rd_kafka_message_t *msg = rd_kafka_consumer_poll(rk, 1000);
        if (msg != NULL) {
            if (msg->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
                msg_consume(msg, NULL);
            }
            rd_kafka_message_destroy(msg);
        }
        rd_kafka_poll(rk,0);
    }
}

I am expecting that msg_consume() will be called as producer is publishing data periodically. I am not sure if the below log message is root of the problem.

"[2019 Nov 8 19:18:09.791668295:180:E:logger:1741] TID 08 : [LOG_TRACE]:RDKAFKA-7-LEADER: rdkafka#consumer-2: [thrd:ssl://10.0.0.1:9092/bootstrap]: ssl://10.0.0.1:9092/1: Mapped 0 partition(s) to broker"

1
capache-kafkalibrdkafka

1 Answers

0
votes

Program started working after I had shifted to a different model of consumer. Code below.

static void logger (const rd_kafka_t *rk, int level,
                    const char *fac, const char *buf)
{
    PRINT_ERR("RDKAFKA-%i-%s: %s: %s", level, fac, rk ? rd_kafka_name(rk) : NULL, buf);
}

rd_kafka_t *
kafka_consumer_create()
{
    rd_kafka_conf_t *conf = NULL;
    rd_kafka_t *rk = NULL;
    char errstr[512];
    rd_kafka_resp_err_t errCode;

    conf = rd_kafka_conf_new();

    if (!conf) {
        return ~SUCCESS;
    }

    rd_kafka_conf_set_log_cb(conf, logger);

    if (rd_kafka_conf_set(conf, "enable.partition.eof", "true",
                          errstr, sizeof(errstr) != RD_KAFKA_CONF_OK)) {
        PRINT_ERR("enable partition eof failed %s", errstr);
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "client.id", "kafka-python-Python 2.7.16",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        PRINT_ERR("DR_MSG : failed to set client id %s", errstr);
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "debug", "generic,broker,topic,security,msg,fetch",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "enable.auto.commit","true",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "bootstrap.servers", "192.168.1.1",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "security.protocol", "SSL",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.certificate.location", "/root/kafka/KafkaClient.crt",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.key.location", "/root/kafka/KafkaClient8.key",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "ssl.ca.location", "/root/kafka/ApicCa.crt",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "group.id", "test-consumer-group",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    if (rd_kafka_conf_set(conf, "auto.offset.reset", "latest",
                          errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
        return ~SUCCESS;
    }

    rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
    if (!rk) {
        return ~SUCCESS;
    }

    conf = NULL; // Disown conf as rd_kafka_new() has ownership now.

    return rk;
}

static void
msg_consume(rd_kafka_message_t *rkmessage, void *opaque)
{
    if (rkmessage == NULL) {
        PRINT_ERR("Aentp rkmessage error");
        return;
    }

    if (rkmessage->err) {
        if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
            PRINT_ERR("AT2");
            DEBUG_PRINT(DBG_AENTP_TRACE,
                        "%% Consumer reached end of %s [%"PRId32"] "
                        "message queue at offset %"PRId64"\n",
                        (rkmessage->rkt) ? rd_kafka_topic_name(rkmessage->rkt) : "NULL",
                        rkmessage->partition, rkmessage->offset);
            return;
        }

        if (rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION ||
            rkmessage->err == RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC) {
            PRINT_ERR("Unknown partition or unknow topic");
            return;
        }

        return;
    }

    if (rkmessage->key_len) {
        DEBUG_PRINT("Key: %.*s\n", (int)rkmessage->key_len, (char *)rkmessage->key);
    }

    DEBUG_PRINT("%.*s\n", (int)rkmessage->len, (char *)rkmessage->payload);
}

void *
kafka_consumer_thread(void *ptr)
{
    rd_kafka_t *rk = kafka_consumer_create();
    rd_kafka_topic_t *rkt = NULL;
    rd_kafka_topic_conf_t *topic_conf;
    rd_kafka_resp_err_t err;
    char errstr[512];

    if (rk == NULL) {        
        return NULL;
    }

    DEBUG_PRINT("Starting Kafka consumer thread");
    topic_conf = rd_kafka_topic_conf_new();
    if (RD_KAFKA_CONF_OK != rd_kafka_topic_conf_set(topic_conf, "auto.offset.reset",
                            "earliest" ,errstr, sizeof(errstr))) {
        AENTP_PRINT_ERR("rd_kafka_topic_conf_set() failed with error: %s\n", errstr);
        return NULL;
    }

    rkt = rd_kafka_topic_new(rk, "records", topic_conf);
    topic_conf = NULL; /* Now owned by topic */

    /* Start consuming */
    if (rd_kafka_consume_start(rkt, 0, RD_KAFKA_OFFSET_END) == -1){
        rd_kafka_resp_err_t err = rd_kafka_last_error();
        PRINT_ERR("%% Failed to start consuming: %s\n", rd_kafka_err2str(err));
        if (err == RD_KAFKA_RESP_ERR__INVALID_ARG) {
            APRINT_ERR("%% Broker based offset storage "
                       "requires a group.id, "
                       "add: -X group.id=yourGroup\n");
        }
        return NULL;
    }

    while(1) {
        rd_kafka_message_t *rkmessage;

        /* Poll for errors, etc. */
        rd_kafka_poll(rk, 0);

        /*
         * Consume single message.
         * See rdkafka_performance.c for high speed
         * consuming of messages.
         */
        rkmessage = rd_kafka_consume(rkt, 0, 1000);
        if (!rkmessage) /* timeout */
            continue;

        msg_consume(rkmessage, NULL);

        /* Return message to rdkafka */
        rd_kafka_message_destroy(rkmessage);

        /* XXX Do we need the seek??? */
        seek_offset = 0;
        if (seek_offset) {
            err = rd_kafka_seek(rkt, 0, seek_offset, 2000);
            if (err)
                AENTP_PRINT_ERR("Seek failed: %s\n", rd_kafka_err2str(err));
            else
                printf("Seeked to %"PRId64"\n", seek_offset);
            seek_offset = 0;
        }
        /* XXX Do we need the seek??? */
    }

    return NULL;
}

Note sure why the previous method does not work.