Fix feature version ranges, currently matching a single version (confluentinc#5130)

Made the conditions for enabling the features future proof, allowing to
remove RPC versions in a subsequent major Apache Kafka version without disabling
features. The existing checks were matching a single version instead of
a range and were failing if the older version was removed.
Happening since 1.x

Author: emasab

CHANGELOG.md

@@ -1,3 +1,23 @@
+# librdkafka v2.11.1
+
+librdkafka v2.11.1 is a maintenance release:
+
+* Made the conditions for enabling the features future proof (#5130).
+
+
+## Fixes
+
+### General fixes
+
+* Issues: #4948, #4956.
+  Made the conditions for enabling the features future proof, allowing to
+  remove RPC versions in a subsequent Apache Kafka version without disabling
+  features. The existing checks were matching a single version instead of
+  a range and were failing if the older version was removed.
+  Happening since 1.x (#5130).
+
+
+
 # librdkafka v2.11.0
 
 librdkafka v2.11.0 is a feature release:
@@ -11,7 +31,7 @@ librdkafka v2.11.0 is a feature release:
   JoinGroup v0 anymore, missing in AK 4.0 and CP 8.0 (#5131).
 * Improve HTTPS CA certificates configuration by probing several paths
   when OpenSSL is statically linked and providing a way to customize their location
-  or value (#).
+  or value (#5133).
 
 
 ## Fixes

src/rdkafka_broker.c

@@ -310,6 +310,24 @@ int16_t rd_kafka_broker_ApiVersion_supported(rd_kafka_broker_t *rkb,
             rkb, ApiKey, minver, maxver, featuresp, rd_true /* do_lock */);
 }
 
+/**
+ * @brief Check that at least a ApiVersion greater or equal to
+ *        \p minver exists for \p ApiKey.
+ *
+ * @returns `rd_true` if the broker supports \p ApiKey with
+ *          a version greater than or equal to \p minver, else `rd_false`.
+ * @locks none
+ * @locks_acquired rd_kafka_broker_lock()
+ * @locality any
+ */
+rd_bool_t rd_kafka_broker_ApiVersion_at_least(rd_kafka_broker_t *rkb,
+                                              int16_t ApiKey,
+                                              int16_t minver) {
+        return rd_kafka_broker_ApiVersion_supported0(
+                   rkb, ApiKey, minver, RD_KAFKAP_RPC_VERSION_MAX, NULL,
+                   rd_true /* do_lock */) != -1;
+}
+
 /**
  * @brief Set broker state.
  *
@@ -2214,53 +2232,6 @@ rd_kafka_broker_reconnect_backoff(const rd_kafka_broker_t *rkb, rd_ts_t now) {
         return (int)(remains / 1000);
 }
 
-
-/**
- * @brief Unittest for reconnect.backoff.ms
- */
-static int rd_ut_reconnect_backoff(void) {
-        rd_kafka_broker_t rkb = RD_ZERO_INIT;
-        rd_kafka_conf_t conf  = {.reconnect_backoff_ms     = 10,
-                                 .reconnect_backoff_max_ms = 90};
-        rd_ts_t now           = 1000000;
-        int backoff;
-
-        rkb.rkb_reconnect_backoff_ms = conf.reconnect_backoff_ms;
-
-        /* broker's backoff is the initial reconnect.backoff.ms=10 */
-        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
-        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
-        RD_UT_ASSERT_RANGE(backoff, 7, 15, "%d");
-
-        /* .. 20 */
-        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
-        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
-        RD_UT_ASSERT_RANGE(backoff, 15, 30, "%d");
-
-        /* .. 40 */
-        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
-        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
-        RD_UT_ASSERT_RANGE(backoff, 30, 60, "%d");
-
-        /* .. 80, the jitter is capped at reconnect.backoff.max.ms=90  */
-        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
-        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
-        RD_UT_ASSERT_RANGE(backoff, 60, conf.reconnect_backoff_max_ms, "%d");
-
-        /* .. 90, capped by reconnect.backoff.max.ms */
-        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
-        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
-        RD_UT_ASSERT_RANGE(backoff, 67, conf.reconnect_backoff_max_ms, "%d");
-
-        /* .. 90, should remain at capped value. */
-        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
-        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
-        RD_UT_ASSERT_RANGE(backoff, 67, conf.reconnect_backoff_max_ms, "%d");
-
-        RD_UT_PASS();
-}
-
-
 /**
  * @brief Initiate asynchronous connection attempt to the next address
  *        in the broker's address list.
@@ -6266,6 +6237,98 @@ void rd_kafka_broker_decommission(rd_kafka_t *rk,
         rd_kafka_wrlock(rk);
 }
 
+/**
+ * @brief Unittest for reconnect.backoff.ms
+ */
+static int rd_ut_reconnect_backoff(void) {
+        rd_kafka_broker_t rkb = RD_ZERO_INIT;
+        rd_kafka_conf_t conf  = {.reconnect_backoff_ms     = 10,
+                                 .reconnect_backoff_max_ms = 90};
+        rd_ts_t now           = 1000000;
+        int backoff;
+
+        rkb.rkb_reconnect_backoff_ms = conf.reconnect_backoff_ms;
+
+        /* broker's backoff is the initial reconnect.backoff.ms=10 */
+        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
+        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
+        RD_UT_ASSERT_RANGE(backoff, 7, 15, "%d");
+
+        /* .. 20 */
+        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
+        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
+        RD_UT_ASSERT_RANGE(backoff, 15, 30, "%d");
+
+        /* .. 40 */
+        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
+        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
+        RD_UT_ASSERT_RANGE(backoff, 30, 60, "%d");
+
+        /* .. 80, the jitter is capped at reconnect.backoff.max.ms=90  */
+        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
+        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
+        RD_UT_ASSERT_RANGE(backoff, 60, conf.reconnect_backoff_max_ms, "%d");
+
+        /* .. 90, capped by reconnect.backoff.max.ms */
+        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
+        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
+        RD_UT_ASSERT_RANGE(backoff, 67, conf.reconnect_backoff_max_ms, "%d");
+
+        /* .. 90, should remain at capped value. */
+        rd_kafka_broker_update_reconnect_backoff(&rkb, &conf, now);
+        backoff = rd_kafka_broker_reconnect_backoff(&rkb, now);
+        RD_UT_ASSERT_RANGE(backoff, 67, conf.reconnect_backoff_max_ms, "%d");
+
+        RD_UT_PASS();
+}
+
+/**
+ * @brief Unittest for reconnect.backoff.ms
+ */
+static int rd_ut_ApiVersion_at_least(void) {
+        rd_kafka_broker_t rkb = RD_ZERO_INIT;
+        mtx_init(&rkb.rkb_lock, mtx_plain);
+
+        struct rd_kafka_ApiVersion av = {
+            .ApiKey = RD_KAFKAP_Metadata,
+            .MinVer = 5,
+            .MaxVer = 10,
+        };
+
+        rkb.rkb_ApiVersions_cnt = 1;
+        rkb.rkb_ApiVersions     = &av;
+
+        RD_UT_ASSERT(
+            rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 0),
+            "Metadata API version should be at least 0");
+        RD_UT_ASSERT(
+            rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 3),
+            "Metadata API version should be at least 3");
+        RD_UT_ASSERT(
+            rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 5),
+            "Metadata API version should be at least 5");
+        RD_UT_ASSERT(
+            rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 7),
+            "Metadata API version should be at least 7");
+        RD_UT_ASSERT(
+            rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 10),
+            "Metadata API version should be at least 10");
+        RD_UT_ASSERT(
+            !rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 11),
+            "Metadata API version shouldn't be at least 11");
+
+        rkb.rkb_ApiVersions_cnt = 0;
+        RD_UT_ASSERT(
+            !rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 0),
+            "Metadata API version shouldn't be at least 0");
+        RD_UT_ASSERT(
+            !rd_kafka_broker_ApiVersion_at_least(&rkb, RD_KAFKAP_Metadata, 3),
+            "Metadata API version shouldn't be at least 3");
+
+        mtx_destroy(&rkb.rkb_lock);
+        RD_UT_PASS();
+}
+
 /**
  * @name Unit tests
  * @{
@@ -6275,6 +6338,7 @@ int unittest_broker(void) {
         int fails = 0;
 
         fails += rd_ut_reconnect_backoff();
+        fails += rd_ut_ApiVersion_at_least();
 
         return fails;
 }

src/rdkafka_broker.h

@@ -477,6 +477,10 @@ int16_t rd_kafka_broker_ApiVersion_supported0(rd_kafka_broker_t *rkb,
                                               int *featuresp,
                                               rd_bool_t do_lock);
 
+rd_bool_t rd_kafka_broker_ApiVersion_at_least(rd_kafka_broker_t *rkb,
+                                              int16_t ApiKey,
+                                              int16_t minver);
+
 rd_kafka_broker_t *rd_kafka_broker_find_by_nodeid0_fl(const char *func,
                                                       int line,
                                                       rd_kafka_t *rk,