Add an interval to retry connection to any random broker after interval has passed. (#5039)

Remove the override to schedule a connection when there's no existing one as it causes too frequent connection retries in case cluster isn't reachable.

Remove scheduled connections count as the `sparse_connect_random` interval is again effective in any case.

Author: emasab

src/rdkafka.c

@@ -2375,7 +2375,6 @@ rd_kafka_t *rd_kafka_new(rd_kafka_type_t type,
         rd_atomic32_init(&rk->rk_logical_broker_cnt, 0);
         rd_atomic32_init(&rk->rk_broker_up_cnt, 0);
         rd_atomic32_init(&rk->rk_broker_down_cnt, 0);
-        rd_atomic32_init(&rk->rk_scheduled_connections_cnt, 0);
 
         rk->rk_rep             = rd_kafka_q_new(rk);
         rk->rk_ops             = rd_kafka_q_new(rk);

src/rdkafka_broker.c

@@ -3485,7 +3485,6 @@ rd_kafka_broker_op_serve(rd_kafka_broker_t *rkb, rd_kafka_op_t *rko) {
                 break;
 
         case RD_KAFKA_OP_CONNECT:
-                rd_atomic32_sub(&rkb->rkb_rk->rk_scheduled_connections_cnt, 1);
                 /* Sparse connections: connection requested, transition
                  * to TRY_CONNECT state to trigger new connection. */
                 if (rkb->rkb_state == RD_KAFKA_BROKER_STATE_INIT) {
@@ -5671,14 +5670,17 @@ static int rd_kafka_broker_filter_never_connected(rd_kafka_broker_t *rkb,
         return rd_atomic32_get(&rkb->rkb_c.connects);
 }
 
-/**
- * @brief Filter out brokers that aren't connecting.
- */
-static int rd_kafka_broker_filter_not_connecting(rd_kafka_broker_t *rkb,
-                                                 void *opaque) {
-        return rkb->rkb_state <= RD_KAFKA_BROKER_STATE_DOWN;
-}
+static void rd_kafka_connect_any_timer_cb(rd_kafka_timers_t *rkts, void *arg) {
+        const char *reason = (const char *)arg;
+        rd_kafka_t *rk     = rkts->rkts_rk;
+        if (rd_kafka_terminating(rk))
+                return;
 
+        /* Acquire the read lock for `rd_kafka_connect_any` */
+        rd_kafka_rdlock(rk);
+        rd_kafka_connect_any(rk, reason);
+        rd_kafka_rdunlock(rk);
+}
 
 /**
  * @brief Sparse connections:
@@ -5693,8 +5695,6 @@ static int rd_kafka_broker_filter_not_connecting(rd_kafka_broker_t *rkb,
 void rd_kafka_connect_any(rd_kafka_t *rk, const char *reason) {
         rd_kafka_broker_t *rkb;
         rd_ts_t suppr;
-        rd_bool_t any_connecting = rd_true;
-        int scheduled_connections;
 
         /* Don't count connections to logical brokers since they serve
          * a specific purpose (group coordinator) and their connections
@@ -5707,33 +5707,22 @@ void rd_kafka_connect_any(rd_kafka_t *rk, const char *reason) {
                 return;
 
         mtx_lock(&rk->rk_suppress.sparse_connect_lock);
-        rkb = rd_kafka_broker_random(
-            rk, -1 /*any state*/, rd_kafka_broker_filter_not_connecting, NULL);
-        if (rkb)
-                rd_kafka_broker_destroy(
-                    rkb); /* refcnt from ..broker_random() */
-        else
-                any_connecting = rd_false;
 
-        scheduled_connections =
-            rd_atomic32_get(&rk->rk_scheduled_connections_cnt);
-
-        if (!any_connecting && scheduled_connections == 0)
-                /* Skip interval */
-                rd_interval_reset(&rk->rk_suppress.sparse_connect_random);
         suppr = rd_interval(&rk->rk_suppress.sparse_connect_random,
                             rk->rk_conf.sparse_connect_intvl * 1000, 0);
+        mtx_unlock(&rk->rk_suppress.sparse_connect_lock);
 
         if (suppr <= 0) {
                 rd_kafka_dbg(rk, BROKER | RD_KAFKA_DBG_GENERIC, "CONNECT",
                              "Not selecting any broker for cluster connection: "
-                             "still suppressed for %" PRId64
-                             "ms, "
-                             "any broker connecting: %s, "
-                             "scheduled connections %d: %s",
-                             -suppr / 1000, RD_STR_ToF(any_connecting),
-                             scheduled_connections, reason);
-                goto done;
+                             "still suppressed for %" PRId64 "ms: %s",
+                             -suppr / 1000, reason);
+                /* Retry after interval + 1ms has passed */
+                rd_kafka_timer_start_oneshot(
+                    &rk->rk_timers, &rk->rk_suppress.sparse_connect_random_tmr,
+                    rd_false /* don't restart */, 1000LL - suppr,
+                    rd_kafka_connect_any_timer_cb, (void *)reason);
+                return;
         }
 
         /* First pass:  only match brokers never connected to,
@@ -5755,7 +5744,7 @@ void rd_kafka_connect_any(rd_kafka_t *rk, const char *reason) {
                 rd_kafka_dbg(rk, BROKER | RD_KAFKA_DBG_GENERIC, "CONNECT",
                              "Cluster connection already in progress: %s",
                              reason);
-                goto done;
+                return;
         }
 
         rd_rkb_dbg(rkb, BROKER | RD_KAFKA_DBG_GENERIC, "CONNECT",
@@ -5766,8 +5755,6 @@ void rd_kafka_connect_any(rd_kafka_t *rk, const char *reason) {
         rd_kafka_broker_schedule_connection(rkb);
 
         rd_kafka_broker_destroy(rkb); /* refcnt from ..broker_random() */
-done:
-        mtx_unlock(&rk->rk_suppress.sparse_connect_lock);
 }
 
 
@@ -5957,11 +5944,9 @@ void rd_kafka_broker_active_toppar_del(rd_kafka_broker_t *rkb,
  */
 void rd_kafka_broker_schedule_connection(rd_kafka_broker_t *rkb) {
         rd_kafka_op_t *rko;
-        rd_atomic32_add(&rkb->rkb_rk->rk_scheduled_connections_cnt, 1);
         rko = rd_kafka_op_new(RD_KAFKA_OP_CONNECT);
         rd_kafka_op_set_prio(rko, RD_KAFKA_PRIO_FLASH);
-        if (!rd_kafka_q_enq(rkb->rkb_ops, rko))
-                rd_atomic32_sub(&rkb->rkb_rk->rk_scheduled_connections_cnt, 1);
+        rd_kafka_q_enq(rkb->rkb_ops, rko);
 }
 
 

src/rdkafka_int.h

@@ -299,9 +299,6 @@ struct rd_kafka_s {
          *   that have had at least one connection attempt
          *   and are configured or learned. */
         rd_atomic32_t rk_broker_down_cnt;
-        /* Number of sparse connections requested
-         * but still not executed. */
-        rd_atomic32_t rk_scheduled_connections_cnt;
 
         /**< Additional bootstrap servers list.
          *   contains all brokers added through rd_kafka_brokers_add().
@@ -673,6 +670,12 @@ struct rd_kafka_s {
                  *   Use 10 < reconnect.backoff.jitter.ms / 2 < 1000.
                  */
                 rd_interval_t sparse_connect_random;
+
+                /**  Sparse connection timer: fires after remaining time of
+                 *   `sparse_connect_random` interval + 1ms.
+                 */
+                rd_kafka_timer_t sparse_connect_random_tmr;
+
                 /**< Lock for sparse_connect_random */
                 mtx_t sparse_connect_lock;