我已经使用librdkafka库用C编写了kafka消费者和生产者。Kafka broker版本为Kafka_2.12-2.3.0。生产者正在成功生成消息，dr_msg_cb功能确认成功交付。然而，消费者并没有收到来自代理的消息。有人能帮忙进一步调试吗？
我可以看到，从消费者到代理的TCP连接已经建立。但TCPdump显示broker并没有向消费者发送任何数据。我对使用者代码启用了调试，下面是消息。
[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
消费者代码如下所示。
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: %.*sn", (int)rkmessage->key_len,
(char *)rkmessage->key);
}
DEBUG_PRINT(DBG_TRACE, "%.*sn", (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);
}
}
我期望在生产者定期发布数据时调用msg_consume((。我不确定下面的日志消息是否是问题的根源。
"【2019年11月8日19:18:09.979668295: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:已将0个分区映射到代理">
程序在我转换到不同的消费者模型后开始工作。下面的代码。
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: %.*sn", (int)rkmessage->key_len, (char *)rkmessage->key);
}
DEBUG_PRINT("%.*sn", (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: %sn", 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: %sn", 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=yourGroupn");
}
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: %sn", rd_kafka_err2str(err));
else
printf("Seeked to %"PRId64"n", seek_offset);
seek_offset = 0;
}
/* XXX Do we need the seek??? */
}
return NULL;
}
请务必注意为什么前面的方法不起作用。