K2 testing final

tests/0001-multiobj.c

@@ -60,6 +60,8 @@ int main_0001_multiobj(int argc, char **argv) {
                 if (!topic)
                         topic = test_mk_topic_name("0001", 0);
 
+                test_create_topic_if_auto_create_disabled(NULL, topic, -1);
+
                 TIMING_START(&t_full, "full create-produce-destroy cycle");
                 rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
@@ -92,7 +94,7 @@ int main_0001_multiobj(int argc, char **argv) {
 
                 /* Topic is created on the first iteration. */
                 if (i > 0)
-                        TIMING_ASSERT(&t_full, 0, 999);
+                        TIMING_ASSERT(&t_full, 0, tmout_multip(999));
                 else
                         /* Allow metadata propagation. */
                         rd_sleep(1);

tests/0002-unkpart.c

@@ -83,6 +83,7 @@ static void do_test_unkpart(void) {
         int i;
         int fails = 0;
         const struct rd_kafka_metadata *metadata;
+        const char *topic;
 
         TEST_SAY(_C_BLU "%s\n" _C_CLR, __FUNCTION__);
 
@@ -94,7 +95,10 @@ static void do_test_unkpart(void) {
         /* Create kafka instance */
         rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
-        rkt = rd_kafka_topic_new(rk, test_mk_topic_name("0002", 0), topic_conf);
+        topic = test_mk_topic_name("0002", 0);
+        test_create_topic_if_auto_create_disabled(rk, topic, 3);
+
+        rkt = rd_kafka_topic_new(rk, topic, topic_conf);
         if (!rkt)
                 TEST_FAIL("Failed to create topic: %s\n",
                           rd_kafka_err2str(rd_kafka_last_error()));
@@ -199,7 +203,9 @@ static void do_test_unkpart_timeout_nobroker(void) {
         rd_kafka_conf_set_dr_msg_cb(conf, test_dr_msg_cb);
         test_curr->exp_dr_err = RD_KAFKA_RESP_ERR__MSG_TIMED_OUT;
 
-        rk  = test_create_handle(RD_KAFKA_PRODUCER, conf);
+        rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
+
+        test_create_topic_if_auto_create_disabled(NULL, topic, 3);
         rkt = rd_kafka_topic_new(rk, topic, NULL);
 
         err = rd_kafka_produce(rkt, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_COPY,

tests/0003-msgmaxsize.c

@@ -79,6 +79,7 @@ int main_0003_msgmaxsize(int argc, char **argv) {
         rd_kafka_conf_t *conf;
         rd_kafka_topic_conf_t *topic_conf;
         char errstr[512];
+        const char *topic;
 
         static const struct {
                 ssize_t keylen;
@@ -108,7 +109,10 @@ int main_0003_msgmaxsize(int argc, char **argv) {
         /* Create kafka instance */
         rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
-        rkt = rd_kafka_topic_new(rk, test_mk_topic_name("0003", 0), topic_conf);
+        topic = test_mk_topic_name("0003", 0);
+        test_create_topic_if_auto_create_disabled(NULL, topic, -1);
+        rkt = rd_kafka_topic_new(rk, topic, topic_conf);
+
         if (!rkt)
                 TEST_FAIL("Failed to create topic: %s\n", rd_strerror(errno));
 

tests/0004-conf.c

@@ -511,26 +511,19 @@ int main_0004_conf(int argc, char **argv) {
         int i;
         const char *topic;
         static const char *gconfs[] = {
-            "message.max.bytes",
-            "12345", /* int property */
-            "client.id",
-            "my id", /* string property */
-            "debug",
-            "topic,metadata,interceptor", /* S2F property */
-            "topic.blacklist",
-            "__.*", /* #778 */
-            "auto.offset.reset",
-            "earliest", /* Global->Topic fallthru */
+            "message.max.bytes", "12345",          /* int property */
+            "client.id", "my id",                  /* string property */
+            "debug", "topic,metadata,interceptor", /* S2F property */
+            "topic.blacklist", "__.*",             /* #778 */
+            "auto.offset.reset", "earliest",       /* Global->Topic fallthru */
 #if WITH_ZLIB
-            "compression.codec",
-            "gzip", /* S2I property */
+            "compression.codec", "gzip", /* S2I property */
 #endif
 #if defined(_WIN32)
-            "ssl.ca.certificate.stores",
-            "Intermediate ,, Root ,",
+            "ssl.ca.certificate.stores", "Intermediate ,, Root ,",
 #endif
-            "client.dns.lookup",
-            "resolve_canonical_bootstrap_servers_only",
+            /* client.dns.lookup was introduced in librdkafka 2.2.0+ - skip
+               for 2.1.x library */
             NULL};
         static const char *tconfs[] = {"request.required.acks",
                                        "-1", /* int */
@@ -557,6 +550,15 @@ int main_0004_conf(int argc, char **argv) {
                         TEST_FAIL("%s\n", errstr);
         }
 
+        /* Add client.dns.lookup if librdkafka version >= 2.2.0 */
+        if (rd_kafka_version() >= 0x02020000) {
+                if (rd_kafka_conf_set(
+                        conf, "client.dns.lookup",
+                        "resolve_canonical_bootstrap_servers_only", errstr,
+                        sizeof(errstr)) != RD_KAFKA_CONF_OK)
+                        TEST_FAIL("%s\n", errstr);
+        }
+
         rd_kafka_conf_set_dr_cb(conf, dr_cb);
         rd_kafka_conf_set_error_cb(conf, error_cb);
         /* interceptor configs are not exposed as strings or in dumps
@@ -721,69 +723,87 @@ int main_0004_conf(int argc, char **argv) {
         }
 
 #if WITH_OAUTHBEARER_OIDC
-        {
-                TEST_SAY(
-                    "Verify that https.ca.location is mutually "
-                    "exclusive with https.ca.pem\n");
+        /* HTTPS CA configuration tests - https.ca.pem available since
+         * librdkafka 2.2.0 */
+        if (rd_kafka_version() >= 0x02020000) {
+                {
+                        TEST_SAY(
+                            "Verify that https.ca.location is mutually "
+                            "exclusive with https.ca.pem\n");
 
-                conf = rd_kafka_conf_new();
+                        conf = rd_kafka_conf_new();
 
-                test_conf_set(conf, "https.ca.pem",
-                              "-----BEGIN CERTIFICATE-----");
-                test_conf_set(conf, "https.ca.location",
-                              "/path/to/certificate.pem");
+                        test_conf_set(conf, "https.ca.pem",
+                                      "-----BEGIN CERTIFICATE-----");
+                        test_conf_set(conf, "https.ca.location",
+                                      "/path/to/certificate.pem");
 
-                rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr,
-                                  sizeof(errstr));
-                TEST_ASSERT(
-                    !rk, "Expected rd_kafka_new() to fail, but it succeeded");
-                TEST_ASSERT(!strcmp(errstr,
+                        rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr,
+                                          sizeof(errstr));
+                        TEST_ASSERT(!rk,
+                                    "Expected rd_kafka_new() to fail, but it "
+                                    "succeeded");
+                        TEST_ASSERT(
+                            !strcmp(errstr,
                                     "`https.ca.location` and "
                                     "`https.ca.pem` are mutually exclusive"),
                             "Expected rd_kafka_new() to fail with: "
                             "\"`https.ca.location` and `https.ca.pem` "
                             "are mutually exclusive\", got: \"%s\"",
                             errstr);
-                rd_kafka_conf_destroy(conf);
+                        rd_kafka_conf_destroy(conf);
+                }
         }
-        {
-                TEST_SAY(
-                    "Verify that https.ca.location gives an error when "
-                    "set to an invalid path\n");
+        if (rd_kafka_version() >= 0x02020000) { /* https.ca.location available
+                                                   since librdkafka 2.2.0 */
+                {
+                        TEST_SAY(
+                            "Verify that https.ca.location gives an error when "
+                            "set to an invalid path\n");
 
-                conf = rd_kafka_conf_new();
+                        conf = rd_kafka_conf_new();
 
-                test_conf_set(conf, "https.ca.location",
-                              "/?/does/!/not/exist!");
+                        test_conf_set(conf, "https.ca.location",
+                                      "/?/does/!/not/exist!");
 
-                rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr,
-                                  sizeof(errstr));
-                TEST_ASSERT(
-                    !rk, "Expected rd_kafka_new() to fail, but it succeeded");
-                TEST_ASSERT(!strcmp(errstr,
+                        rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr,
+                                          sizeof(errstr));
+                        TEST_ASSERT(!rk,
+                                    "Expected rd_kafka_new() to fail, but it "
+                                    "succeeded");
+                        TEST_ASSERT(
+                            !strcmp(errstr,
                                     "`https.ca.location` must be "
                                     "an existing file or directory"),
                             "Expected rd_kafka_new() to fail with: "
                             "\"`https.ca.location` must be "
                             "an existing file or directory\", got: \"%s\"",
                             errstr);
-                rd_kafka_conf_destroy(conf);
-        }
-        {
-                TEST_SAY(
-                    "Verify that https.ca.location doesn't give an error when "
-                    "set to `probe`\n");
+                        rd_kafka_conf_destroy(conf);
+                }
+                {
+                        TEST_SAY(
+                            "Verify that https.ca.location doesn't give an "
+                            "error when "
+                            "set to `probe`\n");
 
-                conf = rd_kafka_conf_new();
+                        conf = rd_kafka_conf_new();
 
-                test_conf_set(conf, "https.ca.location", "probe");
+                        test_conf_set(conf, "https.ca.location", "probe");
 
-                rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr,
-                                  sizeof(errstr));
-                TEST_ASSERT(
-                    rk, "Expected rd_kafka_new() not to fail, but it failed");
+                        rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr,
+                                          sizeof(errstr));
+                        TEST_ASSERT(rk,
+                                    "Expected rd_kafka_new() not to fail, but "
+                                    "it failed");
 
-                rd_kafka_destroy(rk);
+                        rd_kafka_destroy(rk);
+                }
+        } else {
+                TEST_SAY(
+                    "SKIPPING: https.ca.location tests - requires librdkafka "
+                    "version >= 2.2.0 (current: 0x%08x)\n",
+                    rd_kafka_version());
         }
 #endif /* WITH_OAUTHBEARER_OIDC */
 

tests/0005-order.c

@@ -80,6 +80,7 @@ int main_0005_order(int argc, char **argv) {
         int msgcnt = test_quick ? 500 : 50000;
         int i;
         test_timing_t t_produce, t_delivery;
+        const char *topic;
 
         test_conf_init(&conf, &topic_conf, 10);
 
@@ -89,7 +90,9 @@ int main_0005_order(int argc, char **argv) {
         /* Create kafka instance */
         rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
-        rkt = rd_kafka_topic_new(rk, test_mk_topic_name("0005", 0), topic_conf);
+        topic = test_mk_topic_name("0005", 0);
+        test_create_topic_if_auto_create_disabled(rk, topic, 1);
+        rkt = rd_kafka_topic_new(rk, topic, topic_conf);
         if (!rkt)
                 TEST_FAIL("Failed to create topic: %s\n", rd_strerror(errno));
 

tests/0007-autotopic.c

@@ -85,22 +85,25 @@ int main_0007_autotopic(int argc, char **argv) {
         int msgcnt = 10;
         int i;
 
+        if (!test_check_auto_create_topic()) {
+                TEST_SKIP(
+                    "NOTE! This test requires "
+                    "auto.create.topics.enable=true to be configured on "
+                    "the broker!\n");
+                return 0;
+        }
+
         /* Generate unique topic name */
         test_conf_init(&conf, &topic_conf, 10);
 
-        TEST_SAY(
-            "\033[33mNOTE! This test requires "
-            "auto.create.topics.enable=true to be configured on "
-            "the broker!\033[0m\n");
-
         /* Set delivery report callback */
         rd_kafka_conf_set_dr_cb(conf, dr_cb);
 
         /* Create kafka instance */
         rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
-        rkt = rd_kafka_topic_new(rk, test_mk_topic_name("0007_autotopic", 1),
-                                 topic_conf);
+        const char *topic = test_mk_topic_name("0007_autotopic", 1);
+        rkt               = rd_kafka_topic_new(rk, topic, topic_conf);
         if (!rkt)
                 TEST_FAIL("Failed to create topic: %s\n", rd_strerror(errno));
 

tests/0008-reqacks.c

@@ -96,9 +96,23 @@ int main_0008_reqacks(int argc, char **argv) {
             "all brokers!\033[0m\n");
 
         /* Try different request.required.acks settings (issue #75) */
-        for (reqacks = -1; reqacks <= 1; reqacks++) {
+        /* Test all standard acks values, but skip unsupported ones */
+        int start_acks = -1;
+        int end_acks   = 1;
+
+        TEST_SAY("Testing acks values -1, 0, 1 (skipping unsupported ones)\n");
+        for (reqacks = start_acks; reqacks <= end_acks; reqacks++) {
                 char tmp[10];
 
+                /* Convert acks value to string and check if supported */
+                rd_snprintf(tmp, sizeof(tmp), "%d", reqacks);
+                if (!test_is_acks_supported(tmp)) {
+                        TEST_SAY(
+                            "Skipping acks=%d (not supported by cluster)\n",
+                            reqacks);
+                        continue;
+                }
+
                 test_conf_init(&conf, &topic_conf, 10);
 
                 if (reqacks != -1)
@@ -130,6 +144,8 @@ int main_0008_reqacks(int argc, char **argv) {
                     "expecting status %d\n",
                     rd_kafka_name(rk), reqacks, exp_status);
 
+                test_create_topic_if_auto_create_disabled(rk, topic, 1);
+
                 rkt = rd_kafka_topic_new(rk, topic, topic_conf);
                 if (!rkt)
                         TEST_FAIL("Failed to create topic: %s\n",