mjsax commented on code in PR #20730:
URL: https://github.com/apache/kafka/pull/20730#discussion_r2446665332
##########
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,
Review Comment:
Question about existing code: what does `newActiveTasksPendingRevocation`
mean? I find this name confusing. Either I have a newly assigned task, or I
have an existing task that get revoked. How can a new task get revoked?
##########
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
Review Comment:
Otherwise I am wondering "what is the default" and need to check the caller
code.
##########
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:
What does `buildNewMember` mean? I though we are updating the assignment of
an existing member?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -1471,6 +1472,30 @@ private static boolean
areOwnedTasksContainedInAssignedTasks(
return true;
}
+ /**
+ * Checks whether all the tasks contained in the list are included in the
provided assignment with epochs.
+ *
+ * @param ownedTasks The tasks provided by the streams group
member in the request.
+ * @param assignedTasksWithEpochs The tasks that the member should have
(with epochs).
+ * @return A boolean indicating whether the owned partitions are a subset
or not.
+ */
+ private static boolean areOwnedTasksContainedInAssignedTasksWithEpochs(
+ List<StreamsGroupHeartbeatRequestData.TaskIds> ownedTasks,
+ Map<String, Map<Integer, Integer>> assignedTasksWithEpochs
+ ) {
+ if (ownedTasks == null) return false;
Review Comment:
Why is this false? If `ownedTask` is null or empty, it seems everything from
`ownedTask` is contained in `assignedTasksWithEpochs` ?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/streams/TasksTupleWithEpochs.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.streams;
+
+import
org.apache.kafka.coordinator.group.generated.StreamsGroupCurrentMemberAssignmentValue;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * An immutable tuple containing active, standby and warm-up tasks with
assignment epochs.
+ * <p>
+ * Active tasks include epoch information to support fencing of zombie commits.
+ * Standby and warmup tasks do not have epochs as they don't commit offsets.
+ *
+ * @param activeTasksWithEpochs Active tasks with their assignment epochs.
+ * The outer map key is the subtopology ID, the
inner map key is the partition ID,
+ * and the inner map value is the assignment
epoch.
+ * @param standbyTasks Standby tasks.
+ * The key of the map is the subtopology ID, and
the value is the set of partition IDs.
+ * @param warmupTasks Warm-up tasks.
+ * The key of the map is the subtopology ID, and
the value is the set of partition IDs.
+ */
+public record TasksTupleWithEpochs(Map<String, Map<Integer, Integer>>
activeTasksWithEpochs,
+ Map<String, Set<Integer>> standbyTasks,
+ Map<String, Set<Integer>> warmupTasks) {
+
+ public TasksTupleWithEpochs {
+ activeTasksWithEpochs =
Collections.unmodifiableMap(Objects.requireNonNull(activeTasksWithEpochs));
Review Comment:
`unmodifiableMap` is just a wrapper around the provided map, so it only
guards agains modification from inside this class, but if
`activeTasksWithEpochs` is modified from "outside" it could still be altered.
Should we make a deep copy to protect agains modifications? If there
actually a risk that we would modify the map internally and need to protect it?
##########
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
Review Comment:
```suggestion
int targetAssignmentEpoch
```
##########
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:
How could `partitionsWithEpochs` be empty?
--
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]
