Refresh broker list even if no topics to refresh (confluentinc#2466)

Author: edenhill

CONFIGURATION.md

@@ -13,7 +13,7 @@ receive.message.max.bytes                |  *  | 1000 .. 2147483647 |     100000
 max.in.flight.requests.per.connection    |  *  | 1 .. 1000000    |       1000000 | low        | Maximum number of in-flight requests per broker connection. This is a generic property applied to all broker communication, however it is primarily relevant to produce requests. In particular, note that other mechanisms limit the number of outstanding consumer fetch request per broker to one. <br>*Type: integer*
 max.in.flight                            |  *  | 1 .. 1000000    |       1000000 | low        | Alias for `max.in.flight.requests.per.connection`: Maximum number of in-flight requests per broker connection. This is a generic property applied to all broker communication, however it is primarily relevant to produce requests. In particular, note that other mechanisms limit the number of outstanding consumer fetch request per broker to one. <br>*Type: integer*
 metadata.request.timeout.ms              |  *  | 10 .. 900000    |         60000 | low        | Non-topic request timeout in milliseconds. This is for metadata requests, etc. <br>*Type: integer*
-topic.metadata.refresh.interval.ms       |  *  | -1 .. 3600000   |        300000 | low        | Topic metadata refresh interval in milliseconds. The metadata is automatically refreshed on error and connect. Use -1 to disable the intervalled refresh. <br>*Type: integer*
+topic.metadata.refresh.interval.ms       |  *  | -1 .. 3600000   |        300000 | low        | Period of time in milliseconds at which topic and broker metadata is refreshed in order to proactively discover any new brokers, topics, partitions or partition leader changes. Use -1 to disable the intervalled refresh (not recommended). If there are no locally referenced topics (no topic objects created, no messages produced, no subscription or no assignment) then only the broker list will be refreshed every interval but no more often than every 10s. <br>*Type: integer*
 metadata.max.age.ms                      |  *  | 1 .. 86400000   |        900000 | low        | Metadata cache max age. Defaults to topic.metadata.refresh.interval.ms * 3 <br>*Type: integer*
 topic.metadata.refresh.fast.interval.ms  |  *  | 1 .. 60000      |           250 | low        | When a topic loses its leader a new metadata request will be enqueued with this initial interval, exponentially increasing until the topic metadata has been refreshed. This is used to recover quickly from transitioning leader brokers. <br>*Type: integer*
 topic.metadata.refresh.fast.cnt          |  *  | 0 .. 1000       |            10 | low        | **DEPRECATED** No longer used. <br>*Type: integer*

src/rdkafka.c

@@ -1758,10 +1758,23 @@ static void rd_kafka_metadata_refresh_cb (rd_kafka_timers_t *rkts, void *arg) {
             rk->rk_cgrp->rkcg_flags & RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION)
                 sparse = 0;
 
-        if (sparse)
-                rd_kafka_metadata_refresh_known_topics(
-                        rk, NULL, 1/*force*/, "periodic refresh");
-        else
+        if (sparse) {
+                if (rd_kafka_metadata_refresh_known_topics(
+                            rk, NULL, 1/*force*/,
+                            "periodic topic and broker list refresh") ==
+                    RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC &&
+                    rd_interval(&rk->rk_suppress.broker_metadata_refresh,
+                                10*1000*1000 /*10s*/, 0) > 0) {
+                        /* If there are no (locally referenced) topics
+                         * to query, refresh the broker list.
+                         * This avoids getting idle-disconnected for clients
+                         * that have not yet referenced a topic and makes
+                         * sure such a client has an up to date broker list. */
+                        rd_kafka_metadata_refresh_brokers(
+                                rk, NULL,
+                                "periodic broker list refresh");
+                }
+        } else
                 rd_kafka_metadata_refresh_all(rk, NULL, "periodic refresh");
 }
 
@@ -1961,6 +1974,7 @@ rd_kafka_t *rd_kafka_new (rd_kafka_type_t type, rd_kafka_conf_t *app_conf,
         mtx_init(&rk->rk_init_lock, mtx_plain);
 
         rd_interval_init(&rk->rk_suppress.no_idemp_brokers);
+        rd_interval_init(&rk->rk_suppress.broker_metadata_refresh);
         rd_interval_init(&rk->rk_suppress.sparse_connect_random);
         mtx_init(&rk->rk_suppress.sparse_connect_lock, mtx_plain);
 

src/rdkafka_conf.c

@@ -304,12 +304,17 @@ static const struct rd_kafka_property rd_kafka_properties[] = {
 	  "Non-topic request timeout in milliseconds. "
 	  "This is for metadata requests, etc.",
 	  10, 900*1000, 60*1000},
-	{ _RK_GLOBAL, "topic.metadata.refresh.interval.ms", _RK_C_INT,
-	  _RK(metadata_refresh_interval_ms),
-	  "Topic metadata refresh interval in milliseconds. "
-	  "The metadata is automatically refreshed on error and connect. "
-	  "Use -1 to disable the intervalled refresh.",
-	  -1, 3600*1000, 5*60*1000 },
+        { _RK_GLOBAL, "topic.metadata.refresh.interval.ms", _RK_C_INT,
+          _RK(metadata_refresh_interval_ms),
+          "Period of time in milliseconds at which topic and broker "
+          "metadata is refreshed in order to proactively discover any new "
+          "brokers, topics, partitions or partition leader changes. "
+          "Use -1 to disable the intervalled refresh (not recommended). "
+          "If there are no locally referenced topics "
+          "(no topic objects created, no messages produced, "
+          "no subscription or no assignment) then only the broker list will "
+          "be refreshed every interval but no more often than every 10s.",
+          -1, 3600*1000, 5*60*1000 },
 	{ _RK_GLOBAL, "metadata.max.age.ms", _RK_C_INT,
           _RK(metadata_max_age_ms),
           "Metadata cache max age. "

src/rdkafka_int.h

@@ -355,6 +355,15 @@ struct rd_kafka_s {
                 rd_interval_t sparse_connect_random;
                 /**< Lock for sparse_connect_random */
                 mtx_t         sparse_connect_lock;
+
+                /**< Broker metadata refresh interval:
+                 *   this is rate-limiting the number of topic-less
+                 *   broker/cluster metadata refreshes when there are no
+                 *   topics to refresh.
+                 *   Will be refreshed every topic.metadata.refresh.interval.ms
+                 *   but no more often than every 10s.
+                 *   No locks: only accessed by rdkafka main thread. */
+                rd_interval_t broker_metadata_refresh;
         } rk_suppress;
 
         struct {

src/rdkafka_metadata.c

@@ -583,8 +583,9 @@ rd_kafka_parse_Metadata (rd_kafka_broker_t *rkb,
         }
 
         /* Check if cgrp effective subscription is affected by
-         * new metadata. */
-        if (rkb->rkb_rk->rk_cgrp)
+         * new topic metadata.
+         * Ignore if this was a broker-only refresh (no topics) */
+        if ((requested_topics || all_topics) && rkb->rkb_rk->rk_cgrp)
                 rd_kafka_cgrp_metadata_update_check(
                         rkb->rkb_rk->rk_cgrp, 1/*do join*/);