[KIP-848] Fix static group membership

emasab · emasab · commit eda70686eb14 · 2025-03-14T12:11:25.000+01:00
there are three cases:
- member explicitly unsubscribes -&gt; should cause a rebalance
- member closes the consumer -&gt; no rebalance
- member destroys the consumer -&gt; no rebalance

When a new member joins the group and there's already a member with same group instance id that hasn't left
statically the new member is fenced with the new protocol
diff --git a/src/rdkafka_cgrp.c b/src/rdkafka_cgrp.c
@@ -981,7 +981,8 @@ rd_kafka_cgrp_handle_ConsumerGroupHeartbeat_leave(rd_kafka_t *rk,
         goto err;
 }
 
-static void rd_kafka_cgrp_consumer_leave(rd_kafka_cgrp_t *rkcg) {
+static void rd_kafka_cgrp_consumer_leave(rd_kafka_cgrp_t *rkcg,
+                                         rd_bool_t static_leave) {
         int32_t member_epoch = -1;
 
         if (rkcg->rkcg_flags & RD_KAFKA_CGRP_F_WAIT_LEAVE) {
@@ -999,7 +1000,7 @@ static void rd_kafka_cgrp_consumer_leave(rd_kafka_cgrp_t *rkcg) {
                      rd_kafka_cgrp_state_names[rkcg->rkcg_state]);
 
         rkcg->rkcg_flags |= RD_KAFKA_CGRP_F_WAIT_LEAVE;
-        if (RD_KAFKA_CGRP_IS_STATIC_MEMBER(rkcg)) {
+        if (static_leave && RD_KAFKA_CGRP_IS_STATIC_MEMBER(rkcg)) {
                 member_epoch = -2;
         }
 
@@ -1078,7 +1079,8 @@ static rd_bool_t rd_kafka_cgrp_leave_maybe(rd_kafka_cgrp_t *rkcg) {
                 return rd_false;
 
         if (rkcg->rkcg_group_protocol == RD_KAFKA_GROUP_PROTOCOL_CONSUMER) {
-                rd_kafka_cgrp_consumer_leave(rkcg);
+                rd_kafka_cgrp_consumer_leave(
+                    rkcg, rkcg->rkcg_flags & RD_KAFKA_CGRP_F_TERMINATE);
         } else {
                 /* KIP-345: Static group members must not send a
                  * LeaveGroupRequest on termination. */
@@ -1386,7 +1388,6 @@ static void rd_kafka_cgrp_rejoin(rd_kafka_cgrp_t *rkcg, const char *fmt, ...) {
                 rd_kafka_cgrp_leave_maybe(rkcg);
         }
 
-        rd_kafka_cgrp_consumer_reset(rkcg);
         rd_kafka_cgrp_set_join_state(rkcg, RD_KAFKA_CGRP_JOIN_STATE_INIT);
         rd_kafka_cgrp_consumer_expedite_next_heartbeat(rkcg, "rejoining");
 }
@@ -3096,6 +3097,7 @@ void rd_kafka_cgrp_handle_ConsumerGroupHeartbeat(rd_kafka_t *rk,
                              "ConsumerGroupHeartbeat failed due to: %s: "
                              "will rejoin the group",
                              rd_kafka_err2str(err));
+                rd_kafka_cgrp_consumer_reset(rkcg);
                 rkcg->rkcg_consumer_flags |=
                     RD_KAFKA_CGRP_CONSUMER_F_WAIT_REJOIN;
                 return;
@@ -4963,7 +4965,7 @@ rd_kafka_cgrp_max_poll_interval_check_tmr_cb(rd_kafka_timers_t *rkts,
                             1 /*lock*/);
 
         if (rkcg->rkcg_group_protocol == RD_KAFKA_GROUP_PROTOCOL_CONSUMER) {
-                rd_kafka_cgrp_consumer_leave(rkcg);
+                rd_kafka_cgrp_consumer_leave(rkcg, rd_false);
                 rkcg->rkcg_consumer_flags |=
                     RD_KAFKA_CGRP_CONSUMER_F_WAIT_REJOIN;
                 rd_kafka_cgrp_consumer_expedite_next_heartbeat(
diff --git a/src/rdkafka_mock.c b/src/rdkafka_mock.c
@@ -3036,7 +3036,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);
diff --git a/src/rdkafka_mock_cgrp.c b/src/rdkafka_mock_cgrp.c
@@ -1462,18 +1462,26 @@ rd_kafka_mock_cgrp_consumer_member_add(rd_kafka_mock_cgrp_consumer_t *mcgrp,
         /* Find member */
         member = rd_kafka_mock_cgrp_consumer_member_find(mcgrp, MemberId);
         if (!member) {
+                if (RD_KAFKAP_STR_LEN(MemberId) == 0)
+                        /* KIP 1082: MemberId is generated by the client */
+                        return NULL;
+
                 member = rd_kafka_mock_cgrp_consumer_member_find_by_instance_id(
                     mcgrp, InstanceId);
-                if (member && RD_KAFKAP_STR_LEN(MemberId) > 0 &&
+
+                if (member &&
                     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;
+                        /* Member is a new instance and is rejoining
+                         * with a new MemberId. */
+
+                        if (!member->left_static_membership)
+                                /* Old member still active,
+                                 * fence this one */
+                                return NULL;
+
+                        RD_IF_FREE(member->id, rd_free);
+                        member->id = RD_KAFKAP_STR_DUP(MemberId);
+                        member->left_static_membership = rd_false;
                 }
         }
 
@@ -1555,29 +1563,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 is_static If true, the member is leaving with static group membership.
+ * @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);
 }
 
 /**
diff --git a/src/rdkafka_mock_handlers.c b/src/rdkafka_mock_handlers.c
@@ -2884,7 +2884,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 {
diff --git a/src/rdkafka_mock_int.h b/src/rdkafka_mock_int.h
@@ -166,6 +166,9 @@ typedef struct rd_kafka_mock_cgrp_consumer_member_s {
 
         rd_list_t *subscribed_topics; /**< Subscribed topics */
 
+        rd_bool_t left_static_membership; /**< Member has left the group
+                                           *   with static membership. */
+
         struct rd_kafka_mock_connection_s *conn; /**< Connection, may be NULL
                                                   *   if there is no ongoing
                                                   *   request. */
@@ -671,7 +674,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,
diff --git a/tests/0102-static_group_rebalance.c b/tests/0102-static_group_rebalance.c

Original file line number	Diff line number	Diff line change
`@@ -2884,7 +2884,7 @@ rd_kafka_mock_handle_ConsumerGroupHeartbeat(rd_kafka_mock_connection_t *mconn,`
`2884`	`2884`	`}`
`2885`	`2885`	`} else {`
`2886`	`2886`	`rd_kafka_mock_cgrp_consumer_member_leave(`
`2887`		`- mcgrp, member);`
	`2887`	`+ mcgrp, member, MemberEpoch == -2);`
`2888`	`2888`	`member = NULL;`
`2889`	`2889`	`}`
`2890`	`2890`	`} else {`