dajac commented on code in PR #23174:
URL: https://github.com/apache/kafka/pull/23174#discussion_r3802174021
##########
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:
I don't fully get the `value != null` condition. If the future is set to
null, I would think that we also want the previous one to be completed. Or, is
there a case where we don't want this? It is a bit confusing.
##########
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:
nit: We can use `List.of()`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -7706,6 +7706,15 @@ private CoordinatorResult<Void, CoordinatorRecord>
updateMemberThenRebalanceOrCo
String joinReason,
CompletableFuture<JoinGroupResponseData> responseFuture
) {
+ // A prior JoinGroup for this member may still be pending; complete it
before
+ // group.updateMember discards it, or it would never resolve on its
own.
+ if (member.isAwaitingJoin()) {
+ group.completeJoinFuture(member, new JoinGroupResponseData()
+ .setMemberId(member.memberId())
+ .setErrorCode(Errors.REBALANCE_IN_PROGRESS.code())
Review Comment:
It works when the group is in `PREPARING_REBALANCE` state but it may not be
all the time. Returning `NOT_COORDINATOR` seems to be more general. The
response is doomed to failed anyway. What do you think?
--
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]
