Don't let active logical broker connections hinder new connects

The logical broker connections (such as to the group coordinator)
are reserved for specific use and shall not be reused for things like
Metadata requests. The code to lookup a usable broker to send
Metadata requests understood this, but the code to select a broker
to create a new connection to if no usable connections are
found (sparse connnections) did not understand this and already
thought there was an active connection (the logical broker connection)
and refused to set up a new one.

This could lead to consumers stalling consumption if it was fetching
from a single broker and that broker went down, no new connection
to the cluster would be made to refresh metadata.

(cherry picked from commit 3c7f14c)

Author: edenhill

src/rdkafka_broker.c

@@ -317,11 +317,19 @@ void rd_kafka_broker_set_state (rd_kafka_broker_t *rkb, int state) {
 
         if (rkb->rkb_source != RD_KAFKA_INTERNAL) {
                 if (rd_kafka_broker_state_is_up(state) &&
-                    !rd_kafka_broker_state_is_up(rkb->rkb_state))
+                    !rd_kafka_broker_state_is_up(rkb->rkb_state)) {
                         rd_atomic32_add(&rkb->rkb_rk->rk_broker_up_cnt, 1);
-                else if (rd_kafka_broker_state_is_up(rkb->rkb_state) &&
-                         !rd_kafka_broker_state_is_up(state))
+                        if (RD_KAFKA_BROKER_IS_LOGICAL(rkb))
+                                rd_atomic32_add(&rkb->rkb_rk->
+                                                rk_logical_broker_up_cnt, 1);
+
+                } else if (rd_kafka_broker_state_is_up(rkb->rkb_state) &&
+                           !rd_kafka_broker_state_is_up(state)) {
                         rd_atomic32_sub(&rkb->rkb_rk->rk_broker_up_cnt, 1);
+                        if (RD_KAFKA_BROKER_IS_LOGICAL(rkb))
+                                rd_atomic32_sub(&rkb->rkb_rk->
+                                                rk_logical_broker_up_cnt, 1);
+                }
         }
 
 	rkb->rkb_state = state;
@@ -5234,7 +5242,12 @@ void rd_kafka_connect_any (rd_kafka_t *rk, const char *reason) {
         rd_kafka_broker_t *rkb;
         rd_ts_t suppr;
 
-        if (rd_atomic32_get(&rk->rk_broker_up_cnt) > 0 ||
+        /* Don't count connections to logical brokers since they serve
+         * a specific purpose (group coordinator) and their connections
+         * should not be reused for other purposes.
+         * rd_kafka_broker_random() will not return LOGICAL brokers. */
+        if (rd_atomic32_get(&rk->rk_broker_up_cnt) -
+            rd_atomic32_get(&rk->rk_logical_broker_up_cnt) > 0 ||
             rd_atomic32_get(&rk->rk_broker_cnt) == 0)
                 return;
 

src/rdkafka_idempotence.c

@@ -143,7 +143,7 @@ int rd_kafka_idemp_request_pid (rd_kafka_t *rk, rd_kafka_broker_t *rkb,
         } else {
                 /* Increase passed broker's refcount so we don't
                  * have to check if rkb should be destroyed or not below
-                 * (broker_any_usable() returns a new reference). */
+                 * (broker_any() returns a new reference). */
                 rd_kafka_broker_keep(rkb);
         }
 

src/rdkafka_int.h

@@ -163,8 +163,13 @@ struct rd_kafka_s {
 	TAILQ_HEAD(, rd_kafka_broker_s) rk_brokers;
         rd_list_t                  rk_broker_by_id; /* Fast id lookups. */
 	rd_atomic32_t              rk_broker_cnt;
-        rd_atomic32_t              rk_broker_up_cnt; /**< Number of brokers
-                                                      *   in state >= UP */
+        /**< Number of brokers in state >= UP */
+        rd_atomic32_t              rk_broker_up_cnt;
+        /**< Number of logical brokers in state >= UP, this is a sub-set
+         *   of rk_broker_up_cnt. */
+        rd_atomic32_t              rk_logical_broker_up_cnt;
+        /**< Number of brokers that are down, only includes brokers
+         *   that have had at least one connection attempt. */
 	rd_atomic32_t              rk_broker_down_cnt;
         /**< Logical brokers currently without an address.
          *   Used for calculating ERR__ALL_BROKERS_DOWN. */