Move test

Author: PratRanj07

src/rdkafka_cgrp.c

@@ -2914,7 +2914,7 @@ rd_kafka_cgrp_consumer_assignment_with_metadata(
     rd_kafka_cgrp_t *rkcg,
     rd_kafka_topic_partition_list_t *assignment,
     rd_list_t **missing_topic_ids) {
-        int i;
+        int i, j;
         rd_kafka_t *rk = rkcg->rkcg_rk;
         rd_kafka_topic_partition_list_t *assignment_with_metadata =
             rd_kafka_topic_partition_list_new(assignment->cnt);
@@ -2935,8 +2935,8 @@ rd_kafka_cgrp_consumer_assignment_with_metadata(
                 if (rkmce) {
                         topic_name = rd_strdup(rkmce->rkmce_mtopic.topic);
                         // Check if partition exists in metadata
-                        for (int j = 0; j < rkmce->rkmce_mtopic.partition_cnt;
-                             j++) {
+                        int pcnt = rkmce->rkmce_mtopic.partition_cnt;
+                        for (j = 0; j < pcnt; j++) {
                                 if (rkmce->rkmce_mtopic.partitions[j].id ==
                                     partition) {
                                         partition_found = rd_true;

tests/0147-consumer_group_consumer_mock.c

@@ -27,10 +27,13 @@
  */
 
 #include "test.h"
+#include "rdkafka.h"
 
 #include "../src/rdkafka_proto.h"
 
 #include <stdarg.h>
+#include <stdio.h>
+#include <string.h>
 
 
 /**
@@ -928,6 +931,122 @@ static void do_test_quick_unsubscribe_tests(void) {
         }
 }
 
+/**
+ * Integration test for KIP-848 partition metadata refresh:
+ * - Create topic with 2 partitions
+ * - Start consumer group and verify initial assignment
+ * - Increase partition count to 4
+ * - Reset log tracking variables after partition creation
+ * - Wait for HeartbeatRequest, HeartbeatResponse, and metadata refresh logs
+ * - Assert that metadata refresh is triggered for new partitions
+ */
+
+// Globals to track log sequence
+static volatile int seen_heartbeat_req  = 0;
+static volatile int seen_heartbeat_resp = 0;
+static volatile int seen_metadata_log   = 0;
+
+static void reset_log_tracking(void) {
+        seen_heartbeat_req  = 0;
+        seen_heartbeat_resp = 0;
+        seen_metadata_log   = 0;
+}
+
+static void wait_for_metadata_refresh_log(int timeout_ms) {
+        int elapsed = 0;
+        while (elapsed < timeout_ms && !seen_metadata_log) {
+                rd_usleep(500 * 1000, NULL);  // 500 ms
+                elapsed += 500;
+        }
+        TEST_ASSERT(
+            seen_heartbeat_req,
+            "Expected HeartbeatRequest log not seen after partition creation");
+        TEST_ASSERT(
+            seen_heartbeat_resp,
+            "Expected HeartbeatResponse log not seen after partition creation");
+        TEST_ASSERT(seen_metadata_log,
+                    "Expected metadata refresh log not seen after partition "
+                    "creation and heartbeat");
+}
+
+// Custom log callback to capture and process librdkafka logs
+static void test_metadata_log_cb(const rd_kafka_t *rk,
+                                 int level,
+                                 const char *fac,
+                                 const char *buf) {
+        if (strstr(buf, "Sent ConsumerGroupHeartbeatRequest")) {
+                seen_heartbeat_req = 1;
+        }
+        if (seen_heartbeat_req &&
+            strstr(buf, "Received ConsumerGroupHeartbeatResponse")) {
+                seen_heartbeat_resp = 1;
+        }
+        if (seen_heartbeat_resp &&
+            strstr(buf,
+                   "Partition assigned to this consumer is not present in "
+                   "cached metadata")) {
+                seen_metadata_log = 1;
+        }
+}
+
+static rd_kafka_t *create_consumers(const char *group) {
+        rd_kafka_conf_t *conf;
+        test_conf_init(&conf, NULL, 60);
+        test_conf_set(conf, "group.id", group);
+        test_conf_set(conf, "auto.offset.reset", "earliest");
+        test_conf_set(conf, "debug", "cgrp, protocol");
+        rd_kafka_conf_set_log_cb(conf, test_metadata_log_cb);
+        rd_kafka_t *consumer = test_create_consumer(group, NULL, conf, NULL);
+        return consumer;
+}
+
+static void do_test_setup_and_run_metadata_refresh_test(void) {
+        const char *topic      = test_mk_topic_name("cgrp_metadata", 1);
+        int initial_partitions = 2;
+        int new_partitions     = 4;
+        rd_kafka_t *c1, *c2, *rk;
+        const char *group = "grp_metadata";
+
+        SUB_TEST_QUICK();
+
+        TEST_SAY("Creating topic %s with %d partitions\n", topic,
+                 initial_partitions);
+        test_create_topic(NULL, topic, initial_partitions, 1);
+
+        TEST_SAY("Creating consumers\n");
+        c1 = create_consumers(group);
+        c2 = create_consumers(group);
+
+        rk = test_create_handle(RD_KAFKA_PRODUCER, NULL);
+
+        TEST_SAY("Subscribing to topic %s\n", topic);
+        test_consumer_subscribe(c1, topic);
+        test_consumer_subscribe(c2, topic);
+
+        // Wait for initial assignment
+        test_consumer_wait_assignment(c1, rd_false);
+        test_consumer_wait_assignment(c2, rd_false);
+
+        // Create new partitions
+        TEST_SAY("Increasing partition count to %d\n", new_partitions);
+        test_create_partitions(rk, topic, new_partitions);
+
+        // Reset log tracking variables to only consider logs after partition
+        // creation
+        reset_log_tracking();
+
+        // Wait for expected logs for up to 10 seconds
+        wait_for_metadata_refresh_log(10000);
+
+        TEST_SAY("Closing consumers\n");
+        test_consumer_close(c1);
+        test_consumer_close(c2);
+        rd_kafka_destroy(c1);
+        rd_kafka_destroy(c2);
+
+        SUB_TEST_PASS();
+}
+
 int main_0147_consumer_group_consumer_mock(int argc, char **argv) {
         TEST_SKIP_MOCK_CLUSTER(0);
 
@@ -948,5 +1067,7 @@ int main_0147_consumer_group_consumer_mock(int argc, char **argv) {
 
         do_test_quick_unsubscribe_tests();
 
+        do_test_setup_and_run_metadata_refresh_test();
+
         return 0;
 }

tests/0154-metadata_refresh.c

@@ -143,7 +143,6 @@ set(
     0151-purge-brokers.c
     0152-rebootstrap.c
     0153-memberid.c
-    0154-metadata_refresh.c
     8000-idle.cpp
     8001-fetch_from_follower_mock_manual.c
     test.c

tests/CMakeLists.txt

@@ -270,7 +270,6 @@ _TEST_DECL(0150_telemetry_mock);
 _TEST_DECL(0151_purge_brokers_mock);
 _TEST_DECL(0152_rebootstrap_local);
 _TEST_DECL(0153_memberid);
-_TEST_DECL(0154_metadata_refresh);
 
 /* Manual tests */
 _TEST_DECL(8000_idle);
@@ -537,7 +536,6 @@ struct test tests[] = {
     _TEST(0151_purge_brokers_mock, TEST_F_LOCAL),
     _TEST(0152_rebootstrap_local, TEST_F_LOCAL),
     _TEST(0153_memberid, 0, TEST_BRKVER(0, 4, 0, 0)),
-    _TEST(0154_metadata_refresh, 0, TEST_BRKVER(0, 11, 0, 0)),
 
 
     /* Manual tests */

tests/test.c

@@ -233,7 +233,6 @@
     <ClCompile Include="..\..\tests\0151-purge-brokers.c" />
     <ClCompile Include="..\..\tests\0152-rebootstrap.c" />
     <ClCompile Include="..\..\tests\0153-memberid.c" />
-    <ClCompile Include="..\..\tests\0154-metadata_refresh.c" />
     <ClCompile Include="..\..\tests\8000-idle.cpp" />
     <ClCompile Include="..\..\tests\8001-fetch_from_follower_mock_manual.c" />
     <ClCompile Include="..\..\tests\test.c" />