/* * librdkafka - Apache Kafka C library * * Copyright (c) 2012-2022, Magnus Edenhill * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * 1. Redistributions of source code must retain the above copyright notice, * this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE * POSSIBILITY OF SUCH DAMAGE. */ /** * Produce messages, then consume them. * Consume both through the standard interface and through the queue interface. */ #include "test.h" /* Typical include path would be , but this program * is built from within the librdkafka source tree and thus differs. */ #include "rdkafka.h" /* for Kafka driver */ static int prod_msg_remains = 0; static int fails = 0; /** * Delivery reported callback. * Called for each message once to signal its delivery status. */ static void dr_cb(rd_kafka_t *rk, void *payload, size_t len, rd_kafka_resp_err_t err, void *opaque, void *msg_opaque) { if (err != RD_KAFKA_RESP_ERR_NO_ERROR) TEST_FAIL("Message delivery failed: %s\n", rd_kafka_err2str(err)); if (prod_msg_remains == 0) TEST_FAIL("Too many messages delivered (prod_msg_remains %i)", prod_msg_remains); prod_msg_remains--; } /** * Produces 'msgcnt' messages split over 'partition_cnt' partitions. */ static void produce_messages(uint64_t testid, const char *topic, int partition_cnt, int msgcnt) { int r; rd_kafka_t *rk; rd_kafka_topic_t *rkt; rd_kafka_conf_t *conf; rd_kafka_topic_conf_t *topic_conf; char errstr[512]; char msg[128]; int failcnt = 0; int i; rd_kafka_message_t *rkmessages; int32_t partition; int msgid = 0; test_conf_init(&conf, &topic_conf, 20); rd_kafka_conf_set_dr_cb(conf, dr_cb); /* Make sure all replicas are in-sync after producing * so that consume test wont fail. */ rd_kafka_topic_conf_set(topic_conf, "request.required.acks", "-1", errstr, sizeof(errstr)); /* Create kafka instance */ rk = test_create_handle(RD_KAFKA_PRODUCER, conf); rkt = rd_kafka_topic_new(rk, topic, topic_conf); if (!rkt) TEST_FAIL("Failed to create topic: %s\n", rd_strerror(errno)); /* Create messages. */ prod_msg_remains = msgcnt; rkmessages = calloc(sizeof(*rkmessages), msgcnt / partition_cnt); for (partition = 0; partition < partition_cnt; partition++) { int batch_cnt = msgcnt / partition_cnt; for (i = 0; i < batch_cnt; i++) { rd_snprintf(msg, sizeof(msg), "testid=%" PRIu64 ", partition=%i, msg=%i", testid, (int)partition, msgid); rkmessages[i].payload = rd_strdup(msg); rkmessages[i].len = strlen(msg); msgid++; } TEST_SAY("Start produce to partition %i: msgs #%d..%d\n", (int)partition, msgid - batch_cnt, msgid); /* Produce batch for this partition */ r = rd_kafka_produce_batch(rkt, partition, RD_KAFKA_MSG_F_FREE, rkmessages, batch_cnt); if (r == -1) TEST_FAIL( "Failed to produce " "batch for partition %i: %s", (int)partition, rd_kafka_err2str(rd_kafka_last_error())); /* Scan through messages to check for errors. */ for (i = 0; i < batch_cnt; i++) { if (rkmessages[i].err) { failcnt++; if (failcnt < 100) TEST_SAY("Message #%i failed: %s\n", i, rd_kafka_err2str( rkmessages[i].err)); } } /* All messages should've been produced. */ if (r < batch_cnt) { TEST_SAY( "Not all messages were accepted " "by produce_batch(): %i < %i\n", r, batch_cnt); if (batch_cnt - r != failcnt) TEST_SAY( "Discrepency between failed " "messages (%i) " "and return value %i (%i - %i)\n", failcnt, batch_cnt - r, batch_cnt, r); TEST_FAIL("%i/%i messages failed\n", batch_cnt - r, batch_cnt); } TEST_SAY( "Produced %i messages to partition %i, " "waiting for deliveries\n", r, partition); } free(rkmessages); /* Wait for messages to be delivered */ while (rd_kafka_outq_len(rk) > 0) rd_kafka_poll(rk, 100); if (fails) TEST_FAIL("%i failures, see previous errors", fails); if (prod_msg_remains != 0) TEST_FAIL("Still waiting for %i messages to be produced", prod_msg_remains); /* Destroy topic */ rd_kafka_topic_destroy(rkt); /* Destroy rdkafka instance */ TEST_SAY("Destroying kafka instance %s\n", rd_kafka_name(rk)); rd_kafka_destroy(rk); } static int *cons_msgs; static int cons_msgs_size; static int cons_msgs_cnt; static void verify_consumed_msg_reset(int msgcnt) { TEST_SAY("Resetting consumed_msgs (msgcnt %d)\n", msgcnt); if (cons_msgs) { free(cons_msgs); cons_msgs = NULL; } if (msgcnt) { int i; cons_msgs = malloc(sizeof(*cons_msgs) * msgcnt); for (i = 0; i < msgcnt; i++) cons_msgs[i] = -1; } cons_msgs_size = msgcnt; cons_msgs_cnt = 0; } static int int_cmp(const void *_a, const void *_b) { int a = *(int *)_a; int b = *(int *)_b; return RD_CMP(a, b); } static void verify_consumed_msg_check0(const char *func, int line) { int i; int fails = 0; if (cons_msgs_cnt < cons_msgs_size) { TEST_SAY("Missing %i messages in consumer\n", cons_msgs_size - cons_msgs_cnt); fails++; } qsort(cons_msgs, cons_msgs_size, sizeof(*cons_msgs), int_cmp); for (i = 0; i < cons_msgs_size; i++) { if (cons_msgs[i] != i) { TEST_SAY( "Consumed message #%i is wrong, " "expected #%i\n", cons_msgs[i], i); fails++; } } if (fails) TEST_FAIL("See above error(s)"); verify_consumed_msg_reset(0); } #define verify_consumed_msg_check() \ verify_consumed_msg_check0(__FUNCTION__, __LINE__) static void verify_consumed_msg0(const char *func, int line, uint64_t testid, int32_t partition, int msgnum, rd_kafka_message_t *rkmessage) { uint64_t in_testid; int in_part; int in_msgnum; char buf[1024]; if (rkmessage->len + 1 >= sizeof(buf)) TEST_FAIL( "Incoming message too large (%i): " "not sourced by this test", (int)rkmessage->len); rd_snprintf(buf, sizeof(buf), "%.*s", (int)rkmessage->len, (char *)rkmessage->payload); if (sscanf(buf, "testid=%" SCNu64 ", partition=%i, msg=%i", &in_testid, &in_part, &in_msgnum) != 3) TEST_FAIL("Incorrect message format: %s", buf); if (test_level > 2) { TEST_SAY("%s:%i: Our testid %" PRIu64 ", part %i =? %i, " "msg %i =? %i " ", message's: \"%s\"\n", func, line, testid, (int)partition, (int)rkmessage->partition, msgnum, in_msgnum, buf); } if (testid != in_testid || (partition != -1 && partition != in_part) || (msgnum != -1 && msgnum != in_msgnum) || (in_msgnum < 0 || in_msgnum > cons_msgs_size)) goto fail_match; if (cons_msgs_cnt == cons_msgs_size) { TEST_SAY( "Too many messages in cons_msgs (%i) while reading " "message \"%s\"\n", cons_msgs_cnt, buf); verify_consumed_msg_check(); TEST_FAIL("See above error(s)"); } cons_msgs[cons_msgs_cnt++] = in_msgnum; return; fail_match: TEST_FAIL("%s:%i: Our testid %" PRIu64 ", part %i, msg %i/%i did " "not match message's: \"%s\"\n", func, line, testid, (int)partition, msgnum, cons_msgs_size, buf); } #define verify_consumed_msg(testid, part, msgnum, rkmessage) \ verify_consumed_msg0(__FUNCTION__, __LINE__, testid, part, msgnum, \ rkmessage) static void consume_messages(uint64_t testid, const char *topic, int32_t partition, int msg_base, int batch_cnt, int msgcnt) { rd_kafka_t *rk; rd_kafka_topic_t *rkt; rd_kafka_conf_t *conf; rd_kafka_topic_conf_t *topic_conf; int i; test_conf_init(&conf, &topic_conf, 20); /* Create kafka instance */ rk = test_create_handle(RD_KAFKA_CONSUMER, conf); TEST_SAY("Created kafka instance %s\n", rd_kafka_name(rk)); rkt = rd_kafka_topic_new(rk, topic, topic_conf); if (!rkt) TEST_FAIL("Failed to create topic: %s\n", rd_strerror(errno)); TEST_SAY("Consuming %i messages from partition %i\n", batch_cnt, partition); /* Consume messages */ if (rd_kafka_consume_start(rkt, partition, RD_KAFKA_OFFSET_TAIL(batch_cnt)) == -1) TEST_FAIL("consume_start(%i, -%i) failed: %s", (int)partition, batch_cnt, rd_kafka_err2str(rd_kafka_last_error())); for (i = 0; i < batch_cnt;) { rd_kafka_message_t *rkmessage; rkmessage = rd_kafka_consume(rkt, partition, tmout_multip(5000)); if (!rkmessage) TEST_FAIL( "Failed to consume message %i/%i from " "partition %i: %s", i, batch_cnt, (int)partition, rd_kafka_err2str(rd_kafka_last_error())); if (rkmessage->err) { if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { rd_kafka_message_destroy(rkmessage); continue; } TEST_FAIL( "Consume message %i/%i from partition %i " "has error: %s: %s", i, batch_cnt, (int)partition, rd_kafka_err2str(rkmessage->err), rd_kafka_message_errstr(rkmessage)); } verify_consumed_msg(testid, partition, msg_base + i, rkmessage); rd_kafka_message_destroy(rkmessage); i++; } rd_kafka_consume_stop(rkt, partition); /* Destroy topic */ rd_kafka_topic_destroy(rkt); /* Destroy rdkafka instance */ TEST_SAY("Destroying kafka instance %s\n", rd_kafka_name(rk)); rd_kafka_destroy(rk); } static void consume_messages_with_queues(uint64_t testid, const char *topic, int partition_cnt, int msgcnt) { rd_kafka_t *rk; rd_kafka_topic_t *rkt; rd_kafka_conf_t *conf; rd_kafka_topic_conf_t *topic_conf; rd_kafka_queue_t *rkqu; int i; int32_t partition; int batch_cnt = msgcnt / partition_cnt; test_conf_init(&conf, &topic_conf, 20); test_conf_set(conf, "enable.partition.eof", "true"); /* Create kafka instance */ rk = test_create_handle(RD_KAFKA_CONSUMER, conf); /* Create queue */ rkqu = rd_kafka_queue_new(rk); rkt = rd_kafka_topic_new(rk, topic, topic_conf); if (!rkt) TEST_FAIL("Failed to create topic: %s\n", rd_strerror(errno)); TEST_SAY("Consuming %i messages from one queue serving %i partitions\n", msgcnt, partition_cnt); /* Start consuming each partition */ for (partition = 0; partition < partition_cnt; partition++) { /* Consume messages */ TEST_SAY("Start consuming partition %i at offset -%i\n", partition, batch_cnt); if (rd_kafka_consume_start_queue( rkt, partition, RD_KAFKA_OFFSET_TAIL(batch_cnt), rkqu) == -1) TEST_FAIL("consume_start_queue(%i) failed: %s", (int)partition, rd_kafka_err2str(rd_kafka_last_error())); } /* Consume messages from queue */ for (i = 0; i < msgcnt;) { rd_kafka_message_t *rkmessage; rkmessage = rd_kafka_consume_queue(rkqu, tmout_multip(5000)); if (!rkmessage) TEST_FAIL( "Failed to consume message %i/%i from " "queue: %s", i, msgcnt, rd_kafka_err2str(rd_kafka_last_error())); if (rkmessage->err) { if (rkmessage->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { TEST_SAY("Topic %s [%" PRId32 "] reached " "EOF at offset %" PRId64 "\n", rd_kafka_topic_name(rkmessage->rkt), rkmessage->partition, rkmessage->offset); rd_kafka_message_destroy(rkmessage); continue; } TEST_FAIL( "Consume message %i/%i from queue " "has error (offset %" PRId64 ", partition %" PRId32 "): %s", i, msgcnt, rkmessage->offset, rkmessage->partition, rd_kafka_err2str(rkmessage->err)); } verify_consumed_msg(testid, -1, -1, rkmessage); rd_kafka_message_destroy(rkmessage); i++; } /* Stop consuming each partition */ for (partition = 0; partition < partition_cnt; partition++) rd_kafka_consume_stop(rkt, partition); /* Destroy queue */ rd_kafka_queue_destroy(rkqu); /* Destroy topic */ rd_kafka_topic_destroy(rkt); /* Destroy rdkafka instance */ TEST_SAY("Destroying kafka instance %s\n", rd_kafka_name(rk)); rd_kafka_destroy(rk); } /** * Produce to two partitions. * Consume with standard interface from both, one after the other. * Consume with queue interface from both, simultanously. */ static void test_produce_consume(void) { int msgcnt = test_quick ? 100 : 1000; int partition_cnt = 2; int i; uint64_t testid; int msg_base = 0; const char *topic; /* Generate a testid so we can differentiate messages * from other tests */ testid = test_id_generate(); /* Read test.conf to configure topic name */ test_conf_init(NULL, NULL, 20); topic = test_mk_topic_name("0012", 1); TEST_SAY("Topic %s, testid %" PRIu64 "\n", topic, testid); /* Produce messages */ produce_messages(testid, topic, partition_cnt, msgcnt); /* Consume messages with standard interface */ verify_consumed_msg_reset(msgcnt); for (i = 0; i < partition_cnt; i++) { consume_messages(testid, topic, i, msg_base, msgcnt / partition_cnt, msgcnt); msg_base += msgcnt / partition_cnt; } verify_consumed_msg_check(); /* Consume messages with queue interface */ verify_consumed_msg_reset(msgcnt); consume_messages_with_queues(testid, topic, partition_cnt, msgcnt); verify_consumed_msg_check(); return; } int main_0012_produce_consume(int argc, char **argv) { test_produce_consume(); return 0; }