[KIP-1102] Enable clients to rebootstrap based on timeout or error code

emasab · emasab · commit 2c542b05f6a9 · 2025-02-28T10:02:27.000+01:00
diff --git a/CONFIGURATION.md b/CONFIGURATION.md
@@ -13,6 +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.recovery.strategy               |  *  | none, rebootstrap |   rebootstrap | low        | Controls how the client recovers when none of the brokers known to it is available. If set to `none`, the client fails with a fatal error. If set to `rebootstrap`, the client repeats the bootstrap process using `bootstrap.servers` and brokers added through `rd_kafka_brokers_add()`. Rebootstrapping is useful when a client communicates with brokers so infrequently that the set of brokers may change entirely before the client refreshes metadata. Metadata recovery is triggered when all last-known brokers appear unavailable simultaneously. <br>*Type: enum value*
+metadata.recovery.rebootstrap.trigger.ms |  *  | 0 .. 2147483647 |        300000 | low        | If a client configured to rebootstrap using `metadata.recovery.strategy=rebootstrap` is unable to obtain metadata from any of the brokers for this interval, client repeats the bootstrap process using `bootstrap.servers` configuration and brokers added through `rd_kafka_brokers_add()`. <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      |           100 | low        | When a topic loses its leader a new metadata request will be enqueued immediately and then with this initial interval, exponentially increasing upto `retry.backoff.max.ms`, until the topic metadata has been refreshed. If not set explicitly, it will be defaulted to `retry.backoff.ms`. This is used to recover quickly from transitioning leader brokers. <br>*Type: integer*
diff --git a/INTRODUCTION.md b/INTRODUCTION.md
@@ -2058,6 +2058,7 @@ The [Apache Kafka Implementation Proposals (KIPs)](https://cwiki.apache.org/conf
 | KIP-848 - The Next Generation of the Consumer Rebalance Protocol         | 3.7.0 (EA)                  | Early Access                                                                                  |
 | KIP-899 - Allow producer and consumer clients to rebootstrap             | 3.8.0                       | Supported                                                                                     |
 | KIP-951 - Leader discovery optimisations for the client                  | 3.7.0                       | Supported                                                                                     |
+| KIP-1102 - Enable clients to rebootstrap based on timeout or error code  | 4.0.0                       | Supported                                                                                     |
 
 
 
@@ -2074,7 +2075,7 @@ release of librdkafka.
 | 0       | Produce                       | 10         | 10             |
 | 1       | Fetch                         | 16         | 16             |
 | 2       | ListOffsets                   | 8          | 7              |
-| 3       | Metadata                      | 12         | 12             |
+| 3       | Metadata                      | 13         | 13             |
 | 8       | OffsetCommit                  | 9          | 9              |
 | 9       | OffsetFetch                   | 9          | 9              |
 | 10      | FindCoordinator               | 4          | 2              |
diff --git a/src/rdkafka.c b/src/rdkafka.c
@@ -718,6 +718,9 @@ static const struct rd_kafka_err_desc rd_kafka_err_descs[] = {
     _ERR_DESC(RD_KAFKA_RESP_ERR_TELEMETRY_TOO_LARGE,
               "Broker: Client sent a push telemetry request larger than the "
               "maximum size the broker will accept"),
+    _ERR_DESC(RD_KAFKA_RESP_ERR_REBOOTSTRAP_REQUIRED,
+              "Broker: Client metadata is stale, "
+              "client should rebootstrap to obtain new metadata."),
     _ERR_DESC(RD_KAFKA_RESP_ERR__END, NULL)};
 
 
@@ -2862,6 +2865,18 @@ void rd_kafka_rebootstrap(rd_kafka_t *rk) {
                                      rd_kafka_rebootstrap_tmr_cb, NULL);
 }
 
+/**
+ * Restarts rebootstrap timer with the configured interval.
+ *
+ * @locks none
+ * @locality any
+ */
+void rd_kafka_rebootstrap_tmr_restart(rd_kafka_t *rk) {
+        rd_kafka_timer_start_oneshot(
+            &rk->rk_timers, &rk->rebootstrap_tmr, rd_true /*restart*/,
+            rk->rk_conf.metadata_recovery_rebootstrap_trigger_ms * 1000LL,
+            rd_kafka_rebootstrap_tmr_cb, NULL);
+}
 
 /**
  * Counts usage of the legacy/simple consumer (rd_kafka_consume_start() with
diff --git a/src/rdkafka.h b/src/rdkafka.h
@@ -656,6 +656,9 @@ typedef enum {
         /** Client sent a push telemetry request larger than the maximum size
          *  the broker will accept. */
         RD_KAFKA_RESP_ERR_TELEMETRY_TOO_LARGE = 118,
+        /** Client metadata is stale,
+         *  client should rebootstrap to obtain new metadata. */
+        RD_KAFKA_RESP_ERR_REBOOTSTRAP_REQUIRED = 129,
         RD_KAFKA_RESP_ERR_END_ALL,
 } rd_kafka_resp_err_t;
 
diff --git a/src/rdkafka_conf.c b/src/rdkafka_conf.c
@@ -452,6 +452,17 @@ static const struct rd_kafka_property rd_kafka_properties[] = {
      .s2i = {{RD_KAFKA_METADATA_RECOVERY_STRATEGY_NONE, "none"},
              {RD_KAFKA_METADATA_RECOVERY_STRATEGY_REBOOTSTRAP, "rebootstrap"},
              {0, NULL}}},
