Subscription version to avoid stale metadata

emasab · emasab · commit 53fb83dcff54 · 2025-02-19T13:52:59.000+01:00
calls cause an unknown topic or partition error
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -24,6 +24,8 @@ librdkafka v2.9.0 is a feature release:
  * When consumer is closed before destroying the client, the operations queue
    isn't purged anymore as it contains operations
    unrelated to the consumer group (#).
+ * When making multiple changes to the consumer subscription in a short time,
+   no unknown topic error is returned for topics that are in the new subscription but weren't in previous one (#).
 
 
 ## Fixes
@@ -81,6 +83,12 @@ librdkafka v2.9.0 is a feature release:
    isn't purged anymore as it contains operations
    unrelated to the consumer group.
    Happens since 1.x (#).
+ * Issues: #
+   When making multiple changes to the consumer subscription in a short time,
+   no unknown topic error is returned for topics that are in the new subscription
+   but weren't in previous one. This was due to the metadata request relative
+   to previous subscription.
+   Happens since 1.x (#).
 
 
 
diff --git a/src/rdkafka_admin.c b/src/rdkafka_admin.c
@@ -1511,8 +1511,9 @@ rd_kafka_admin_MetadataRequest(rd_kafka_broker_t *rkb,
             rd_false /* No admin operation requires topic creation. */,
             include_cluster_authorized_operations,
             include_topic_authorized_operations,
-            rd_false /* No admin operation should update cgrp. */, force_racks,
-            resp_cb, replyq,
+            rd_false /* No admin operation should update cgrp. */,
+            -1 /* No subscription version is used */, force_racks, resp_cb,
+            replyq,
             rd_true /* Admin operation metadata requests are always forced. */,
             opaque);
 }
diff --git a/src/rdkafka_buf.h b/src/rdkafka_buf.h
@@ -384,6 +384,9 @@ struct rd_kafka_buf_s { /* rd_kafka_buf_t */
                         rd_bool_t all_topics;  /**< Full/All topics requested */
                         rd_bool_t cgrp_update; /**< Update cgrp with topic
                                                 *   status from response. */
+                        int32_t cgrp_subscription_version;
+                        /**< Consumer group subscription version, to
+                         *   check before updating cgrp state. */
                         rd_bool_t force_racks; /**< Force the returned metadata
                                                 *   to contain partition to
                                                 *   rack mapping. */
diff --git a/src/rdkafka_cgrp.c b/src/rdkafka_cgrp.c
@@ -476,6 +476,7 @@ rd_kafka_cgrp_t *rd_kafka_cgrp_new(rd_kafka_t *rk,
         rd_interval_init(&rkcg->rkcg_timeout_scan_intvl);
         rd_atomic32_init(&rkcg->rkcg_assignment_lost, rd_false);
         rd_atomic32_init(&rkcg->rkcg_terminated, rd_false);
+        rd_atomic32_init(&rkcg->rkcg_subscription_version, 0);
         rkcg->rkcg_current_assignment = rd_kafka_topic_partition_list_new(0);
         rkcg->rkcg_target_assignment  = NULL;
         rkcg->rkcg_next_target_assignment = NULL;
@@ -2232,6 +2233,8 @@ static void rd_kafka_cgrp_handle_JoinGroup(rd_kafka_t *rk,
                      * on receiving the response since some topics
                      * may be missing. */
                     rd_false,
+                    /* cgrp_update=false: no subscription version is used */
+                    -1,
                     /* force_racks is true if any memeber has a client rack set,
                        since we will require partition to rack mapping in that
                        case for rack-aware assignors. */
@@ -2336,7 +2339,10 @@ static rd_kafka_op_res_t rd_kafka_cgrp_handle_Metadata_op(rd_kafka_t *rk,
         if (rko->rko_err == RD_KAFKA_RESP_ERR__DESTROY)
                 return RD_KAFKA_OP_RES_HANDLED; /* Terminating */
 
-        rd_kafka_cgrp_metadata_update_check(rkcg, rd_false /*dont rejoin*/);
+        if (rd_atomic32_get(&rkcg->rkcg_subscription_version) ==
+            rko->rko_u.metadata.subscription_version)
+                rd_kafka_cgrp_metadata_update_check(rkcg,
+                                                    rd_false /*dont rejoin*/);
 
         return RD_KAFKA_OP_RES_HANDLED;
 }
@@ -2353,6 +2359,7 @@ static rd_kafka_op_res_t rd_kafka_cgrp_handle_Metadata_op(rd_kafka_t *rk,
  */
 static int rd_kafka_cgrp_metadata_refresh(rd_kafka_cgrp_t *rkcg,
                                           int *metadata_agep,
+                                          int32_t cgrp_subscription_version,
                                           const char *reason) {
         rd_kafka_t *rk = rkcg->rkcg_rk;
         rd_kafka_op_t *rko;
@@ -2421,9 +2428,9 @@ static int rd_kafka_cgrp_metadata_refresh(rd_kafka_cgrp_t *rkcg,
                                  rd_kafka_cgrp_handle_Metadata_op);
         rd_kafka_op_set_replyq(rko, rkcg->rkcg_ops, 0);
 
-        err = rd_kafka_metadata_request(rkcg->rkcg_rk, NULL, &topics,
-                                        rd_false /*!allow auto create */,
-                                        rd_true /*cgrp_update*/, reason, rko);
+        err = rd_kafka_metadata_request(
+            rkcg->rkcg_rk, NULL, &topics, rd_false /*!allow auto create */,
+            rd_true /*cgrp_update*/, cgrp_subscription_version, reason, rko);
         if (err) {
                 rd_kafka_dbg(rk, CGRP | RD_KAFKA_DBG_METADATA, "CGRPMETADATA",
                              "%s: need to refresh metadata (%dms old) "
@@ -2439,7 +2446,8 @@ static int rd_kafka_cgrp_metadata_refresh(rd_kafka_cgrp_t *rkcg,
 
 
 
-static void rd_kafka_cgrp_join(rd_kafka_cgrp_t *rkcg) {
+static void rd_kafka_cgrp_join(rd_kafka_cgrp_t *rkcg,
+                               int32_t cgrp_subscription_version) {
         int metadata_age;
 
         if (rkcg->rkcg_state != RD_KAFKA_CGRP_STATE_UP ||
@@ -2475,6 +2483,7 @@ static void rd_kafka_cgrp_join(rd_kafka_cgrp_t *rkcg) {
         /* We need up-to-date full metadata to continue,
          * refresh metadata if necessary. */
         if (rd_kafka_cgrp_metadata_refresh(rkcg, &metadata_age,
+                                           cgrp_subscription_version,
                                            "consumer join") == 1) {
                 rd_kafka_dbg(rkcg->rkcg_rk, CGRP | RD_KAFKA_DBG_CONSUMER,
                              "JOIN",
@@ -2879,7 +2888,8 @@ void rd_kafka_cgrp_consumer_next_target_assignment_request_metadata(
         rd_kafka_op_set_replyq(rko, rkcg->rkcg_ops, NULL);
         rd_kafka_MetadataRequest(
             rkb, NULL, missing_topic_ids, "ConsumerGroupHeartbeat API Response",
-            rd_false /*!allow_auto_create*/, rd_false, rd_false, rko);
+            rd_false /*!allow_auto_create*/, rd_false,
+            -1 /* no subscription version is used */, rd_false, rko);
         rd_list_destroy(missing_topic_ids);
 }
 
@@ -5136,9 +5146,10 @@ rd_kafka_cgrp_calculate_subscribe_revoking_partitions(
         return revoking;
 }
 
-static void
+static int32_t
 rd_kafka_cgrp_subscription_set(rd_kafka_cgrp_t *rkcg,
                                rd_kafka_topic_partition_list_t *rktparlist) {
+        int32_t ret = rd_atomic32_add(&rkcg->rkcg_subscription_version, 1);
         rkcg->rkcg_subscription = rktparlist;
         if (rkcg->rkcg_subscription) {
                 /* Insert all non-wildcard topics in cache immediately.
@@ -5149,6 +5160,7 @@ rd_kafka_cgrp_subscription_set(rd_kafka_cgrp_t *rkcg,
                     rkcg->rkcg_rk, rkcg->rkcg_subscription, NULL,
                     0 /*dont replace*/);
         }
+        return ret;
 }
 
 /**
@@ -5166,6 +5178,7 @@ rd_kafka_cgrp_modify_subscription(rd_kafka_cgrp_t *rkcg,
         rd_kafka_topic_partition_list_t *errored;
         int metadata_age;
         int old_cnt = rkcg->rkcg_subscription->cnt;
+        int32_t cgrp_subscription_version;
 
         rkcg->rkcg_flags &= ~RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION;
 
@@ -5182,9 +5195,11 @@ rd_kafka_cgrp_modify_subscription(rd_kafka_cgrp_t *rkcg,
             rkcg, unsubscribing_topics);
 
         rd_kafka_topic_partition_list_destroy(rkcg->rkcg_subscription);
-        rd_kafka_cgrp_subscription_set(rkcg, rktparlist);
+        cgrp_subscription_version =
+            rd_kafka_cgrp_subscription_set(rkcg, rktparlist);
 
         if (rd_kafka_cgrp_metadata_refresh(rkcg, &metadata_age,
+                                           cgrp_subscription_version,
                                            "modify subscription") == 1) {
                 rd_kafka_dbg(rkcg->rkcg_rk, CGRP | RD_KAFKA_DBG_CONSUMER,
                              "MODSUB",
@@ -5323,7 +5338,7 @@ static rd_kafka_resp_err_t rd_kafka_cgrp_unsubscribe(rd_kafka_cgrp_t *rkcg,
 static rd_kafka_resp_err_t
 rd_kafka_cgrp_subscribe(rd_kafka_cgrp_t *rkcg,
                         rd_kafka_topic_partition_list_t *rktparlist) {
-
+        int32_t subscription_version;
         rd_kafka_dbg(rkcg->rkcg_rk, CGRP | RD_KAFKA_DBG_CONSUMER, "SUBSCRIBE",
                      "Group \"%.*s\": subscribe to new %ssubscription "
                      "of %d topics (join-state %s)",
@@ -5389,9 +5404,9 @@ rd_kafka_cgrp_subscribe(rd_kafka_cgrp_t *rkcg,
         if (rd_kafka_topic_partition_list_regex_cnt(rktparlist) > 0)
                 rkcg->rkcg_flags |= RD_KAFKA_CGRP_F_WILDCARD_SUBSCRIPTION;
 
-        rd_kafka_cgrp_subscription_set(rkcg, rktparlist);
+        subscription_version = rd_kafka_cgrp_subscription_set(rkcg, rktparlist);
 
-        rd_kafka_cgrp_join(rkcg);
+        rd_kafka_cgrp_join(rkcg, subscription_version);
 
         return RD_KAFKA_RESP_ERR_NO_ERROR;
 }
@@ -5749,7 +5764,8 @@ static void rd_kafka_cgrp_join_state_serve(rd_kafka_cgrp_t *rkcg) {
 
                 if (rd_interval_immediate(&rkcg->rkcg_join_intvl, 1000 * 1000,
                                           now) > 0)
-                        rd_kafka_cgrp_join(rkcg);
+                        rd_kafka_cgrp_join(
+                            rkcg, -1 /* current subscription version */);
                 break;
 
         case RD_KAFKA_CGRP_JOIN_STATE_WAIT_JOIN:
diff --git a/src/rdkafka_cgrp.h b/src/rdkafka_cgrp.h
@@ -333,6 +333,8 @@ typedef struct rd_kafka_cgrp_s {
 
         rd_atomic32_t rkcg_terminated; /**< Consumer has been closed */
 
+        rd_atomic32_t rkcg_subscription_version; /**< Subscription version */
+
         /* Protected by rd_kafka_*lock() */
         struct {
                 rd_ts_t ts_rebalance;       /* Timestamp of
@@ -403,6 +405,12 @@ void rd_kafka_cgrp_metadata_update_check(rd_kafka_cgrp_t *rkcg,
                                          rd_bool_t do_join);
 #define rd_kafka_cgrp_get(rk) ((rk)->rk_cgrp)
 
+#define rd_kafka_cgrp_same_subscription_version(rk_cgrp,                       \
+                                                cgrp_subscription_version)     \
+        ((rk_cgrp) &&                                                          \
+         (cgrp_subscription_version == -1 ||                                   \
+          rd_atomic32_get(&(rk_cgrp)->rkcg_subscription_version) ==            \
+              cgrp_subscription_version))
 
 void rd_kafka_cgrp_assigned_offsets_commit(
     rd_kafka_cgrp_t *rkcg,
diff --git a/src/rdkafka_metadata.c b/src/rdkafka_metadata.c
@@ -135,6 +135,7 @@ rd_kafka_metadata(rd_kafka_t *rk,
                                   * partial request may make it seem
                                   * like some subscribed topics are missing. */
                                  all_topics ? rd_true : rd_false,
+                                 -1 /* same subscription version */,
                                  rd_false /* force_racks */, rko);
 
         rd_list_destroy(&topics);
@@ -484,12 +485,13 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
         rd_bool_t cgrp_update                = rd_false;
         rd_bool_t has_reliable_leader_epochs =
             rd_kafka_has_reliable_leader_epochs(rkb);
-        int ApiVersion             = rkbuf->rkbuf_reqhdr.ApiVersion;
-        rd_kafkap_str_t cluster_id = RD_ZERO_INIT;
-        int32_t controller_id      = -1;
-        rd_kafka_resp_err_t err    = RD_KAFKA_RESP_ERR_NO_ERROR;
-        int broker_changes         = 0;
-        int cache_changes          = 0;
+        int ApiVersion                = rkbuf->rkbuf_reqhdr.ApiVersion;
+        rd_kafkap_str_t cluster_id    = RD_ZERO_INIT;
+        int32_t controller_id         = -1;
+        rd_kafka_resp_err_t err       = RD_KAFKA_RESP_ERR_NO_ERROR;
+        int broker_changes            = 0;
+        int cache_changes             = 0;
+        int cgrp_subscription_version = -1;
 
         /* If client rack is present, the metadata cache (topic or full) needs
          * to contain the partition to rack map. */
@@ -504,6 +506,8 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
                 cgrp_update =
                     request->rkbuf_u.Metadata.cgrp_update && rk->rk_cgrp;
                 compute_racks |= request->rkbuf_u.Metadata.force_racks;
+                cgrp_subscription_version =
+                    request->rkbuf_u.Metadata.cgrp_subscription_version;
         }
 
         /* If there's reason is NULL, set it to a human-readable string. */
@@ -1031,9 +1035,13 @@ rd_kafka_parse_Metadata0(rd_kafka_broker_t *rkb,
          * the effective subscription of available topics) as to not
          * propagate non-included topics as non-existent. */
         if (cgrp_update &&
-            (requested_topics || requested_topic_ids || all_topics))
+            (all_topics ||
+             ((requested_topics || requested_topic_ids) &&
+              rd_kafka_cgrp_same_subscription_version(
+                  rkb->rkb_rk->rk_cgrp, cgrp_subscription_version)))) {
                 rd_kafka_cgrp_metadata_update_check(rkb->rkb_rk->rk_cgrp,
                                                     rd_true /*do join*/);
+        }
 
         /* Try to acquire a Producer ID from this broker if we
          * don't have one. */
@@ -1347,6 +1355,7 @@ rd_kafka_metadata_refresh_topics(rd_kafka_t *rk,
                                  rd_bool_t force,
                                  rd_bool_t allow_auto_create,
                                  rd_bool_t cgrp_update,
+                                 int32_t cgrp_subscription_version,
                                  const char *reason) {
         rd_list_t q_topics;
         int destroy_rkb = 0;
@@ -1413,9 +1422,9 @@ rd_kafka_metadata_refresh_topics(rd_kafka_t *rk,
                      "Requesting metadata for %d/%d topics: %s",
                      rd_list_cnt(&q_topics), rd_list_cnt(topics), reason);
 
-        rd_kafka_MetadataRequest(rkb, &q_topics, NULL, reason,
-                                 allow_auto_create, cgrp_update,
-                                 rd_false /* force_racks */, NULL);
+        rd_kafka_MetadataRequest(
+            rkb, &q_topics, NULL, reason, allow_auto_create, cgrp_update,
+            cgrp_subscription_version, rd_false /* force_racks */, NULL);
 
         rd_list_destroy(&q_topics);
 
@@ -1464,7 +1473,7 @@ rd_kafka_metadata_refresh_known_topics(rd_kafka_t *rk,
         else
                 err = rd_kafka_metadata_refresh_topics(
                     rk, rkb, &topics, force, allow_auto_create_topics,
-                    rd_false /*!cgrp_update*/, reason);
+                    rd_false /*!cgrp_update*/, -1, reason);
 
         rd_list_destroy(&topics);
 
@@ -1530,7 +1539,8 @@ rd_kafka_metadata_refresh_consumer_topics(rd_kafka_t *rk,
         else
                 err = rd_kafka_metadata_refresh_topics(
                     rk, rkb, &topics, rd_true /*force*/,
-                    allow_auto_create_topics, rd_true /*cgrp_update*/, reason);
+                    allow_auto_create_topics, rd_true /*cgrp_update*/,
+                    rd_atomic32_get(&rkcg->rkcg_subscription_version), reason);
 
         rd_list_destroy(&topics);
 
@@ -1557,8 +1567,9 @@ rd_kafka_resp_err_t rd_kafka_metadata_refresh_brokers(rd_kafka_t *rk,
                                                       const char *reason) {
         return rd_kafka_metadata_request(rk, rkb, NULL /*brokers only*/,
                                          rd_false /*!allow auto create topics*/,
-                                         rd_false /*no cgrp update */, reason,
-                                         NULL);
+                                         rd_false /*no cgrp update */,
+                                         -1 /* same subscription version */,
+                                         reason, NULL);
 }
 
 
@@ -1592,7 +1603,8 @@ rd_kafka_resp_err_t rd_kafka_metadata_refresh_all(rd_kafka_t *rk,
         rd_list_init(&topics, 0, NULL); /* empty list = all topics */
         rd_kafka_MetadataRequest(
             rkb, &topics, NULL, reason, rd_false /*no auto create*/,
-            rd_true /*cgrp update*/, rd_false /* force_rack */, NULL);
+            rd_true /*cgrp update*/, -1 /* same subscription version */,
+            rd_false /* force_rack */, NULL);
         rd_list_destroy(&topics);
 
         if (destroy_rkb)
@@ -1618,6 +1630,7 @@ rd_kafka_metadata_request(rd_kafka_t *rk,
                           const rd_list_t *topics,
                           rd_bool_t allow_auto_create_topics,
                           rd_bool_t cgrp_update,
+                          int32_t cgrp_subscription_version,
                           const char *reason,
                           rd_kafka_op_t *rko) {
         int destroy_rkb = 0;
@@ -1629,9 +1642,9 @@ rd_kafka_metadata_request(rd_kafka_t *rk,
                 destroy_rkb = 1;
         }
 
-        rd_kafka_MetadataRequest(rkb, topics, NULL, reason,
-                                 allow_auto_create_topics, cgrp_update,
-                                 rd_false /* force racks */, rko);
+        rd_kafka_MetadataRequest(
+            rkb, topics, NULL, reason, allow_auto_create_topics, cgrp_update,
+            cgrp_subscription_version, rd_false /* force racks */, rko);
 
         if (destroy_rkb)
                 rd_kafka_broker_destroy(rkb);
@@ -1695,7 +1708,7 @@ static void rd_kafka_metadata_leader_query_tmr_cb(rd_kafka_timers_t *rkts,
                 rd_kafka_metadata_refresh_topics(
                     rk, NULL, &topics, rd_true /*force*/,
                     rk->rk_conf.allow_auto_create_topics,
-                    rd_false /*!cgrp_update*/, "partition leader query");
+                    rd_false /*!cgrp_update*/, -1, "partition leader query");
 
                 /* Back off next query exponentially till we reach
                  * the retry backoff max ms */
diff --git a/src/rdkafka_metadata.h b/src/rdkafka_metadata.h
@@ -147,6 +147,7 @@ rd_kafka_metadata_refresh_topics(rd_kafka_t *rk,
                                  rd_bool_t force,
                                  rd_bool_t allow_auto_create,
                                  rd_bool_t cgrp_update,
+                                 int32_t cgrp_subscription_version,
                                  const char *reason);
 rd_kafka_resp_err_t
 rd_kafka_metadata_refresh_known_topics(rd_kafka_t *rk,
@@ -170,6 +171,7 @@ rd_kafka_metadata_request(rd_kafka_t *rk,
                           const rd_list_t *topics,
                           rd_bool_t allow_auto_create_topics,
                           rd_bool_t cgrp_update,
+                          int32_t cgrp_subscription_version,
                           const char *reason,
                           rd_kafka_op_t *rko);
 
diff --git a/src/rdkafka_op.h b/src/rdkafka_op.h
@@ -400,6 +400,8 @@ struct rd_kafka_op_s {
                 struct {
                         rd_kafka_metadata_t *md;
                         rd_kafka_metadata_internal_t *mdi;
+                        /* subscription version for this call */
+                        int32_t subscription_version;
                         int force; /* force request regardless of outstanding
                                     * metadata requests. */
                 } metadata;
diff --git a/src/rdkafka_partition.c b/src/rdkafka_partition.c
diff --git a/src/rdkafka_request.c b/src/rdkafka_request.c
diff --git a/src/rdkafka_request.h b/src/rdkafka_request.h
diff --git a/src/rdkafka_topic.c b/src/rdkafka_topic.c
