Fix for KIP-1102 time based re-bootstrap condition (#5177)

Re-bootstrap is now triggered only after metadata.recovery.rebootstrap.trigger.ms
have passed since first metadata refresh request after last successful
metadata response. The calculation was since last successful metadata response
so it's possible it did overlap with the periodic topic.metadata.refresh.interval.ms
and cause a re-bootstrap even if not needed.

Author: emasab

CHANGELOG.md

@@ -4,13 +4,23 @@ librdkafka v2.12.0 is a feature release:
 
 * Fix compression types read issue in GetTelemetrySubscriptions response
   for big-endian architectures (#5183, @paravoid).
+* Fix for KIP-1102 time based re-bootstrap condition (#5177).
+* Fix for discarding the member epoch in a consumer group heartbeat response when leaving with an inflight HB (#4672).
+* Fix for an error being raised after a commit due to an existing error in the topic partition (#4672).
 
 
 ## Fixes
 
 ### General fixes
-* Fix for discarding the member epoch in a consumer group heartbeat response when leaving with an inflight HB (#4672).
-* Fix for an error being raised after a commit due to an existing error in the topic partition (#4672).
+
+* Issues: #5178.
+  Fix for KIP-1102 time based re-bootstrap condition.
+  Re-bootstrap is now triggered only after `metadata.recovery.rebootstrap.trigger.ms`
+  have passed since first metadata refresh request after last successful
+  metadata response. The calculation was since last successful metadata response
+  so it's possible it did overlap with the periodic `topic.metadata.refresh.interval.ms`
+  and cause a re-bootstrap even if not needed.
+  Happening since 2.11.0 (#5177).
 
 ### Telemetry fixes
 

src/rdkafka.c

@@ -2833,6 +2833,10 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
 
 /**
  * Schedules a rebootstrap of the cluster immediately.
+ *
+ * @locks none
+ * @locks_acquired rd_kafka_timers_lock()
+ * @locality any
  */
 void rd_kafka_rebootstrap(rd_kafka_t *rk) {
         if (rk->rk_conf.metadata_recovery_strategy ==
@@ -2851,22 +2855,42 @@ void rd_kafka_rebootstrap(rd_kafka_t *rk) {
 }
 
 /**
- * Restarts rebootstrap timer with the configured interval.
+ * Starts rebootstrap timer with the configured interval. Only if not
+ * started or stopped.
  *
  * @locks none
+ * @locks_acquired rd_kafka_timers_lock()
  * @locality any
  */
-void rd_kafka_rebootstrap_tmr_restart(rd_kafka_t *rk) {
+void rd_kafka_rebootstrap_tmr_start_maybe(rd_kafka_t *rk) {
         if (rk->rk_conf.metadata_recovery_strategy ==
             RD_KAFKA_METADATA_RECOVERY_STRATEGY_NONE)
                 return;
 
         rd_kafka_timer_start_oneshot(
-            &rk->rk_timers, &rk->rebootstrap_tmr, rd_true /*restart*/,
+            &rk->rk_timers, &rk->rebootstrap_tmr, rd_false /*don't restart*/,
             rk->rk_conf.metadata_recovery_rebootstrap_trigger_ms * 1000LL,
             rd_kafka_rebootstrap_tmr_cb, NULL);
 }
 
+/**
+ * Stops rebootstrap timer, for example after a successful metadata response.
+ *
+ * @return 1 if the timer was started (before being stopped), else 0.
+ *
+ * @locks none
+ * @locks_acquired rd_kafka_timers_lock()
+ * @locality any
+ */
+int rd_kafka_rebootstrap_tmr_stop(rd_kafka_t *rk) {
+        if (rk->rk_conf.metadata_recovery_strategy ==
+            RD_KAFKA_METADATA_RECOVERY_STRATEGY_NONE)
+                return 0;
+
+        return rd_kafka_timer_stop(&rk->rk_timers, &rk->rebootstrap_tmr,
+                                   rd_true /* lock */);
+}
+
 /**
  * Counts usage of the legacy/simple consumer (rd_kafka_consume_start() with
  * friends) since it does not have an API for stopping the cgrp we will need to

src/rdkafka_int.h

@@ -1251,7 +1251,9 @@ rd_kafka_resp_err_t rd_kafka_background_thread_create(rd_kafka_t *rk,
 
 void rd_kafka_rebootstrap(rd_kafka_t *rk);
 
-void rd_kafka_rebootstrap_tmr_restart(rd_kafka_t *rk);
+void rd_kafka_rebootstrap_tmr_start_maybe(rd_kafka_t *rk);
+
+int rd_kafka_rebootstrap_tmr_stop(rd_kafka_t *rk);
 
 void rd_kafka_reset_any_broker_down_reported(rd_kafka_t *rk);
 

src/rdkafka_metadata.c

@@ -1033,7 +1033,6 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
         rd_kafka_wrlock(rkb->rkb_rk);
 
         rkb->rkb_rk->rk_ts_metadata = rd_clock();
-        rd_kafka_rebootstrap_tmr_restart(rkb->rkb_rk);
 
         /* Update cached cluster id. */
         if (RD_KAFKAP_STR_LEN(&cluster_id) > 0 &&

src/rdkafka_request.c

@@ -2620,6 +2620,7 @@ static void rd_kafka_handle_Metadata(rd_kafka_t *rk,
         if (err)
                 goto err;
 
+        rd_kafka_rebootstrap_tmr_stop(rk);
         if (rko && rko->rko_replyq.q) {
                 /* Reply to metadata requester, passing on the metadata.
                  * Reuse requesting rko for the reply. */
@@ -2744,6 +2745,18 @@ static void rd_kafka_handle_Metadata(rd_kafka_t *rk,
  *
  * @sa rd_kafka_MetadataRequest().
  * @sa rd_kafka_MetadataRequest_resp_cb().
+ *
+ * @locality any
+ * @locks none
+ * @locks_acquired
+ *  - mtx_lock(&rkb->rkb_rk->rk_metadata_cache.rkmc_full_lock) in
+ *    this function
+ *  - rd_kafka_broker_lock() in
+ *    rd_kafka_broker_ApiVersion_supported()
+ *  - rd_kafka_timers_lock() in
+ *    rd_kafka_rebootstrap_tmr_start_maybe()
+ *  - mtx_lock(&rkq->rkq_lock) in
+ *    rd_kafka_broker_buf_enq_replyq()
  */
 static rd_kafka_resp_err_t
 rd_kafka_MetadataRequest0(rd_kafka_broker_t *rkb,
@@ -2877,9 +2890,28 @@ rd_kafka_MetadataRequest0(rd_kafka_broker_t *rkb,
                 rkbuf->rkbuf_u.Metadata.decr = full_incr;
                 rkbuf->rkbuf_u.Metadata.decr_lock =
                     &rkb->rkb_rk->rk_metadata_cache.rkmc_full_lock;
+        } else if (!resp_cb) {
+                /*
+                 * Conditions for starting the rebootstrap timer:
+                 *
+                 * !full_incr:
+                 * In case it's a full request, forced or not, it won't be
+                 * retried on the same broker to avoid blocking metadata
+                 * requests, because of the lock, when that broker isn't
+                 * available. We don't start the timer as we cannot ensure the
+                 * request is retried for the duration of
+                 * `metadata.recovery.rebootstrap.trigger.ms`.
+                 *
+                 * !resp_cb:
+                 * Same reasoning applies when we use a custom callback, for the
+                 * AdminClient requests for example.
+                 *
+                 * Start the rebootstrap timer only if it's the first
+                 * metadata refresh request after last successful response,
+                 * so the timer is not reset if already scheduled. */
+                rd_kafka_rebootstrap_tmr_start_maybe(rkb->rkb_rk);
         }
 
-
         if (topic_cnt > 0) {
                 char *topic;
                 int i;