lucasbru commented on code in PR #20730:
URL: https://github.com/apache/kafka/pull/20730#discussion_r2447288914
##########
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(
+ memberEpoch,
+ newTasksPendingRevocationWithEpochs,
+ newAssignedTasksWithEpochs,
+ newTasksPendingAssignmentWithEpochs,
hasUnreleasedActiveTasks || hasUnreleasedStandbyTasks ||
hasUnreleasedWarmupTasks
);
}
+ /**
+ * Helper method to add epochs to active tasks. This method looks up
epochs from existing assignments
+ * (memberAssignedTasks or memberTasksPendingRevocation) or uses the
provided default epoch
+ * for newly assigned tasks.
+ *
+ * @param activeTasks The active tasks without epochs.
+ * @param memberAssignedTasks The member's currently assigned
tasks with epochs.
+ * @param memberTasksPendingRevocation The member's tasks pending
revocation with epochs.
+ * @param defaultEpoch The default epoch to use for tasks
not found in existing assignments.
+ * @return Active tasks with epochs attached.
+ */
+ private Map<String, Map<Integer, Integer>> addEpochsToTasks(
+ Map<String, Set<Integer>> activeTasks,
+ TasksTupleWithEpochs memberAssignedTasks,
+ TasksTupleWithEpochs memberTasksPendingRevocation,
+ int defaultEpoch
+ ) {
+ Map<String, Map<Integer, Integer>> activeTasksWithEpochs = new
HashMap<>();
+
+ // For each active task, try to find its epoch from existing
assignments
+ activeTasks.forEach((subtopologyId, partitions) -> {
+ Map<Integer, Integer> partitionsWithEpochs = new HashMap<>();
+ for (Integer partition : partitions) {
+ // First check in assigned tasks
+ Integer epoch = memberAssignedTasks.activeTasksWithEpochs()
+ .getOrDefault(subtopologyId, Map.of())
+ .get(partition);
+
+ // If not found, check in tasks pending revocation
+ if (epoch == null) {
+ epoch =
memberTasksPendingRevocation.activeTasksWithEpochs()
+ .getOrDefault(subtopologyId, Map.of())
+ .get(partition);
+ }
+
+ // If still not found, use the default epoch
+ if (epoch == null) {
+ epoch = defaultEpoch;
+ }
+
+ partitionsWithEpochs.put(partition, epoch);
+ }
+ if (!partitionsWithEpochs.isEmpty()) {
Review Comment:
Good point. It cannot.
--
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]
