This is an automated email from the ASF dual-hosted git repository.
dajac pushed a commit to branch 4.3
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.3 by this push:
new fec5c19c37a KAFKA-20601: Allow static member rejoin when consumer
group is full. (#22356)
fec5c19c37a is described below
commit fec5c19c37a33a7b18624c12665a55cafba2a01d
Author: ChickenchickenLove <[email protected]>
AuthorDate: Tue Jun 23 16:05:59 2026 +0900
KAFKA-20601: Allow static member rejoin when consumer group is full.
(#22356)
Previously, a static member rejoining a full modern consumer group could
be rejected by the max group size check before the coordinator resolved
that it was replacing an existing static member.
Now, the max size check allows existing static members by `instanceId`
to proceed to the static member validation/replacement path. New members
are still rejected when the consumer group is at capacity, and
unreleased static instance IDs still fail with the appropriate
exception.
Reviewers: nileshkumar3 <[email protected]>, David Jacot
<[email protected]>
---
.../coordinator/group/GroupMetadataManager.java | 20 +-
.../group/GroupMetadataManagerTest.java | 239 +++++++++++++++++++++
2 files changed, 254 insertions(+), 5 deletions(-)
diff --git
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
index e68981ca6d7..e91a739fca3 100644
---
a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
+++
b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java
@@ -1561,15 +1561,25 @@ public class GroupMetadataManager {
* Checks whether the consumer group can accept a new member or not based
on the
* max group size defined.
*
- * @param group The consumer group.
- * @param memberId The member id.
+ * @param group The consumer group.
+ * @param memberId The member id.
+ * @param instanceId The instance id.
*
* @throws GroupMaxSizeReachedException if the maximum capacity has been
reached.
*/
private void throwIfConsumerGroupIsFull(
ConsumerGroup group,
- String memberId
+ String memberId,
+ String instanceId
) throws GroupMaxSizeReachedException {
+ // If a static member already exists, we do not enforce the maximum
group size check.
+ // An existing static member will fall into one of the following two
cases,
+ // and neither affects the group size:
+ // 1. The member is replaced due to the static member rejoining.
+ // 2. 'UnreleasedInstanceIdException' is raised due to an epoch
mismatch.
+ if (group.hasStaticMember(instanceId))
+ return;
+
// If the consumer group has reached its maximum capacity, the member
is rejected if it is not
// already a member of the consumer group.
if (group.numMembers() >= config.consumerGroupMaxSize() &&
(memberId.isEmpty() || !group.hasMember(memberId))) {
@@ -2336,7 +2346,7 @@ public class GroupMetadataManager {
// Get or create the consumer group.
boolean createIfNotExists = memberEpoch == 0;
final ConsumerGroup group = getOrMaybeCreateConsumerGroup(groupId,
createIfNotExists, records);
- throwIfConsumerGroupIsFull(group, memberId);
+ throwIfConsumerGroupIsFull(group, memberId, instanceId);
// Get or create the member.
if (memberId.isEmpty()) memberId = Uuid.randomUuid().toString();
@@ -2502,7 +2512,7 @@ public class GroupMetadataManager {
final boolean isUnknownMember = memberId.equals(UNKNOWN_MEMBER_ID);
if (isUnknownMember) memberId = Uuid.randomUuid().toString();
- throwIfConsumerGroupIsFull(group, memberId);
+ throwIfConsumerGroupIsFull(group, memberId, instanceId);
throwIfClassicProtocolIsNotSupported(group, memberId,
request.protocolType(), protocols);
if (JoinGroupRequest.requiresKnownMemberId(request,
context.requestVersion())) {
diff --git
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
index 302d20db241..99e22bbeac5 100644
---
a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
+++
b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java
@@ -3722,6 +3722,177 @@ public class GroupMetadataManagerTest {
.setTopicPartitions(List.of())));
}
+ @Test
+ public void testStaticMemberCanRejoinWhenConsumerGroupIsFull() {
+ String groupId = "fooup";
+ String instanceId = "instance-id";
+ String oldMemberId = "old-member-id";
+ String newMemberId = "new-member-id";
+ int groupMaxSize = 1;
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 1)
+ .buildCoordinatorMetadataImage();
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ newMemberId, new
MemberAssignmentImpl(mkAssignment(mkTopicAssignment(fooTopicId, 0)))
+ )));
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG,
groupMaxSize)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(new ConsumerGroupMember.Builder(oldMemberId)
+ .setState(MemberState.STABLE)
+ .setInstanceId(instanceId)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(9)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of("foo", "bar"))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)), 10))
+ .build())
+ .withAssignment(oldMemberId,
mkAssignment(mkTopicAssignment(fooTopicId, 0)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(computeGroupHash(Map.of(fooTopicName,
computeTopicHash(fooTopicName, metadataImage)))))
+ .build();
+
+ CoordinatorResult<ConsumerGroupHeartbeatResponseData,
CoordinatorRecord> result = context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setInstanceId(instanceId)
+ .setMemberId(newMemberId)
+ .setMemberEpoch(0)
+ .setServerAssignor("range")
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of("foo"))
+ .setTopicPartitions(List.of()));
+
+ assertResponseEquals(
+ new ConsumerGroupHeartbeatResponseData()
+ .setMemberId(newMemberId)
+ .setMemberEpoch(11)
+ .setHeartbeatIntervalMs(5000)
+ .setAssignment(new
ConsumerGroupHeartbeatResponseData.Assignment()
+ .setTopicPartitions(List.of(
+ new
ConsumerGroupHeartbeatResponseData.TopicPartitions()
+ .setTopicId(fooTopicId)
+ .setPartitions(List.of(0))
+ ))),
+ result.response()
+ );
+ }
+
+ @Test
+ public void
testStaticMemberRejoinWithUnreleasedInstanceIdFailsWhenConsumerGroupIsFull() {
+ String groupId = "fooup";
+ String instanceId = "instance-id";
+ String oldMemberId = "old-member-id";
+ String newMemberId = "new-member-id";
+ int groupMaxSize = 1;
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 1)
+ .buildCoordinatorMetadataImage();
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ oldMemberId, new
MemberAssignmentImpl(mkAssignment(mkTopicAssignment(fooTopicId, 0)))
+ )));
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG,
groupMaxSize)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(new ConsumerGroupMember.Builder(oldMemberId)
+ .setState(MemberState.STABLE)
+ .setInstanceId(instanceId)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of("foo", "bar"))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)), 10))
+ .build())
+ .withAssignment(oldMemberId,
mkAssignment(mkTopicAssignment(fooTopicId, 0)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(computeGroupHash(Map.of(fooTopicName,
computeTopicHash(fooTopicName, metadataImage)))))
+ .build();
+
+ assertThrows(UnreleasedInstanceIdException.class, () ->
context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setInstanceId(instanceId)
+ .setMemberId(newMemberId)
+ .setMemberEpoch(0)
+ .setServerAssignor("range")
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of("foo"))
+ .setTopicPartitions(List.of())));
+ }
+
+ @Test
+ public void testNewStaticMemberIsRejectedWhenConsumerGroupIsFull() {
+ String groupId = "fooup";
+ String instanceId = "instance-id";
+ String otherInstanceId = "other-instance-id";
+ String oldMemberId = "old-member-id";
+ String newMemberId = "new-member-id";
+ int groupMaxSize = 1;
+
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(fooTopicId, fooTopicName, 1)
+ .buildCoordinatorMetadataImage();
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ assignor.prepareGroupAssignment(new GroupAssignment(Map.of(
+ oldMemberId, new
MemberAssignmentImpl(mkAssignment(mkTopicAssignment(fooTopicId, 0)))
+ )));
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_ASSIGNORS_CONFIG,
List.of(assignor))
+ .withMetadataImage(metadataImage)
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG,
groupMaxSize)
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(new ConsumerGroupMember.Builder(oldMemberId)
+ .setState(MemberState.STABLE)
+ .setInstanceId(instanceId)
+ .setMemberEpoch(10)
+ .setPreviousMemberEpoch(9)
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of("foo", "bar"))
+ .setServerAssignorName("range")
+ .setAssignedPartitions(toAssignmentWithEpochs(mkAssignment(
+ mkTopicAssignment(fooTopicId, 0)), 10))
+ .build())
+ .withAssignment(oldMemberId,
mkAssignment(mkTopicAssignment(fooTopicId, 0)))
+ .withAssignmentEpoch(10)
+ .withMetadataHash(computeGroupHash(Map.of(fooTopicName,
computeTopicHash(fooTopicName, metadataImage)))))
+ .build();
+
+ assertThrows(GroupMaxSizeReachedException.class, () ->
context.consumerGroupHeartbeat(
+ new ConsumerGroupHeartbeatRequestData()
+ .setGroupId(groupId)
+ .setInstanceId(otherInstanceId)
+ .setMemberId(newMemberId)
+ .setMemberEpoch(0)
+ .setServerAssignor("range")
+ .setRebalanceTimeoutMs(5000)
+ .setSubscribedTopicNames(List.of("foo"))
+ .setTopicPartitions(List.of())));
+ }
+
@Test
public void testConsumerGroupStates() {
String groupId = "fooup";
@@ -13830,6 +14001,74 @@ public class GroupMetadataManagerTest {
assertEquals("The consumer group has reached its maximum capacity of 1
members.", ex.getMessage());
}
+ @Test
+ public void
testStaticMemberCanRejoinConsumerGroupWithClassicProtocolWhenGroupIsFull()
throws Exception {
+ String groupId = "group-id";
+ String oldMemberId = "old-member";
+ String instanceId = "instance-id";
+
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(new ConsumerGroupMember.Builder(oldMemberId)
+ .setInstanceId(instanceId)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(9)
+ .build())
+ .withAssignmentEpoch(10))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG,
1)
+
.withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MIGRATION_POLICY_CONFIG,
ConsumerGroupMigrationPolicy.UPGRADE.toString())
+ .build();
+
+ JoinGroupRequestData request = new
GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
+ .withGroupId(groupId)
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withGroupInstanceId(instanceId)
+
.withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(List.of(),
List.of()))
+ .build();
+
+ GroupMetadataManagerTestContext.JoinResult joinResult =
context.sendClassicGroupJoin(request, true, true);
+ joinResult.appendFuture.complete(null);
+ assertTrue(joinResult.joinFuture.isDone());
+
+ JoinGroupResponseData response = joinResult.joinFuture.get();
+ assertEquals(Errors.NONE.code(), response.errorCode());
+ assertNotEquals(UNKNOWN_MEMBER_ID, response.memberId());
+ assertNotEquals(oldMemberId, response.memberId());
+ assertEquals(response.memberId(),
context.groupMetadataManager.consumerGroup(groupId).staticMemberId(instanceId));
+ }
+
+ @Test
+ public void
testNewStaticMemberClassicGroupJoinThrowsGroupMaxSizeReachedExceptionWhenConsumerGroupIsFull()
throws Exception {
+ String groupId = "group-id";
+ String oldMemberId = "old-member";
+ String instanceId = "instance-id";
+ String newInstanceId = "new-instance-id";
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withConsumerGroup(new ConsumerGroupBuilder(groupId, 10)
+ .withMember(new ConsumerGroupMember.Builder(oldMemberId)
+ .setInstanceId(instanceId)
+ .setState(MemberState.STABLE)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(9)
+ .build())
+ .withAssignmentEpoch(10))
+ .withConfig(GroupCoordinatorConfig.CONSUMER_GROUP_MAX_SIZE_CONFIG,
1)
+ .build();
+
+ JoinGroupRequestData request = new
GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
+ .withGroupId(groupId)
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withGroupInstanceId(newInstanceId)
+
.withProtocols(GroupMetadataManagerTestContext.toConsumerProtocol(List.of(),
List.of()))
+ .build();
+
+ assertThrows(GroupMaxSizeReachedException.class,
+ () -> context.sendClassicGroupJoin(request, true, true)
+ );
+ }
+
+
@Test
public void
testJoiningConsumerGroupThrowsExceptionIfProtocolIsNotSupported() {
String groupId = "group-id";