+    {_RK_GLOBAL, "metadata.recovery.rebootstrap.trigger.ms", _RK_C_INT,
+     _RK(metadata_recovery_rebootstrap_trigger_ms),
+     "If a client configured to rebootstrap using "
+     "`metadata.recovery.strategy=rebootstrap` "
+     "is unable to obtain metadata from any "
+     "of the brokers for this interval, "
+     "client repeats the bootstrap process using "
+     "`bootstrap.servers` configuration "
+     "and brokers added through "
+     "`rd_kafka_brokers_add()`.",
+     0, INT_MAX, 300000},
     {_RK_GLOBAL | _RK_DEPRECATED | _RK_HIDDEN, "metadata.request.timeout.ms",
      _RK_C_INT, _RK(metadata_request_timeout_ms), "Not used.", 10, 900 * 1000,
      10},
diff --git a/src/rdkafka_conf.h b/src/rdkafka_conf.h
@@ -207,6 +207,7 @@ struct rd_kafka_conf_s {
         int msg_copy_max_size;
         int recv_max_msg_size;
         int max_inflight;
+        int metadata_recovery_rebootstrap_trigger_ms;
         int metadata_request_timeout_ms;
         int metadata_refresh_interval_ms;
         int metadata_refresh_fast_cnt;
diff --git a/src/rdkafka_int.h b/src/rdkafka_int.h
@@ -1237,5 +1237,6 @@ rd_kafka_resp_err_t rd_kafka_background_thread_create(rd_kafka_t *rk,
 
 void rd_kafka_rebootstrap(rd_kafka_t *rk);
 
+void rd_kafka_rebootstrap_tmr_restart(rd_kafka_t *rk);
 
 #endif /* _RDKAFKA_INT_H_ */
diff --git a/src/rdkafka_metadata.c b/src/rdkafka_metadata.c
@@ -574,7 +574,8 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
         int broker_changes            = 0;
         int cache_changes             = 0;
         int cgrp_subscription_version = -1;
-
+        int16_t ErrorCode             = 0;
+        
         /* If client rack is present, the metadata cache (topic or full) needs
          * to contain the partition to rack map. */
         rd_bool_t has_client_rack = rk->rk_conf.client_rack &&
@@ -874,6 +875,10 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
                     ClusterAuthorizedOperations;
         }
 
+        if (ApiVersion >= 13) {
+                rd_kafka_buf_read_i16(rkbuf, &ErrorCode);
+        }
+
         rd_kafka_buf_skip_tags(rkbuf);
 
         /* Entire Metadata response now parsed without errors:
@@ -886,6 +891,14 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
                 goto err;
         }
 
+        if (ErrorCode) {
+                rd_rkb_dbg(rkb, METADATA, "METADATA",
+                           "Received top level error code: %" PRId16,
+                           ErrorCode);
+                err = ErrorCode;
+                goto err;
+        }
+
         /* Update our list of brokers. */
         for (i = 0; i < md->broker_cnt; i++) {
                 rd_rkb_dbg(rkb, METADATA, "METADATA",
@@ -1035,6 +1048,7 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
         rd_kafka_wrlock(rkb->rkb_rk);
 
         rkb->rkb_rk->rk_ts_metadata = rd_clock();
+        rd_kafka_rebootstrap_tmr_restart(rkb->rkb_rk);
 
         /* Update cached cluster id. */
         if (RD_KAFKAP_STR_LEN(&cluster_id) > 0 &&
diff --git a/src/rdkafka_request.c b/src/rdkafka_request.c
@@ -2612,12 +2612,18 @@ static void rd_kafka_handle_Metadata(rd_kafka_t *rk,
         goto done;
 
 err:
-        actions = rd_kafka_err_action(rkb, err, request,
+        actions = rd_kafka_err_action(
+            rkb, err, request,
 
-                                      RD_KAFKA_ERR_ACTION_RETRY,
-                                      RD_KAFKA_RESP_ERR__PARTIAL,
+            RD_KAFKA_ERR_ACTION_SPECIAL, RD_KAFKA_RESP_ERR_REBOOTSTRAP_REQUIRED,
 
-                                      RD_KAFKA_ERR_ACTION_END);
+            RD_KAFKA_ERR_ACTION_RETRY, RD_KAFKA_RESP_ERR__PARTIAL,
+
+            RD_KAFKA_ERR_ACTION_END);
+
+        if (actions & RD_KAFKA_ERR_ACTION_SPECIAL) {
+                rd_kafka_rebootstrap(rk);
+        }
 
         if (actions & RD_KAFKA_ERR_ACTION_RETRY) {
                 /* In case it's a brokers full refresh call,
@@ -2747,7 +2753,7 @@ rd_kafka_MetadataRequest0(rd_kafka_broker_t *rkb,
         int *full_incr                 = NULL;
         void *handler_arg              = NULL;
         rd_kafka_resp_cb_t *handler_cb = rd_kafka_handle_Metadata;
-        int16_t metadata_max_version   = 12;
+        int16_t metadata_max_version   = 13;
         rd_kafka_replyq_t use_replyq   = replyq;
 
         /* In case we want cluster authorized operations in the Metadata

Original file line number	Diff line number	Diff line change
`@@ -1237,5 +1237,6 @@ rd_kafka_resp_err_t rd_kafka_background_thread_create(rd_kafka_t *rk,`
`1237`	`1237`
`1238`	`1238`	`void rd_kafka_rebootstrap(rd_kafka_t *rk);`
`1239`	`1239`
	`1240`	`+void rd_kafka_rebootstrap_tmr_restart(rd_kafka_t *rk);`
`1240`	`1241`
`1241`	`1242`	`#endif /* _RDKAFKA_INT_H_ */`