You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/07/09 00:04:58 UTC
[GitHub] [kafka] guozhangwang commented on a diff in pull request #12397: KAFKA-10199: Cleanup TaskManager and Task interfaces
guozhangwang commented on code in PR #12397:
URL: https://github.com/apache/kafka/pull/12397#discussion_r917191699
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -34,29 +33,40 @@
import java.util.TreeMap;
import java.util.stream.Collectors;
+import static org.apache.kafka.common.utils.Utils.filterMap;
+import static org.apache.kafka.common.utils.Utils.union;
+
+/**
+ * All tasks contained by the Streams instance.
+ *
+ * Note that these tasks are shared between the TaskManager (stream thread) and the StateUpdater (restore thread),
+ * i.e. all running active tasks are processed by the former and all restoring active tasks and standby tasks are
+ * processed by the latter.
+ */
class Tasks {
private final Logger log;
- private final TopologyMetadata topologyMetadata;
-
- private final Map<TaskId, Task> allTasksPerId = Collections.synchronizedSortedMap(new TreeMap<>());
- private final Map<TaskId, Task> readOnlyTasksPerId = Collections.unmodifiableMap(allTasksPerId);
- private final Collection<Task> readOnlyTasks = Collections.unmodifiableCollection(allTasksPerId.values());
// TODO: change type to `StreamTask`
private final Map<TaskId, Task> activeTasksPerId = new TreeMap<>();
+ // TODO: change type to `StandbyTask`
+ private final Map<TaskId, Task> standbyTasksPerId = new TreeMap<>();
+
+ // Tasks may have been assigned for a NamedTopology that is not yet known by this host. When that occurs we stash
+ // these unknown tasks until either the corresponding NamedTopology is added and we can create them at last, or
+ // we receive a new assignment and they are revoked from the thread.
+
+ // Tasks may have been assigned but not yet created because:
+ // 1. They are for a NamedTopology that is yet known by this host.
+ // 2. They are to be recycled from an existing restoring task yet to be returned from the task updater.
+ //
+ // When that occurs we stash these pending tasks until either they are finally clear to be created,
+ // or they are revoked from a new assignment.
+ private final Map<TaskId, Set<TopicPartition>> pendingActiveTasks = new HashMap<>();
Review Comment:
This is mainly for case 1) in the description.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -65,13 +75,10 @@ class Tasks {
private Consumer<byte[], byte[]> mainConsumer;
Tasks(final LogContext logContext,
- final TopologyMetadata topologyMetadata,
Review Comment:
This is for case 2) in description: we detect unknown named topologies in the caller (`TaskManager`) so `Tasks` would be just dumb book-keeping.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTaskCreator.java:
##########
@@ -73,30 +67,14 @@ class StandbyTaskCreator {
);
}
- void removeRevokedUnknownTasks(final Set<TaskId> assignedTasks) {
Review Comment:
This is for case 2) in description.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java:
##########
@@ -151,7 +156,8 @@ final void transitionTo(final Task.State newState) {
@Override
public void updateInputPartitions(final Set<TopicPartition> topicPartitions, final Map<String, List<String>> allTopologyNodesToSourceTopics) {
- this.inputPartitions = topicPartitions;
+ this.inputPartitions.clear();
Review Comment:
Minor cleanup: do deep copy since the passed in `topicPartition` param maybe modified.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java:
##########
@@ -83,7 +83,6 @@ class DefaultStateUpdaterTest {
private final Time time = new MockTime(1L);
private final StreamsConfig config = new StreamsConfig(configProps());
private final ChangelogReader changelogReader = mock(ChangelogReader.class);
- private final java.util.function.Consumer<Set<TopicPartition>> offsetResetter = topicPartitions -> { };
Review Comment:
This is a leftover from previous commit, we do not need this anymore.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -573,6 +572,45 @@ public void closeCleanAndRecycleState() {
log.info("Closed clean and recycled state");
}
+ /**
+ * Create a standby task from this active task without closing and re-initializing the state stores.
+ * The task should have been in suspended state when calling this function
+ *
+ * TODO: we should be able to not need the input partitions as input param in future but always reuse
+ * the task's input partitions when we have fixed partitions -> tasks mapping
+ */
+ public StandbyTask recycle(final Set<TopicPartition> inputPartitions) {
Review Comment:
Consolidate the logic of recycling within a task, so that it can be extracted from the creator later for recycling after tasks returned from state updater.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -346,22 +365,34 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
throw first.getValue();
}
- tasks.handleNewAssignmentAndCreateTasks(activeTasksToCreate, standbyTasksToCreate, activeTasks.keySet(), standbyTasks.keySet());
+ tasks.createTasks(activeTasksToCreate, standbyTasksToCreate);
+ }
+
+ private Map<TaskId, Set<TopicPartition>> pendingTasksToCreate(final Map<TaskId, Set<TopicPartition>> tasksToCreate) {
+ final Map<TaskId, Set<TopicPartition>> pendingTasks = new HashMap<>();
+ final Iterator<Map.Entry<TaskId, Set<TopicPartition>>> iter = tasksToCreate.entrySet().iterator();
+ while (iter.hasNext()) {
+ final Map.Entry<TaskId, Set<TopicPartition>> entry = iter.next();
+ final TaskId taskId = entry.getKey();
+ if (taskId.topologyName() != null && !topologyMetadata.namedTopologiesView().contains(taskId.topologyName())) {
+ pendingTasks.put(taskId, entry.getValue());
+ iter.remove();
+ }
+ }
+ return pendingTasks;
}
- private void handleCloseAndRecycle(final Set<Task> tasksToRecycle,
- final Set<Task> tasksToCloseClean,
- final Set<Task> tasksToCloseDirty,
- final Map<TaskId, Set<TopicPartition>> activeTasksToCreate,
- final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate,
- final LinkedHashMap<TaskId, RuntimeException> taskCloseExceptions) {
+ private void closeAndRecycleTasks(final Map<Task, Set<TopicPartition>> tasksToRecycle,
Review Comment:
This is a cleanup: we construct the pending tasks to recycle outside the handleCloseAndRecycle so that we do not need to pass in activeTasksToCreate / standbyTasksToCreate anymore.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -859,16 +899,6 @@ void closeAndCleanUpTasks(final Collection<Task> activeTasks, final Collection<T
closeTaskDirty(task);
}
- // TODO: change type to `StreamTask`
Review Comment:
This is for case 3) in the description: this is already in the `completeTaskCloseClean` so is a duplicated code.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreator.java:
##########
@@ -142,33 +135,16 @@ StreamsProducer threadProducer() {
return threadProducer;
}
- void removeRevokedUnknownTasks(final Set<TaskId> assignedTasks) {
Review Comment:
This is for case 2) in description.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -285,34 +286,52 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
final LinkedHashMap<TaskId, RuntimeException> taskCloseExceptions = new LinkedHashMap<>();
final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new HashMap<>(activeTasks);
final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new HashMap<>(standbyTasks);
+ final Map<Task, Set<TopicPartition>> tasksToRecycle = new HashMap<>();
final Comparator<Task> byId = Comparator.comparing(Task::id);
- final Set<Task> tasksToRecycle = new TreeSet<>(byId);
final Set<Task> tasksToCloseClean = new TreeSet<>(byId);
final Set<Task> tasksToCloseDirty = new TreeSet<>(byId);
- // first rectify all existing tasks
+ tasks.purgePendingTasks(activeTasks.keySet(), standbyTasks.keySet());
+
+ // first rectify all existing tasks:
+ // 1. for tasks that are already owned, just update input partitions / resume and skip re-creating them
+ // 2. for tasks that have changed active/standby status, just recycle and skip re-creating them
+ // 3. otherwise, close them since they are no longer owned
for (final Task task : tasks.allTasks()) {
- if (activeTasks.containsKey(task.id()) && task.isActive()) {
- tasks.updateInputPartitionsAndResume(task, activeTasks.get(task.id()));
- activeTasksToCreate.remove(task.id());
- } else if (standbyTasks.containsKey(task.id()) && !task.isActive()) {
- tasks.updateInputPartitionsAndResume(task, standbyTasks.get(task.id()));
- standbyTasksToCreate.remove(task.id());
- } else if (activeTasks.containsKey(task.id()) || standbyTasks.containsKey(task.id())) {
- // check for tasks that were owned previously but have changed active/standby status
- tasksToRecycle.add(task);
+ final TaskId taskId = task.id();
+ if (activeTasksToCreate.containsKey(taskId)) {
+ if (task.isActive()) {
+ final Set<TopicPartition> topicPartitions = activeTasksToCreate.get(taskId);
Review Comment:
This is for case 2) in description: extract all `task.mutate` functions out of `Tasks` and only inside `TaskManager`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java:
##########
@@ -106,7 +111,7 @@ public TaskId id() {
@Override
public Set<TopicPartition> inputPartitions() {
- return inputPartitions;
+ return Collections.unmodifiableSet(inputPartitions);
Review Comment:
Minor cleanup: do not allow callers to modify the set.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -292,7 +292,6 @@ public void suspend() {
partitionGroup.clear();
} finally {
transitToSuspend();
- log.info("Suspended running");
Review Comment:
Minor cleanup: duplicated logging.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org