[KIP-848] fix generated memberid uniqueness (confluentinc#5101)

Author: pranavrth

src/rdrand.c

@@ -30,6 +30,7 @@
 #include "rdrand.h"
 #include "rdtime.h"
 #include "tinycthread.h"
+#include "rdmurmur2.h"
 
 int rd_jitter(int low, int high) {
         int rand_num;
@@ -40,8 +41,17 @@ int rd_jitter(int low, int high) {
         if (unlikely(seed == 0)) {
                 struct timeval tv;
                 rd_gettimeofday(&tv, NULL);
-                seed = (unsigned int)(tv.tv_usec / 1000);
+                seed = (unsigned int)(tv.tv_usec);
                 seed ^= (unsigned int)(intptr_t)thrd_current();
+
+                /* When many threads are created at the same time and the
+                 * thread id is different only by a few bits it's possible that
+                 * `rand_r`, that is initially multiplying by `1103515245`,
+                 * truncates the variable bits and uses the same seed for
+                 * different threads. By applying `murmur2` we ensure that seed
+                 * variability is distributed across various bits at different
+                 * positions. */
+                seed = (unsigned int)rd_murmur2(&seed, sizeof(seed));
         }
 
         rand_num = rand_r(&seed);

tests/0153-memberid.c

@@ -0,0 +1,128 @@
+/*
+ * librdkafka - Apache Kafka C library
+ *
+ * Copyright (c) 2025, Confluent Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright notice,
+ *    this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright notice,
+ *    this list of conditions and the following disclaimer in the documentation
+ *    and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "test.h"
+typedef struct consumer_s {
+        const char *group_id;
+        char *memberid;
+} consumer_t;
+
+static int
+is_fatal_cb(rd_kafka_t *rk, rd_kafka_resp_err_t err, const char *reason) {
+        if (err == RD_KAFKA_RESP_ERR__TIMED_OUT)
+                return 0;
+        return 1;
+}
+
+static int consumer_thread(void *arg) {
+        rd_kafka_conf_t *conf;
+        rd_kafka_t *consumer;
+        consumer_t *consumer_args = arg;
+
+        test_curr->is_fatal_cb = is_fatal_cb;
+
+        test_conf_init(&conf, NULL, 60);
+
+        consumer =
+            test_create_consumer(consumer_args->group_id, NULL, conf, NULL);
+
+        consumer_args->memberid = rd_kafka_memberid(consumer);
+
+        test_consumer_close(consumer);
+        rd_kafka_destroy(consumer);
+        test_curr->is_fatal_cb = NULL;
+        return 0;
+}
+
+void do_test_unique_memberid() {
+        int i;
+        int j;
+        int have_only_unique_memberids = 1;
+        const char *group_id           = test_mk_topic_name("0153-memberid", 1);
+
+#define CONSUMER_CNT 500
+        thrd_t thread_id[CONSUMER_CNT];
+        consumer_t consumer_args[CONSUMER_CNT];
+
+        SUB_TEST_QUICK();
+
+        for (i = 0; i < CONSUMER_CNT; i++) {
+                consumer_args[i].group_id = group_id;
+                consumer_args[i].memberid = NULL;
+                thrd_create(&thread_id[i], consumer_thread, &consumer_args[i]);
+        }
+
+        for (i = 0; i < CONSUMER_CNT; i++) {
+                thrd_join(thread_id[i], NULL);
+        }
+
+        for (i = 0; i < CONSUMER_CNT; i++) {
+                if (have_only_unique_memberids) {
+                        for (j = i + 1; j < CONSUMER_CNT; j++) {
+                                if (strcmp(consumer_args[i].memberid,
+                                           consumer_args[j].memberid) == 0) {
+                                        TEST_SAY(
+                                            "Consumer %d has the same member "
+                                            "ID as consumer %d: %s\n",
+                                            i, j, consumer_args[i].memberid);
+                                        have_only_unique_memberids = 0;
+                                        break;
+                                }
+                        }
+                }
+                rd_free(consumer_args[i].memberid);
+        }
+
+        if (have_only_unique_memberids) {
+                TEST_SAY("All %d consumers have unique member IDs\n",
+                         CONSUMER_CNT);
+        } else {
+                TEST_FAIL("Not all consumers have unique member IDs\n");
+        }
+
+        SUB_TEST_PASS();
+}
+
+int main_0153_memberid(int argc, char **argv) {
+
+        if (test_consumer_group_protocol_classic()) {
+                TEST_SKIP(
+                    "Member ID is not generated on the client side in classic "
+                    "protocol, skipping test");
+                return 0;
+        }
+
+        if (!strcmp(test_mode, "valgrind")) {
+                TEST_SKIP("Test is too heavy for valgrind, skipping it");
+                return 0;
+        }
+
+        do_test_unique_memberid();
+
+        return 0;
+}

tests/CMakeLists.txt

@@ -142,6 +142,7 @@ set(
     0150-telemetry_mock.c
     0151-purge-brokers.c
     0152-rebootstrap.c
+    0153-memberid.c
     8000-idle.cpp
     8001-fetch_from_follower_mock_manual.c
     test.c

tests/test.c

@@ -269,6 +269,7 @@ _TEST_DECL(0149_broker_same_host_port_mock);
 _TEST_DECL(0150_telemetry_mock);
 _TEST_DECL(0151_purge_brokers_mock);
 _TEST_DECL(0152_rebootstrap_local);
+_TEST_DECL(0153_memberid);
 
 /* Manual tests */
 _TEST_DECL(8000_idle);
@@ -534,6 +535,7 @@ struct test tests[] = {
     _TEST(0150_telemetry_mock, 0),
     _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)),
 
 
     /* Manual tests */

win32/tests/tests.vcxproj

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