lucasbru commented on code in PR #20730:
URL: https://github.com/apache/kafka/pull/20730#discussion_r2447061734
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/CurrentAssignmentBuilder.java:
##########
@@ -368,31 +374,103 @@ private StreamsGroupMember computeNextAssignment(int
memberEpoch,
.contains(member.processId())
);
- return buildNewMember(
- memberEpoch,
- new TasksTuple(
+ // Add epochs to the computed task tuples
+ // Preserve previous epochs for tasks that were already assigned or
pending revocation,
+ // and use the target assignment epoch for newly assigned tasks.
+ TasksTupleWithEpochs newTasksPendingRevocationWithEpochs = new
TasksTupleWithEpochs(
+ addEpochsToTasks(
newActiveTasksPendingRevocation,
- newStandbyTasksPendingRevocation,
- newWarmupTasksPendingRevocation
+ memberAssignedTasks,
+ memberTasksPendingRevocation,
+ targetAssignmentEpoch
),
- new TasksTuple(
+ newStandbyTasksPendingRevocation,
+ newWarmupTasksPendingRevocation
+ );
+ TasksTupleWithEpochs newAssignedTasksWithEpochs = new
TasksTupleWithEpochs(
+ addEpochsToTasks(
newActiveAssignedTasks,
- newStandbyAssignedTasks,
- newWarmupAssignedTasks
+ memberAssignedTasks,
+ memberTasksPendingRevocation,
+ targetAssignmentEpoch
),
- new TasksTuple(
+ newStandbyAssignedTasks,
+ newWarmupAssignedTasks
+ );
+ TasksTupleWithEpochs newTasksPendingAssignmentWithEpochs = new
TasksTupleWithEpochs(
+ addEpochsToTasks(
newActiveTasksPendingAssignment,
- newStandbyTasksPendingAssignment,
- newWarmupTasksPendingAssignment
+ memberAssignedTasks,
+ memberTasksPendingRevocation,
+ targetAssignmentEpoch
),
+ newStandbyTasksPendingAssignment,
+ newWarmupTasksPendingAssignment
+ );
+
+ return buildNewMember(
Review Comment:
Members are immutable, so we indeed build a new object that represents the
same member.
I think I'm going to keep it that way since it's not completely unreasonable
and consistent with consumer group naming.
--
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]
