Avoid returning an all brokers down error on planned disconnections (confluentinc#5126)

* Skip increasing brokers down count
for planned disconnections, those that were logged as DEBUG, including idle disconnections.
Only the first time it reaches the DOWN state it's skipped but it's counted if there are further disconnections.

* Avoid returning an all brokers down error on planned disconnections
This is done by avoiding to count planned disconnections, such as idle  disconnections, broker host change and similar, as events that can cause the client to reach the "all brokers down" state, returning an error and possibly restarting a re-bootstrap sequence since 2.10.0

* Add more information to disconnection error messages

* Skip mock test when using SSL

* `rd_atomic32_set` and `rd_atomic64_set`
return the previous value to be able to check if the atomic set changed the value

* Fix for `rd_kafka_broker_set_error`

* Regarding broker errors, only log warnings and only report errors to the user for levels less or equal to `ERR`

* Reduce flakyness: test 0083

* Reset the `rkb_down_reported` field when
a broker reaches the `UP` state, to check again other brokers before reaching the `ALL_BROKERS_DOWN` state. Avoids repeated `ALL_BROKERS_DOWN`, maybe triggered by idle disconnections when only connecting to a single broker

* Consider the disconnection after timeout
as a planned disconnection that doesn't count for reaching the all brokers down state

* Reset any broker down reported on re-bootstrap too

* Revert `rd_kafka_connect_any` to the previous version. Given we reset the broker down reported state it's not needed anymore to prioritize the learned brokers as all brokers will be tried again before starting a new re-boostrap sequence

* Add the `rk_rebootstrap_in_progress` field to prevent duplicated re-bootstrap starts if the initial re-bootstrap sequence isn't completed yet.

* Remove flakyness from test 0151 - `do_test_down_then_up_no_rebootstrap_loop` and make it consistent with current changes

* Fix test 0034
given the down reported state is reset, after retrying on re-bootstrap a new ALL_BROKERS_DOWN error is issued.
This happens after 1 second max (`rd_kafka_broker_reconnect_backoff`) and the test 0034 timeout is 5 seconds so it doesn't timeout and set the broker up again.
With an explicit timeout check the test is reliable.

* Fix memory leaks in tests

* Fix flakyness in test 0075 when using valgrind due to timing and multiple connect requests

* Set up a fixed max jitter of 2s

* Remove flakyness from test 0086_purge_remote

* Fix for a deadlock when calling `rd_kafka_filter_broker_by_GetTelemetrySubscription` introduced with confluentinc#5130 (not released)

Author: emasab

CHANGELOG.md

@@ -3,6 +3,9 @@
 librdkafka v2.11.1 is a maintenance release:
 
 * Made the conditions for enabling the features future proof (#5130).
+* Avoid returning an all brokers down error on planned disconnections (#5126).
+* An "all brokers down" error isn't returned when we haven't tried to connect
+  to all brokers since last successful connection (#5126).
 
 
 ## Fixes
@@ -16,6 +19,20 @@ librdkafka v2.11.1 is a maintenance release:
   a range and were failing if the older version was removed.
   Happening since 1.x (#5130).
 
+* Issues: #5142.
+  Avoid returning an all brokers down error on planned disconnections.
+  This is done by avoiding to count planned disconnections, such as idle
+  disconnections, broker host change and similar as events that can cause
+  the client to reach the "all brokers down" state, returning an error and
+  since 2.10.0 possibly starting a re-bootstrap sequence.
+* Issues: #5142.
+  An "all brokers down" error isn't returned when we haven't tried to connect
+  to all brokers since last successful connection. It happened because the down
+  state is cached and can be stale when a connection isn't needed to that
+  particular broker. Solved by resetting the cached broker down state when any
+  broker successfully connects, so that broker needs to be tried again.
+  Happening since 1.x (#5126).
+
 
 
 # librdkafka v2.11.0

CONFIGURATION.md

@@ -32,7 +32,7 @@ socket.max.fails                         |  *  | 0 .. 1000000    |             1
 broker.address.ttl                       |  *  | 0 .. 86400000   |          1000 | low        | How long to cache the broker address resolving results (milliseconds). <br>*Type: integer*
 broker.address.family                    |  *  | any, v4, v6     |           any | low        | Allowed broker IP address families: any, v4, v6 <br>*Type: enum value*
 socket.connection.setup.timeout.ms       |  *  | 1000 .. 2147483647 |         30000 | medium     | Maximum time allowed for broker connection setup (TCP connection setup as well SSL and SASL handshake). If the connection to the broker is not fully functional after this the connection will be closed and retried. <br>*Type: integer*
-connections.max.idle.ms                  |  *  | 0 .. 2147483647 |             0 | medium     | Close broker connections after the specified time of inactivity. Disable with 0. If this property is left at its default value some heuristics are performed to determine a suitable default value, this is currently limited to identifying brokers on Azure (see librdkafka issue #3109 for more info). <br>*Type: integer*
+connections.max.idle.ms                  |  *  | 0 .. 2147483647 |             0 | medium     | Close broker connections after the specified time of inactivity. Disable with 0. If this property is left at its default value some heuristics are performed to determine a suitable default value, this is currently limited to identifying brokers on Azure (see librdkafka issue #3109 for more info). Actual value can be lower, up to 2s lower, only if `connections.max.idle.ms` >= 4s, as jitter is added to avoid disconnecting all brokers at the same time. <br>*Type: integer*
 reconnect.backoff.jitter.ms              |  *  | 0 .. 3600000    |             0 | low        | **DEPRECATED** No longer used. See `reconnect.backoff.ms` and `reconnect.backoff.max.ms`. <br>*Type: integer*
 reconnect.backoff.ms                     |  *  | 0 .. 3600000    |           100 | medium     | The initial time to wait before reconnecting to a broker after the connection has been closed. The time is increased exponentially until `reconnect.backoff.max.ms` is reached. -25% to +50% jitter is applied to each reconnect backoff. A value of 0 disables the backoff and reconnects immediately. <br>*Type: integer*
 reconnect.backoff.max.ms                 |  *  | 0 .. 3600000    |         10000 | medium     | The maximum time to wait before reconnecting to a broker after the connection has been closed. <br>*Type: integer*

src/rdatomic.h

@@ -111,22 +111,25 @@ static RD_INLINE int32_t RD_UNUSED rd_atomic32_get(rd_atomic32_t *ra) {
 #endif
 }
 
+/**
+ * @brief Set the atomic value to \p v and return the previous value to
+ *        that can be used to determine if the value was changed.
+ */
 static RD_INLINE int32_t RD_UNUSED rd_atomic32_set(rd_atomic32_t *ra,
                                                    int32_t v) {
 #ifdef _WIN32
         return InterlockedExchange((LONG *)&ra->val, v);
 #elif !HAVE_ATOMICS_32
         int32_t r;
         mtx_lock(&ra->lock);
-        r = ra->val = v;
+        r       = rd->val;
+        ra->val = v;
         mtx_unlock(&ra->lock);
         return r;
 #elif HAVE_ATOMICS_32_ATOMIC
-        __atomic_store_n(&ra->val, v, __ATOMIC_SEQ_CST);
-        return v;
+        return __atomic_exchange_n(&ra->val, v, __ATOMIC_SEQ_CST);
 #elif HAVE_ATOMICS_32_SYNC
-        (void)__sync_lock_test_and_set(&ra->val, v);
-        return v;
+        return __sync_lock_test_and_set(&ra->val, v);
 #else
         return ra->val = v;  // FIXME
 #endif
@@ -200,24 +203,25 @@ static RD_INLINE int64_t RD_UNUSED rd_atomic64_get(rd_atomic64_t *ra) {
 #endif
 }
 
-
+/**
+ * @brief Set the atomic value to \p v and return the previous value to
+ *        that can be used to determine if the value was changed.
+ */
 static RD_INLINE int64_t RD_UNUSED rd_atomic64_set(rd_atomic64_t *ra,
                                                    int64_t v) {
 #ifdef _WIN32
         return InterlockedExchange64(&ra->val, v);
 #elif !HAVE_ATOMICS_64
         int64_t r;
         mtx_lock(&ra->lock);
-        ra->val = v;
         r       = ra->val;
+        ra->val = v;
         mtx_unlock(&ra->lock);
         return r;
 #elif HAVE_ATOMICS_64_ATOMIC
-        __atomic_store_n(&ra->val, v, __ATOMIC_SEQ_CST);
-        return v;
+        return __atomic_exchange_n(&ra->val, v, __ATOMIC_SEQ_CST);
 #elif HAVE_ATOMICS_64_SYNC
-        (void)__sync_lock_test_and_set(&ra->val, v);
-        return v;
+        return __sync_lock_test_and_set(&ra->val, v);
 #else
         return ra->val = v;  // FIXME
 #endif

src/rdkafka.c

@@ -2060,6 +2060,25 @@ static void rd_kafka_1s_tmr_cb(rd_kafka_timers_t *rkts, void *arg) {
         rd_kafka_coord_cache_expire(&rk->rk_coord_cache);
 }
 
+/**
+ * @brief Reset broker down reported flag for all brokers.
+ *        In case it was set to 1 it will be reset to 0 and
+ *        the broker down count will be decremented.
+ *
+ * @locks none
+ * @locks_acquired rd_kafka_rdlock()
+ * @locality any
+ */
+void rd_kafka_reset_any_broker_down_reported(rd_kafka_t *rk) {
+        rd_kafka_broker_t *rkb;
+        rd_kafka_rdlock(rk);
+        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
+                if (rd_atomic32_set(&rkb->rkb_down_reported, 0) == 1)
+                        rd_atomic32_sub(&rk->rk_broker_down_cnt, 1);
+        }
+        rd_kafka_rdunlock(rk);
+}
+
 /**
  * @brief Re-bootstrap timer callback.
  *
@@ -2087,6 +2106,9 @@ static void rd_kafka_rebootstrap_tmr_cb(rd_kafka_timers_t *rkts, void *arg) {
 
         rd_kafka_dbg(rk, ALL, "REBOOTSTRAP", "Starting re-bootstrap sequence");
 
+        rd_atomic32_set(&rk->rk_rebootstrap_in_progress, 1);
+        rd_kafka_reset_any_broker_down_reported(rk);
+
         if (rk->rk_conf.brokerlist) {
                 rd_kafka_brokers_add0(
                         rk,
@@ -2098,7 +2120,7 @@ static void rd_kafka_rebootstrap_tmr_cb(rd_kafka_timers_t *rkts, void *arg) {
         rd_kafka_rdlock(rk);
         if (rd_list_cnt(&rk->additional_brokerlists) == 0) {
                 rd_kafka_rdunlock(rk);
-                return;
+                goto done;
         }
 
         rd_list_init_copy(&additional_brokerlists, &rk->additional_brokerlists);
@@ -2113,6 +2135,8 @@ static void rd_kafka_rebootstrap_tmr_cb(rd_kafka_timers_t *rkts, void *arg) {
                          * names even if requested */);
         }
         rd_list_destroy(&additional_brokerlists);
+done:
+        rd_atomic32_set(&rk->rk_rebootstrap_in_progress, 0);
 }
 
 static void rd_kafka_stats_emit_tmr_cb(rd_kafka_timers_t *rkts, void *arg) {
@@ -2377,6 +2401,7 @@ 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_rebootstrap_in_progress, 0);
 
         rk->rk_rep             = rd_kafka_q_new(rk);
         rk->rk_ops             = rd_kafka_q_new(rk);
@@ -2814,9 +2839,15 @@ void rd_kafka_rebootstrap(rd_kafka_t *rk) {
             RD_KAFKA_METADATA_RECOVERY_STRATEGY_NONE)
                 return;
 
-        rd_kafka_timer_start_oneshot(&rk->rk_timers, &rk->rebootstrap_tmr,
-                                     rd_true /*restart*/, 0,
-                                     rd_kafka_rebootstrap_tmr_cb, NULL);
+        if (rd_atomic32_set(&rk->rk_rebootstrap_in_progress, 1) == 0) {
+                /* Only when not already in progress 0 -> 1.
+                 * After setting down a learned broker it could reconnect and
+                 * disconnect again before previous reboostrap completes,
+                 * causing a new re-bootstrap. */
+                rd_kafka_timer_start_oneshot(
+                    &rk->rk_timers, &rk->rebootstrap_tmr, rd_true /*restart*/,
+                    0, rd_kafka_rebootstrap_tmr_cb, NULL);
+        }
 }
 
 /**