varunv-cflt commented on code in PR #23174:
URL: https://github.com/apache/kafka/pull/23174#discussion_r3854853171
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroupMember.java:
##########
@@ -397,16 +398,35 @@ public void setAssignment(byte[] value) {
}
/**
+ * Set the member's join future. If a join future is already pending when
a new,
+ * non-null one is set, the new request supersedes it, so the earlier one
is
+ * completed with REBALANCE_IN_PROGRESS first -- otherwise it would never
resolve
+ * on its own.
+ *
* @param value the updated join future.
*/
public void setAwaitingJoinFuture(CompletableFuture<JoinGroupResponseData>
value) {
+ if (value != null && awaitingJoinFuture != null) {
Review Comment:
@dajac - When join has to rollback, the value will be set to null and the
future will be completed later explicitly with error code. The `value != null`
condition is to primarily handle that case so that we do not respond with an
incorrect error. Updated a comment around the code to reflect the same
@squah-confluent - Added completeJoinFuture and completeSyncFuture helpers
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -9848,6 +9907,82 @@ public void testSyncGroupLeaderAfterFollower() throws
Exception {
assertTrue(group.isInState(STABLE));
}
+ @Test
+ public void
testSyncGroupDuplicateFollowerSyncDuringCompletingRebalanceCompletesPreviousSyncFuture()
throws Exception {
+ // Mirrors
testJoinGroupExistingMemberRejoinDuringPreparingRebalanceCompletesPreviousJoinFuture,
+ // but for SyncGroup: a member's stale, still-pending sync future must
be completed with a
+ // retriable error rather than silently orphaned when superseded.
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData leaderJoinResponse =
context.joinClassicGroupAsDynamicMemberAndCompleteRebalance("group-id");
+ ClassicGroup group =
context.groupMetadataManager.getOrMaybeCreateClassicGroup("group-id", false);
+
+ JoinGroupRequestData joinRequest = new
GroupMetadataManagerTestContext.JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .withRebalanceTimeoutMs(10000)
+ .withSessionTimeoutMs(5000)
+ .build();
+
+ GroupMetadataManagerTestContext.JoinResult followerJoinResult =
context.sendClassicGroupJoin(joinRequest);
+ assertFalse(followerJoinResult.joinFuture.isDone());
+
+ GroupMetadataManagerTestContext.JoinResult leaderJoinResult =
context.sendClassicGroupJoin(
+ joinRequest.setMemberId(leaderJoinResponse.memberId())
+ );
+ assertTrue(leaderJoinResult.joinFuture.isDone());
+ assertTrue(followerJoinResult.joinFuture.isDone());
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
+
+ int nextGenerationId =
leaderJoinResult.joinFuture.get().generationId();
+ String followerId = followerJoinResult.joinFuture.get().memberId();
+
+ SyncGroupRequestData syncRequest = new
GroupMetadataManagerTestContext.SyncGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(followerId)
+ .withGenerationId(nextGenerationId)
+ .build();
+
+ // Follower syncs once, pending on the leader.
+ GroupMetadataManagerTestContext.SyncResult firstSync =
context.sendClassicGroupSync(syncRequest);
+ assertTrue(firstSync.records.isEmpty());
+ assertFalse(firstSync.syncFuture.isDone());
+
+ // Same follower retries SyncGroup before the first sync has completed.
+ GroupMetadataManagerTestContext.SyncResult secondSync =
context.sendClassicGroupSync(syncRequest);
+ assertTrue(secondSync.records.isEmpty());
+
+ // Superseded first sync must be completed promptly with a retriable
error.
+ assertTrue(firstSync.syncFuture.isDone());
+ assertEquals(Errors.REBALANCE_IN_PROGRESS.code(),
firstSync.syncFuture.get().errorCode());
+ assertFalse(secondSync.syncFuture.isDone());
+
+ // Leader syncs, completing the rebalance.
+ List<SyncGroupRequestAssignment> assignment = new ArrayList<>();
Review Comment:
Done
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
