Purge brokers no longer reported in metadata

src/rdkafka_broker.c

@@ -1407,9 +1407,8 @@ rd_kafka_broker_t *rd_kafka_broker_random0(const char *func,
         int fcnt = 0;
 
         TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
-                if (rd_kafka_broker_or_instance_terminating(rkb))
-                        continue;
-                if (RD_KAFKA_BROKER_IS_LOGICAL(rkb))
+                if (rd_kafka_broker_or_instance_terminating(rkb) ||
+                    RD_KAFKA_BROKER_IS_LOGICAL(rkb))
                         continue;
 
                 rd_kafka_broker_lock(rkb);
@@ -3483,7 +3482,7 @@ rd_kafka_broker_op_serve(rd_kafka_broker_t *rkb, rd_kafka_op_t *rko) {
                 break;
 
         case RD_KAFKA_OP_CONNECT:
-                rd_atomic32_add(&rkb->rkb_rk->rk_scheduled_connections_cnt, -1);
+                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) {
@@ -5540,7 +5539,6 @@ void rd_kafka_broker_update(rd_kafka_t *rk,
                         rko = rd_kafka_op_new(RD_KAFKA_OP_NODE_UPDATE);
                         rd_strlcpy(rko->rko_u.node.nodename, nodename,
                                    sizeof(rko->rko_u.node.nodename));
-                        rko->rko_u.node.nodeid = mdb->id;
                         /* Perform a blocking op request so that all
                          * broker-related state, such as the rk broker list,
                          * is up to date by the time this call returns.
@@ -5941,7 +5939,7 @@ void rd_kafka_broker_schedule_connection(rd_kafka_broker_t *rkb) {
         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_add(&rkb->rkb_rk->rk_scheduled_connections_cnt, -1);
+                rd_atomic32_sub(&rkb->rkb_rk->rk_scheduled_connections_cnt, 1);
 }
 
 
@@ -6118,19 +6116,14 @@ void rd_kafka_broker_start_reauth_cb(rd_kafka_timers_t *rkts, void *_rkb) {
 
 int32_t *rd_kafka_brokers_learned_ids(rd_kafka_t *rk, size_t *cntp) {
         rd_kafka_broker_t *rkb;
-
-        size_t all_broker_cnt = rd_atomic32_get(&rk->rk_broker_cnt);
-        /* This over-allocates but simplifies the code. */
-        int32_t *ids = malloc(sizeof(*ids) * all_broker_cnt);
-        int32_t *p   = ids;
+        int32_t *ids, *p;
+        int32_t i;
 
         *cntp = 0;
         rd_kafka_rdlock(rk);
-        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
-                if (rkb->rkb_source != RD_KAFKA_LEARNED ||
-                    rd_kafka_broker_termination_in_progress(rkb))
-                        continue;
-
+        ids = malloc(sizeof(*ids) * rd_list_cnt(&rk->rk_broker_by_id));
+        p   = ids;
+        RD_LIST_FOREACH(rkb, &rk->rk_broker_by_id, i) {
                 *p++ = rkb->rkb_nodeid;
                 (*cntp)++;
         }

src/rdkafka_metadata.c

@@ -504,8 +504,7 @@ static void rd_kafka_metadata_decommission_unavailable_brokers(
         TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
                 rd_bool_t purge_broker;
 
-                if (rkb->rkb_source != RD_KAFKA_LEARNED &&
-                    rkb->rkb_source != RD_KAFKA_CONFIGURED)
+                if (rkb->rkb_source == RD_KAFKA_LOGICAL)
                         continue;
 
                 purge_broker = rd_true;

src/rdkafka_op.h

@@ -413,7 +413,6 @@ struct rd_kafka_op_s {
                 } dr;
 
                 struct {
-                        int32_t nodeid;
                         char nodename[RD_KAFKA_NODENAME_SIZE];
                 } node;
 

tests/0149-broker-same-host-port.c

@@ -28,18 +28,76 @@
 
 #include "test.h"
 
+static rd_bool_t broker1_changed;
+static rd_bool_t broker2_changed;
+static rd_bool_t broker3_changed;
 
+/**
+ * @brief Keep track of which brokers have changed their nodename.
+ */
+static void broker_same_host_port_mock_log_cb(const rd_kafka_t *rk,
+                                              int level,
+                                              const char *fac,
+                                              const char *buf) {
+        const char *nodename = "to localhost:11192";
+        if (strstr(buf, "/1: Nodename changed") && strstr(buf, nodename))
+                broker1_changed = rd_true;
+
+        if (strstr(buf, "/2: Nodename changed") && strstr(buf, nodename))
+                broker2_changed = rd_true;
+
+        if (strstr(buf, "/3: Nodename changed") && strstr(buf, nodename))
+                broker3_changed = rd_true;
+}
+
+static void broker_same_host_port_mock_verify_broker_ids(rd_kafka_t *rk) {
+        const rd_kafka_metadata_t *md;
+        rd_kafka_resp_err_t err;
+        int32_t *ids;
+        size_t cnt               = 0;
+        const size_t num_brokers = 3;
+        size_t i;
+
+        /* Trigger Metadata request which will get initial broker hostnames. */
+        err = rd_kafka_metadata(rk, 0, NULL, &md, tmout_multip(5000));
+        /* Metadata timeout can happen if nodename change did already
+         * take place and there was a disconnection followed by a retry */
+        if (err && err != RD_KAFKA_RESP_ERR__TIMED_OUT)
+                TEST_ASSERT(!err, "%s", rd_kafka_err2str(err));
+        else if (!err)
+                rd_kafka_metadata_destroy(md);
+
+        ids = rd_kafka_brokers_learned_ids(rk, &cnt);
+
+        TEST_ASSERT(cnt == num_brokers,
+                    "expected %" PRIusz " brokers in cache, not %" PRIusz,
+                    num_brokers, cnt);
+
+        for (i = 0; i < cnt; i++) {
+                int32_t expected_id = i + 1;
+
+                TEST_ASSERT(ids[i] == expected_id,
+                            "expected broker %d in cache, not %d", expected_id,
+                            ids[i]);
+        }
+        if (ids)
+                free(ids);
+}
+
+/**
+ * @brief It should be possible to set the same hostname to brokers with
+ *        broker ids, when doing that, verify that the brokers are kept
+ *        separate instances and that the hostname is propagated to all of them.
+ */
 int main_0149_broker_same_host_port_mock(int argc, char **argv) {
         rd_kafka_mock_cluster_t *cluster;
         const char *bootstraps;
         rd_kafka_t *rk;
         rd_kafka_conf_t *conf;
-        const rd_kafka_metadata_t *md;
-        rd_kafka_resp_err_t err;
         const size_t num_brokers = 3;
-        size_t cnt               = 0;
-        int32_t *ids;
         size_t i;
+        test_conf_log_interceptor_t *log_interceptor;
+        const char *debug_contexts[2] = {"broker", NULL};
 
         if (test_needs_auth()) {
                 TEST_SKIP("Mock cluster does not support SSL/SASL\n");
@@ -48,41 +106,33 @@ int main_0149_broker_same_host_port_mock(int argc, char **argv) {
 
         cluster = test_mock_cluster_new(num_brokers, &bootstraps);
 
-        for (i = 1; i <= num_brokers; i++) {
-                rd_kafka_mock_broker_set_host_port(cluster, i, "localhost",
-                                                   9092);
-        }
-
-        test_conf_init(&conf, NULL, 10);
-
+        test_conf_init(&conf, NULL, tmout_multip(10));
         test_conf_set(conf, "bootstrap.servers", bootstraps);
+        log_interceptor = test_conf_set_log_interceptor(
+            conf, broker_same_host_port_mock_log_cb, debug_contexts);
 
         rk = test_create_handle(RD_KAFKA_PRODUCER, conf);
 
-        /* Trigger Metadata request which will update learned brokers. */
-        err = rd_kafka_metadata(rk, 0, NULL, &md, tmout_multip(5000));
-        rd_kafka_metadata_destroy(md);
-        TEST_ASSERT(!err, "%s", rd_kafka_err2str(err));
-
-        ids = rd_kafka_brokers_learned_ids(rk, &cnt);
+        TEST_SAY("Initial metadata request\n");
+        /* Trigger Metadata request which will get initial broker hostnames. */
+        broker_same_host_port_mock_verify_broker_ids(rk);
 
-        TEST_ASSERT(cnt == num_brokers,
-                    "expected %" PRIusz " brokers in cache, not %" PRIusz,
-                    num_brokers, cnt);
+        TEST_SAY("Changing nodenames\n");
+        for (i = 1; i <= num_brokers; i++) {
+                rd_kafka_mock_broker_set_host_port(cluster, i, "localhost",
+                                                   11192);
+        }
 
-        for (i = 0; i < cnt; i++) {
-                /* Brokers are added at the head of the list. */
-                int32_t expected_id = cnt - i;
+        TEST_SAY("Modified nodenames metadata request\n");
+        /* Trigger Metadata request which will get initial changed hostnames. */
+        broker_same_host_port_mock_verify_broker_ids(rk);
 
-                TEST_ASSERT(ids[i] == expected_id,
-                            "expected broker %d in cache, not %d", expected_id,
-                            ids[i]);
-        }
+        while (!(broker1_changed && broker2_changed && broker3_changed))
+                rd_usleep(100000, 0);
 
-        if (ids)
-                free(ids);
         rd_kafka_destroy(rk);
         test_mock_cluster_destroy(cluster);
+        rd_free(log_interceptor);
 
         return 0;
 }

tests/0151-purge-brokers.c

@@ -28,16 +28,9 @@
 
 #include "test.h"
 
-static const char *test_debug;
 static rd_atomic32_t do_test_remove_then_add_received_terminate;
 static rd_atomic32_t verification_complete;
 
-static int int32_cmp(const void *a, const void *b) {
-        int *int_a = (int *)a;
-        int *int_b = (int *)b;
-        return *int_a - *int_b;
-}
-
 /** @brief Verify that \p expected_broker_ids
  *         and \p actual_broker_ids correspond in
  *         count and value. Sorts \p actual_broker_ids .
@@ -50,8 +43,6 @@ static rd_bool_t fetch_metadata_verify_brokers(int32_t *expected_broker_ids,
         if (actual_broker_id_cnt != expected_broker_id_cnt)
                 return rd_false;
 
-        qsort(actual_broker_ids, actual_broker_id_cnt,
-              sizeof(*actual_broker_ids), int32_cmp);
         for (i = 0; i < actual_broker_id_cnt; i++) {
                 if (actual_broker_ids[i] != expected_broker_ids[i])
                         return rd_false;
@@ -325,22 +316,12 @@ static void do_test_remove_then_add_log_cb(const rd_kafka_t *rk,
                                            int level,
                                            const char *fac,
                                            const char *buf) {
-        int secs, msecs;
-        struct timeval tv;
         if (!rd_atomic32_get(&do_test_remove_then_add_received_terminate) &&
             strstr(buf, "/1: Handle terminates in state") > 0) {
                 rd_atomic32_set(&do_test_remove_then_add_received_terminate, 1);
                 while (!rd_atomic32_get(&verification_complete))
                         rd_usleep(100 * 1000, 0);
         }
-
-        if (test_debug) {
-                rd_gettimeofday(&tv, NULL);
-                secs  = (int)tv.tv_sec;
-                msecs = (int)(tv.tv_usec / 1000);
-                fprintf(stderr, "%%%i|%u.%03u|%s|%s| %s\n", level, secs, msecs,
-                        fac, rk ? rd_kafka_name(rk) : "", buf);
-        }
 }
 
 /**
@@ -363,24 +344,20 @@ static rd_bool_t do_test_remove_then_add_await_after_action_cb(int action) {
  *        Add a pause after receiving the TERMINATE op to allow to
  *        proceed with adding it again before it's decommissioned.
  */
+static test_conf_log_interceptor_t *log_interceptor;
 static void
 do_test_remove_then_add_edit_configuration_cb(rd_kafka_conf_t *conf) {
+        const char *debug_contexts[2] = {"broker", NULL};
+
         /* This timeout verifies that the correct brokers are returned
          * without duplicates as soon as possible. */
         test_timeout_set(6);
         /* Hidden property that forces connections to all brokers,
          * increasing likelyhood of wrong behaviour if the decommissioned broker
          * starts re-connecting. */
         test_conf_set(conf, "enable.sparse.connections", "false");
-        if (!test_debug ||
-            (!strstr(test_debug, "broker") && !strstr(test_debug, "all"))) {
-                char debug_with_broker[512];
-                rd_snprintf(debug_with_broker, sizeof(debug_with_broker),
-                            "%s%s%s", test_debug ? test_debug : "",
-                            test_debug ? "," : "", "broker");
-                test_conf_set(conf, "debug", debug_with_broker);
-        }
-        rd_kafka_conf_set_log_cb(conf, do_test_remove_then_add_log_cb);
+        log_interceptor = test_conf_set_log_interceptor(
+            conf, do_test_remove_then_add_log_cb, debug_contexts);
 }
 
 /**
@@ -411,6 +388,8 @@ static void do_test_remove_then_add(void) {
             expected_brokers_cnt, do_test_remove_then_add_edit_configuration_cb,
             do_test_remove_then_add_await_after_action_cb);
 
+        rd_free(log_interceptor);
+        log_interceptor = NULL;
         SUB_TEST_PASS();
 }
 
@@ -420,8 +399,6 @@ int main_0151_purge_brokers_mock(int argc, char **argv) {
                 TEST_SKIP("Mock cluster does not support SSL/SASL\n");
                 return 0;
         }
-        test_debug = test_getenv("TEST_DEBUG", NULL);
-
 
         do_test_replace_with_new_cluster();