Skip certain tests if broker version does not support required features

tests/0070-null_empty.cpp

@@ -181,7 +181,8 @@ static void do_test_null_empty (bool api_version_request) {
 
 extern "C" {
   int main_0070_null_empty (int argc, char **argv) {
-    do_test_null_empty(true);
+    if (test_broker_version >= TEST_BRKVER(0,10,0,0))
+      do_test_null_empty(true);
     do_test_null_empty(false);
     return 0;
   }

tests/0076-produce_retry.c

@@ -306,22 +306,28 @@ static void do_test_produce_retries_disconnect (const char *topic,
 
 int main_0076_produce_retry (int argc, char **argv) {
         const char *topic = test_mk_topic_name("0076_produce_retry", 1);
+        const rd_bool_t has_idempotence =
+                test_broker_version >= TEST_BRKVER(0,11,0,0);
 
 #if WITH_SOCKEM
-        /* Idempotence, no try fail, should succeed. */
-        do_test_produce_retries(topic, 1, 0, 0);
+        if (has_idempotence) {
+                /* Idempotence, no try fail, should succeed. */
+                do_test_produce_retries(topic, 1, 0, 0);
+                /* Idempotence, try fail, should succeed. */
+                do_test_produce_retries(topic, 1, 1, 0);
+        }
         /* No idempotence, try fail, should fail. */
         do_test_produce_retries(topic, 0, 1, 1);
-        /* Idempotence, try fail, should succeed. */
-        do_test_produce_retries(topic, 1, 1, 0);
 #endif
 
-        /* Idempotence, no try fail, should succeed. */
-        do_test_produce_retries_disconnect(topic, 1, 0, 0);
+        if (has_idempotence) {
+                /* Idempotence, no try fail, should succeed. */
+                do_test_produce_retries_disconnect(topic, 1, 0, 0);
+                /* Idempotence, try fail, should succeed. */
+                do_test_produce_retries_disconnect(topic, 1, 1, 0);
+        }
         /* No idempotence, try fail, should fail. */
         do_test_produce_retries_disconnect(topic, 0, 1, 1);
-        /* Idempotence, try fail, should succeed. */
-        do_test_produce_retries_disconnect(topic, 1, 1, 0);
 
         return 0;
 }

tests/0081-admin.c

@@ -1169,9 +1169,13 @@ static void do_test_apis (rd_kafka_type_t cltype) {
         do_test_DeleteTopics("temp queue, op timeout 0", rk, NULL, 0);
         do_test_DeleteTopics("main queue, op timeout 15000", rk, mainq, 1500);
 
-        /* Create Partitions */
-        do_test_CreatePartitions("temp queue, op timeout 6500", rk, NULL, 6500);
-        do_test_CreatePartitions("main queue, op timeout 0", rk, mainq, 0);
+        if (test_broker_version >= TEST_BRKVER(1,0,0,0)) {
+                /* Create Partitions */
+                do_test_CreatePartitions("temp queue, op timeout 6500",
+                                         rk, NULL, 6500);
+                do_test_CreatePartitions("main queue, op timeout 0",
+                                         rk, mainq, 0);
+        }
 
         /* AlterConfigs */
         do_test_AlterConfigs(rk, mainq);

tests/0084-destroy_flags.c

@@ -180,10 +180,16 @@ static void destroy_flags (int local_mode) {
         const int flag_combos[] = { 0,
                                     RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE };
         const char *topic = test_mk_topic_name(__FUNCTION__, 1);
+        const rd_bool_t can_subscribe =
+                test_broker_version >= TEST_BRKVER(0,9,0,0);
         int i, j;
 
         for (i = 0 ; i < (int)RD_ARRAYSIZE(args) ; i++) {
                 for (j = 0 ; j < (int)RD_ARRAYSIZE(flag_combos) ; j++) {
+                        if (!can_subscribe &&
+                            (args[i].consumer_subscribe ||
+                             args[i].consumer_unsubscribe))
+                                continue;
                         do_test_destroy_flags(topic,
                                               flag_combos[j],
                                               local_mode,

tests/0086-purge.c

@@ -293,11 +293,17 @@ static void do_test_purge (const char *what, int remote,
 
 
 int main_0086_purge_remote (int argc, char **argv) {
+        const rd_bool_t has_idempotence =
+                test_broker_version >= TEST_BRKVER(0,11,0,0);
+
         do_test_purge("remote", 1/*remote*/, 0/*idempotence*/, 0/*!gapless*/);
-        do_test_purge("remote,idempotence", 1/*remote*/, 1/*idempotence*/,
-                      0/*!gapless*/);
-        do_test_purge("remote,idempotence,gapless", 1/*remote*/,
-                      1/*idempotence*/, 1/*!gapless*/);
+
+        if (has_idempotence) {
+                do_test_purge("remote,idempotence",
+                              1/*remote*/, 1/*idempotence*/, 0/*!gapless*/);
+                do_test_purge("remote,idempotence,gapless",
+                              1/*remote*/, 1/*idempotence*/, 1/*!gapless*/);
+        }
         return 0;
 }
 

tests/interactive_broker_version.py

@@ -69,10 +69,9 @@ def test_version (version, cmd=None, deploy=True, conf={}, debug=False, exec_cnt
     fd, test_conf_file = tempfile.mkstemp(prefix='test_conf', text=True)
     os.write(fd, ('test.sql.command=sqlite3 rdktests\n').encode('ascii'))
     os.write(fd, 'broker.address.family=v4\n'.encode('ascii'))
-    if version != 'trunk':
+    if version.startswith('0.9') or version.startswith('0.8'):
+        os.write(fd, 'api.version.request=false\n'.encode('ascii'))
         os.write(fd, ('broker.version.fallback=%s\n' % version).encode('ascii'))
-    else:
-        os.write(fd, 'api.version.request=true\n'.encode('ascii'))
     # SASL (only one mechanism supported)
     mech = defconf.get('sasl_mechanisms', '').split(',')[0]
     if mech != '':

tests/test.c

@@ -249,7 +249,9 @@ struct test tests[] = {
         _TEST(0006_symbols, TEST_F_LOCAL),
         _TEST(0007_autotopic, 0),
         _TEST(0008_reqacks, 0),
-        _TEST(0009_mock_cluster, TEST_F_LOCAL),
+        _TEST(0009_mock_cluster, TEST_F_LOCAL,
+              /* Mock cluster requires MsgVersion 2 */
+              TEST_BRKVER(0,11,0,0)),
         _TEST(0011_produce_batch, 0,
               /* Produces a lot of messages */
               _THRES(.ucpu = 40.0, .scpu = 8.0)),
@@ -290,7 +292,7 @@ struct test tests[] = {
               _THRES(.ucpu = 20.0, .scpu = 10.0)),
 	_TEST(0042_many_topics, 0),
 	_TEST(0043_no_connection, TEST_F_LOCAL),
-	_TEST(0044_partition_cnt, 0,
+	_TEST(0044_partition_cnt, 0, TEST_BRKVER(1,0,0,0),
               /* Produces a lot of messages */
               _THRES(.ucpu = 30.0)),
 	_TEST(0045_subscribe_update, 0, TEST_BRKVER(0,9,0,0)),
@@ -331,7 +333,7 @@ struct test tests[] = {
         _TEST(0068_produce_timeout, TEST_F_SOCKEM),
 #endif
         _TEST(0069_consumer_add_parts, TEST_F_KNOWN_ISSUE_WIN32,
-              TEST_BRKVER(0,9,0,0)),
+              TEST_BRKVER(1,0,0,0)),
         _TEST(0070_null_empty, 0),
         _TEST(0072_headers_ut, TEST_F_LOCAL),
         _TEST(0073_headers, 0, TEST_BRKVER(0,11,0,0)),
@@ -340,7 +342,9 @@ struct test tests[] = {
         _TEST(0075_retry, TEST_F_SOCKEM),
 #endif
         _TEST(0076_produce_retry, TEST_F_SOCKEM),
-        _TEST(0077_compaction, 0, TEST_BRKVER(0,9,0,0)),
+        _TEST(0077_compaction, 0,
+              /* The test itself requires message headers */
+              TEST_BRKVER(0,11,0,0)),
         _TEST(0078_c_from_cpp, TEST_F_LOCAL),
         _TEST(0079_fork, TEST_F_LOCAL|TEST_F_KNOWN_ISSUE,
               .extra = "using a fork():ed rd_kafka_t is not supported and will "
@@ -368,12 +372,13 @@ struct test tests[] = {
 #endif
         _TEST(0095_all_brokers_down, TEST_F_LOCAL),
         _TEST(0097_ssl_verify, 0),
-        _TEST(0098_consumer_txn, 0),
+        _TEST(0098_consumer_txn, 0, TEST_BRKVER(0,11,0,0)),
         _TEST(0099_commit_metadata, 0),
         _TEST(0100_thread_interceptors, TEST_F_LOCAL),
         _TEST(0101_fetch_from_follower, 0, TEST_BRKVER(2,4,0,0)),
         _TEST(0102_static_group_rebalance, 0, TEST_BRKVER(2,3,0,0)),
-        _TEST(0104_fetch_from_follower_mock, TEST_F_LOCAL),
+        _TEST(0104_fetch_from_follower_mock, TEST_F_LOCAL,
+              TEST_BRKVER(2,4,0,0)),
 
         /* Manual tests */
         _TEST(8000_idle, TEST_F_MANUAL),
@@ -2685,7 +2690,7 @@ int test_msgver_add_msg0 (const char *func, int line,
                                           "%s:%d: msgid expected in header %s "
                                           "but %s exists for "
                                           "message at offset %"PRId64
-                                          " has no headers",
+                                          " has no headers\n",
                                           func, line, mv->msgid_hdr,
                                           hdrs ? "no such header" : "no headers",
                                           rkmessage->offset);