[KIP-848] [mock cluster] Improved static group membership implementation (#5030)

When member is leaving with -2 a new member that is joining can replace the previous member id with the new one. If previous member didn't leave the new member joining with same group.instance.id is fenced in new protocol.

Author: emasab

src/rdkafka_mock.c

@@ -3156,7 +3156,8 @@ static int ut_cgrp_consumer_member_next_assignment0(
                           fixtures[i].comment);
 
                 if (fixtures[i].session_timed_out) {
-                        rd_kafka_mock_cgrp_consumer_member_leave(mcgrp, member);
+                        rd_kafka_mock_cgrp_consumer_member_leave(mcgrp, member,
+                                                                 rd_false);
                         member = rd_kafka_mock_cgrp_consumer_member_add(
                             mcgrp, conn, &MemberId, &InstanceId,
                             &SubscribedTopic, 1, &SubscribedTopicRegex);

src/rdkafka_mock_cgrp.c

@@ -1486,6 +1486,15 @@ static rd_bool_t rd_kafka_mock_cgrp_consumer_member_subscribed_topic_names_set(
         return changed;
 }
 
+static void rd_kafka_mock_cgrp_consumer_member_topic_id_set(
+    rd_kafka_mock_cgrp_consumer_member_t *member,
+    const rd_kafkap_str_t *MemberId) {
+        /* KIP 1082: MemberId is generated by the client */
+        rd_assert(RD_KAFKAP_STR_LEN(MemberId) > 0);
+        RD_IF_FREE(member->id, rd_free);
+        member->id = RD_KAFKAP_STR_DUP(MemberId);
+}
+
 /**
  * @brief Adds a member to consumer group \p mcgrp. If member with same
  *        \p MemberId is already present, only updates the connection and
@@ -1521,17 +1530,24 @@ rd_kafka_mock_cgrp_consumer_member_t *rd_kafka_mock_cgrp_consumer_member_add(
         if (!member) {
                 member = rd_kafka_mock_cgrp_consumer_member_find_by_instance_id(
                     mcgrp, InstanceId);
-                if (member && RD_KAFKAP_STR_LEN(MemberId) > 0 &&
-                    rd_kafkap_str_cmp_str(MemberId, member->id) != 0) {
-                        /* Either member is a new instance and is rejoining
-                         * with same InstanceId, so MemberId is NULL,
-                         * or it's rejoining after unsubscribing,
-                         * then it must have the same MemberId as before,
-                         * as it lasts for member lifetime.
-                         * It both don't hold, we cannot add the member
-                         * to the group. */
-                        return NULL;
+
+                if (member) {
+                        if (!member->left_static_membership) {
+                                /* Old member still active,
+                                 * fence this one */
+                                return NULL;
+                        }
+
+                        if (rd_kafkap_str_cmp_str(MemberId, member->id) != 0) {
+                                /* Member is a new instance and is rejoining
+                                 * with a new MemberId. */
+                                rd_kafka_mock_cgrp_consumer_member_topic_id_set(
+                                    member, MemberId);
+                        }
+                        member->left_static_membership = rd_false;
                 }
+        } else {
+                member->left_static_membership = rd_false;
         }
 
         if (!member) {
@@ -1548,7 +1564,8 @@ rd_kafka_mock_cgrp_consumer_member_t *rd_kafka_mock_cgrp_consumer_member_add(
                 member        = rd_calloc(1, sizeof(*member));
                 member->mcgrp = mcgrp;
 
-                member->id = RD_KAFKAP_STR_DUP(MemberId);
+                rd_kafka_mock_cgrp_consumer_member_topic_id_set(member,
+                                                                MemberId);
 
                 if (!RD_KAFKAP_STR_IS_NULL(InstanceId))
                         member->instance_id = RD_KAFKAP_STR_DUP(InstanceId);
@@ -1617,28 +1634,39 @@ static void rd_kafka_mock_cgrp_consumer_member_destroy(
         rd_free(member);
 }
 
+static void rd_kafka_mock_cgrp_consumer_member_leave_static(
+    rd_kafka_mock_cgrp_consumer_member_t *member) {
+        member->left_static_membership = rd_true;
+        rd_kafka_mock_cgrp_consumer_member_returned_assignment_set(member,
+                                                                   NULL);
+}
+
 
 /**
  * @brief Called when a member must leave a consumer group.
  *
  * @param mcgrp Consumer group to leave.
  * @param member Member that leaves.
+ * @param leave_static If true, the member is leaving with static group
+ * membership.
  *
  * @locks mcluster->lock MUST be held.
  */
 void rd_kafka_mock_cgrp_consumer_member_leave(
     rd_kafka_mock_cgrp_consumer_t *mcgrp,
-    rd_kafka_mock_cgrp_consumer_member_t *member) {
+    rd_kafka_mock_cgrp_consumer_member_t *member,
+    rd_bool_t leave_static) {
         rd_bool_t is_static = member->instance_id != NULL;
 
         rd_kafka_dbg(mcgrp->cluster->rk, MOCK, "MOCK",
-                     "Member %s is leaving group %s, is static: %s", member->id,
-                     mcgrp->id, RD_STR_ToF(is_static));
-        if (!is_static)
+                     "Member %s is leaving group %s, is static: %s, "
+                     "static leave: %s",
+                     member->id, mcgrp->id, RD_STR_ToF(is_static),
+                     RD_STR_ToF(leave_static));
+        if (!is_static || !leave_static)
                 rd_kafka_mock_cgrp_consumer_member_destroy(mcgrp, member);
         else
-                rd_kafka_mock_cgrp_consumer_member_returned_assignment_set(
-                    member, NULL);
+                rd_kafka_mock_cgrp_consumer_member_leave_static(member);
 }
 
 /**

src/rdkafka_mock_handlers.c

@@ -2896,7 +2896,7 @@ rd_kafka_mock_handle_ConsumerGroupHeartbeat(rd_kafka_mock_connection_t *mconn,
                                 }
                         } else {
                                 rd_kafka_mock_cgrp_consumer_member_leave(
-                                    mcgrp, member);
+                                    mcgrp, member, MemberEpoch == -2);
                                 member = NULL;
                         }
                 } else {

src/rdkafka_mock_int.h

@@ -170,6 +170,9 @@ typedef struct rd_kafka_mock_cgrp_consumer_member_s {
         rd_list_t *subscribed_topic_names; /**< Subscribed topic names received
                                               in the heartbeat */
         char *subscribed_topic_regex;      /**< Subscribed regex */
+
+        rd_bool_t left_static_membership;        /**< Member left the group
+                                                  *   with static membership. */
         struct rd_kafka_mock_connection_s *conn; /**< Connection, may be NULL
                                                   *   if there is no ongoing
                                                   *   request. */
@@ -675,7 +678,8 @@ rd_kafka_mock_cgrp_consumer_get(rd_kafka_mock_cluster_t *mcluster,
 
 void rd_kafka_mock_cgrp_consumer_member_leave(
     rd_kafka_mock_cgrp_consumer_t *mcgrp,
-    rd_kafka_mock_cgrp_consumer_member_t *member);
+    rd_kafka_mock_cgrp_consumer_member_t *member,
+    rd_bool_t static_leave);
 
 void rd_kafka_mock_cgrp_consumer_member_fenced(
     rd_kafka_mock_cgrp_consumer_t *mcgrp,

tests/0102-static_group_rebalance.c

@@ -673,16 +673,30 @@ static rd_bool_t is_api_key(rd_kafka_mock_request_t *request, void *opaque) {
         return rd_kafka_mock_request_api_key(request) == api_key;
 }
 
+/**
+ * @enum do_test_static_membership_mock_variation_t
+ * @brief Variations of the static membership mock test.
+ */
+typedef enum do_test_static_membership_mock_variation_t {
+        /** Consumer 1 leaves with unsubscribe and rejoins the group */
+        DO_TEST_STATIC_MEMBERSHIP_MOCK_VARIATION_SAME_INSTANCE = 0,
+        /** Consumer 1 leaves with unsubscribe and a new consumer with same
+         *  group.instance.id joins the group */
+        DO_TEST_STATIC_MEMBERSHIP_MOCK_VARIATION_NEW_INSTANCE = 1,
+        DO_TEST_STATIC_MEMBERSHIP_MOCK_VARIATION__CNT
+} do_test_static_membership_mock_variation_t;
+
 /**
  * @brief Static group membership tests with the mock cluster.
  *        Checks that consumer returns the same assignment
  *        and generation id after re-joining.
  *
- *        variation 1) consumer 1 leaves and rejoins the group.
- *        variation 2) consumer 1 leaves and a new consumer with same
- *                     group.instance.id joins the group.
+ * @param variation Test variation to run.
+ *
+ * @sa `do_test_static_membership_mock_variation_t`
  */
-static void do_test_static_membership_mock(int variation) {
+static void do_test_static_membership_mock(
+    do_test_static_membership_mock_variation_t variation) {
         const char *bootstraps;
         rd_kafka_mock_cluster_t *mcluster;
         int32_t api_key   = RD_KAFKAP_ConsumerGroupHeartbeat;
@@ -693,8 +707,11 @@ static void do_test_static_membership_mock(int variation) {
         rd_kafka_topic_partition_list_t *prev_assignment1, *prev_assignment2,
             *next_assignment1, *next_assignment2;
 
-        SUB_TEST_QUICK("%s", variation == 1 ? "with unsubscribe"
-                                            : "with new instance");
+        SUB_TEST_QUICK(
+            "%s",
+            variation == DO_TEST_STATIC_MEMBERSHIP_MOCK_VARIATION_SAME_INSTANCE
+                ? "with same instance"
+                : "with new instance");
 
         mcluster = test_mock_cluster_new(3, &bootstraps);
         rd_kafka_mock_topic_create(mcluster, topic, 2, 3);
@@ -730,7 +747,8 @@ static void do_test_static_membership_mock(int variation) {
                                          &api_key);
         rd_kafka_mock_stop_request_tracking(mcluster);
 
-        if (variation == 2) {
+        if (variation ==
+            DO_TEST_STATIC_MEMBERSHIP_MOCK_VARIATION_NEW_INSTANCE) {
                 /* Don't destroy it immediately because the
                  * topic partition lists still hold a reference. */
                 consumer_1_to_destroy = consumer1;
@@ -800,11 +818,7 @@ int main_0102_static_group_rebalance(int argc, char **argv) {
 
 int main_0102_static_group_rebalance_mock(int argc, char **argv) {
         TEST_SKIP_MOCK_CLUSTER(0);
-
-        TEST_SKIP(
-            "Static membership mock fixes are present in another PR. Test it "
-            "with that PR.\n");
-        return 0;
+        int variation;
 
         if (test_consumer_group_protocol_classic()) {
                 TEST_SKIP(
@@ -813,7 +827,11 @@ int main_0102_static_group_rebalance_mock(int argc, char **argv) {
                 return 0;
         }
 
-        do_test_static_membership_mock(1);
-        do_test_static_membership_mock(2);
+        for (variation = DO_TEST_STATIC_MEMBERSHIP_MOCK_VARIATION_SAME_INSTANCE;
+             variation < DO_TEST_STATIC_MEMBERSHIP_MOCK_VARIATION__CNT;
+             variation++) {
+                do_test_static_membership_mock(variation);
+        }
+
         return 0;
 }