lucasbru commented on code in PR #23873:
URL: https://github.com/apache/kafka/pull/23873#discussion_r4324839153
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -3633,10 +3637,17 @@ private StreamsGroupMember
getOrMaybeCreateStaticStreamsGroupMember(
throwIfInstanceIdIsUnreleased(previousStaticMemberOrNull,
group.groupId(), memberId, instanceId);
// Copy the previous member but with the new member id.
- StreamsGroupMember newMember = new
StreamsGroupMember.Builder(previousStaticMemberOrNull, memberId)
+ StreamsGroupMember.Builder newMemberBuilder = new
StreamsGroupMember.Builder(previousStaticMemberOrNull, memberId)
.setMemberEpoch(0)
- .setPreviousMemberEpoch(0)
- .build();
+ .setPreviousMemberEpoch(0);
+ if (processId != null &&
!processId.equals(previousStaticMemberOrNull.processId())) {
Review Comment:
Since this is a corner case, could we simplify this: when a static member
rejoins from a different process, don't copy its assigned tasks at all and only
keep the relabeled target assignment, so the reconciler grants them with its
usual checks? The new process has no local state for those tasks anyway, so we
don't lose much, and we wouldn't need a second copy of the one-process check
here. The same could then apply to the other join paths (e.g. rejoining with
the same member ID after leaving with -2), though I don't think Streams can hit
those today.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/StreamsGroupStaticMemberGroupMetadataManagerTest.java:
##########
@@ -649,6 +660,81 @@ public void
testStaticMemberRejoinWithSameProcessIdDoesNotBumpStreamsGroupEpoch(
assertEquals(groupEpoch, result.response().data().memberEpoch());
}
+ @Test
+ public void
testStaticMemberRejoinFromNewProcessDropsTaskHeldThereAsStandby() {
+ String groupId = "fooup";
+ int groupEpoch = DEFAULT_GROUP_EPOCH;
+
+ String instanceId1 = Uuid.randomUuid().toString();
+ String instanceId2 = Uuid.randomUuid().toString();
+ String oldMemberId1 = Uuid.randomUuid().toString();
+ String oldMemberId2 = Uuid.randomUuid().toString();
+ String rejoinMemberId1 = Uuid.randomUuid().toString();
+ String rejoinMemberId2 = Uuid.randomUuid().toString();
+
+ String oldProcessId = "old-process-id";
+ String newProcessId = "new-process-id";
+
+ String subtopologyId = "subtopology1";
+ String sourceTopicName = "foo";
+ String changelogTopicName = "foo-changelog";
+ StreamsGroupHeartbeatRequestData.Topology topology = new
StreamsGroupHeartbeatRequestData.Topology()
+ .setSubtopologies(List.of(new
StreamsGroupHeartbeatRequestData.Subtopology()
+ .setSubtopologyId(subtopologyId)
+ .setSourceTopics(List.of(sourceTopicName))
+ .setStateChangelogTopics(List.of(new
StreamsGroupHeartbeatRequestData.TopicInfo().setName(changelogTopicName)))));
+ CoordinatorMetadataImage metadataImage = new MetadataImageBuilder()
+ .addTopic(Uuid.randomUuid(), sourceTopicName, 1)
+ .addTopic(Uuid.randomUuid(), changelogTopicName, 1)
+ .buildCoordinatorMetadataImage();
+
+ // Two members of one process left with epoch -2. The first one held
the active task of the only, stateful,
+ // task. Both rejoin from a new process.
+ GroupMetadataManagerTestContext context = new
GroupMetadataManagerTestContext.Builder()
+ .withStreamsGroupTaskAssignors(List.of(new StickyTaskAssignor()))
+ .withMetadataImage(metadataImage)
+
.withConfig(GroupCoordinatorConfig.STREAMS_GROUP_NUM_STANDBY_REPLICAS_CONFIG, 1)
+ .withStreamsGroup(new StreamsGroupBuilder(groupId, groupEpoch)
+
.withMember(streamsGroupMemberBuilderWithDefaults(oldMemberId1, instanceId1)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setProcessId(oldProcessId)
+
.setAssignedTasks(mkTasksTupleWithCommonEpoch(TaskRole.ACTIVE, groupEpoch,
mkTasks(subtopologyId, 0)))
+ .build())
+
.withMember(streamsGroupMemberBuilderWithDefaults(oldMemberId2, instanceId2)
+ .setMemberEpoch(LEAVE_GROUP_STATIC_MEMBER_EPOCH)
+ .setPreviousMemberEpoch(groupEpoch)
+ .setProcessId(oldProcessId)
+ .build())
+ .withTargetAssignment(oldMemberId1,
mkTasksTuple(TaskRole.ACTIVE, mkTasks(subtopologyId, 0)))
+ .withTargetAssignment(oldMemberId2, TasksTuple.EMPTY)
+ .withTargetAssignmentEpoch(groupEpoch)
+ .withTopology(StreamsTopology.fromHeartbeatRequest(topology))
+ .withValidatedTopologyEpoch(0)
+ .withMetadataHash(computeGroupHash(Map.of(
+ sourceTopicName, computeTopicHash(sourceTopicName,
metadataImage),
+ changelogTopicName, computeTopicHash(changelogTopicName,
metadataImage))))
+ .withLastAssignmentConfigs(new AssignmentConfigsImpl(1,
List.of())))
+ .build();
+ StreamsGroup group =
context.groupMetadataManager.streamsGroup(groupId);
+
+ // The second member rejoins first and gets the standby task, since
the active task is still on the old process.
+ CoordinatorResult<StreamsGroupHeartbeatResult, CoordinatorRecord>
result = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, rejoinMemberId2, instanceId2,
newProcessId)
+ );
+ assertEquals(mkResponseTasks(subtopologyId, 0),
result.response().data().standbyTasks());
+
+ // The first member copies its old member's active task, which the
second member already holds on the new
+ // process as a standby task, so the copy drops it.
+ result = context.streamsGroupHeartbeat(
+ staticJoinHeartbeat(groupId, rejoinMemberId1, instanceId1,
newProcessId)
+ );
+ assertEquals(List.of(), result.response().data().activeTasks());
+ assertEquals(TasksTupleWithEpochs.EMPTY,
group.getMemberOrThrow(rejoinMemberId1).assignedTasks());
+ assertNull(group.currentActiveTaskProcessId(subtopologyId, 0));
Review Comment:
Could we continue the test until the task is running again? E.g. sleep past
the assignment interval and heartbeat both members. When I tried that, the
sticky assignor promotes the second member's standby to active and the first
member ends up empty. Right now the test would also pass if the task stayed
unassigned forever.
--
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]
