jeffkbkim commented on code in PR #14017:
URL: https://github.com/apache/kafka/pull/14017#discussion_r1272854832
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -5205,11 +5245,102 @@ public void
testReplaceStaticMemberInStableStateErrors() throws Exception {
GenericGroupMember revertedMember =
group.member(group.staticMemberId("group-instance-id"));
- assertEquals(oldMember.memberId(), revertedMember.memberId());
- assertEquals(oldMember.groupInstanceId(),
revertedMember.groupInstanceId());
- assertEquals(oldMember.rebalanceTimeoutMs(),
revertedMember.rebalanceTimeoutMs());
- assertEquals(oldMember.sessionTimeoutMs(),
revertedMember.sessionTimeoutMs());
- assertEquals(oldMember.supportedProtocols(),
revertedMember.supportedProtocols());
+ assertEquals(oldMemberId, revertedMember.memberId());
+ assertEquals(Optional.of("group-instance-id"),
revertedMember.groupInstanceId());
+ assertEquals(4000, revertedMember.rebalanceTimeoutMs());
+ assertEquals(3000, revertedMember.sessionTimeoutMs());
+ assertEquals(protocols, revertedMember.supportedProtocols());
+ assertEquals(1, group.size());
+ assertEquals(1, group.generationId());
+ assertTrue(group.isInState(STABLE));
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplaceStaticMemberInStableStateSucceeds(
+ boolean supportSkippingAssignment
+ ) throws Exception {
+ // If the append future succeeds, the soft state is updated with the
new member.
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+ GenericGroup group = context.createGenericGroup("group-id");
+
+ JoinGroupRequestProtocolCollection protocols = new
JoinGroupRequestProtocolCollection(0);
+
+ protocols.add(new JoinGroupRequestProtocol()
+ .setName("range")
+ .setMetadata(ConsumerProtocol.serializeSubscription(new
ConsumerPartitionAssignor.Subscription(
+ Collections.singletonList("foo"))).array())
+ );
+
+ JoinGroupRequestData request = new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withGroupInstanceId("group-instance-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withProtocolType("consumer")
+ .withProtocols(protocols)
+ .build();
+
+ JoinGroupResponseData response =
context.joinGenericGroupAndCompleteJoin(
+ request,
+ true,
+ supportSkippingAssignment
+ );
+
+ assertEquals(Errors.NONE.code(), response.errorCode());
+ assertEquals(1, group.size());
+ assertEquals(1, group.generationId());
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
+
+ String oldMemberId = response.memberId();
+ // Simulate successful sync group phase
+ group.transitionTo(STABLE);
+
+ // Static member rejoins with UNKNOWN_MEMBER_ID and the append
succeeds.
+ protocols.add(new JoinGroupRequestProtocol()
+ .setName("roundrobin")
+ .setMetadata(ConsumerProtocol.serializeSubscription(new
ConsumerPartitionAssignor.Subscription(
+ Collections.singletonList("bar"))).array()));
+
+ CompletableFuture<JoinGroupResponseData> responseFuture = new
CompletableFuture<>();
+ CoordinatorResult<Void, Record> result = context.sendGenericGroupJoin(
+ request
+ .setProtocols(protocols)
+ .setRebalanceTimeoutMs(7000)
+ .setSessionTimeoutMs(6000),
+ responseFuture,
+ true,
+ supportSkippingAssignment);
+
+ assertEquals(
+ Collections.singletonList(newGroupMetadataRecord(group,
MetadataVersion.latest())),
+ result.records()
+ );
+ assertFalse(responseFuture.isDone());
+
+ // Simulate a successful write to the log.
+ result.appendFuture().complete(null);
Review Comment:
i like it explicitly stated since it shows when we expect the append future
to be completed.
we only replay in `sleep()`. the append future can be completed from both
join group and sync group requests along with sleep(), and follower sync
requests do not generate records so it doesn't help simplify. also i think it
confuses the reader more on what is actually going under the hood
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
