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/08/11 03:53:53 UTC
[GitHub] [kafka] guozhangwang commented on a diff in pull request #12466: KAFKA-10199: Handle task closure and recycling from state updater
guozhangwang commented on code in PR #12466:
URL: https://github.com/apache/kafka/pull/12466#discussion_r943076070
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -656,6 +657,71 @@ boolean tryToCompleteRestoration(final long now, final java.util.function.Consum
return allRunning;
}
+ private void addTaskstoStateUpdater() {
+ for (final Task task : tasks.drainPendingTaskToInit()) {
+ task.initializeIfNeeded();
+ stateUpdater.add(task);
+ }
+ }
+
+ private void handleRemovedTasksFromStateUpdater() {
+ final Map<TaskId, RuntimeException> taskExceptions = new LinkedHashMap<>();
+ final Set<Task> tasksToCloseDirty = new TreeSet<>(Comparator.comparing(Task::id));
+
+ for (final Task task : stateUpdater.drainRemovedTasks()) {
+ final TaskId taskId = task.id();
+ Set<TopicPartition> inputPartitions;
+ if ((inputPartitions = tasks.removePendingTaskToRecycle(task.id())) != null) {
+ try {
+ final Task newTask = task.isActive() ?
+ convertActiveToStandby((StreamTask) task, inputPartitions) :
+ convertStandbyToActive((StandbyTask) task, inputPartitions);
+ newTask.initializeIfNeeded();
+ stateUpdater.add(newTask);
+ } catch (final RuntimeException e) {
+ final String uncleanMessage = String.format("Failed to recycle task %s cleanly. " +
+ "Attempting to close remaining tasks before re-throwing:", taskId);
+ log.error(uncleanMessage, e);
+
+ if (task.state() != State.CLOSED) {
+ tasksToCloseDirty.add(task);
+ }
+
+ taskExceptions.putIfAbsent(taskId, e);
+ }
+ } else if (tasks.removePendingTaskToClose(task.id())) {
+ try {
+ task.closeClean();
+ if (task.isActive()) {
+ activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
+ }
+ } catch (final RuntimeException e) {
+ final String uncleanMessage = String.format("Failed to close task %s cleanly. " +
+ "Attempting to close remaining tasks before re-throwing:", task.id());
+ log.error(uncleanMessage, e);
+
+ if (task.state() != State.CLOSED) {
+ tasksToCloseDirty.add(task);
+ }
+
+ taskExceptions.putIfAbsent(task.id(), e);
+ }
+ } else if ((inputPartitions = tasks.removePendingTaskToUpdateInputPartitions(task.id())) != null) {
+ task.updateInputPartitions(inputPartitions, topologyMetadata.nodeToSourceTopics(task.id()));
+ stateUpdater.add(task);
Review Comment:
Found this issue when adding unit test: after update the input partitions, need to add it back to state updater.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -656,6 +657,71 @@ boolean tryToCompleteRestoration(final long now, final java.util.function.Consum
return allRunning;
}
+ private void addTaskstoStateUpdater() {
+ for (final Task task : tasks.drainPendingTaskToInit()) {
+ task.initializeIfNeeded();
+ stateUpdater.add(task);
+ }
+ }
+
+ private void handleRemovedTasksFromStateUpdater() {
+ final Map<TaskId, RuntimeException> taskExceptions = new LinkedHashMap<>();
+ final Set<Task> tasksToCloseDirty = new TreeSet<>(Comparator.comparing(Task::id));
+
+ for (final Task task : stateUpdater.drainRemovedTasks()) {
+ final TaskId taskId = task.id();
+ Set<TopicPartition> inputPartitions;
+ if ((inputPartitions = tasks.removePendingTaskToRecycle(task.id())) != null) {
+ try {
+ final Task newTask = task.isActive() ?
+ convertActiveToStandby((StreamTask) task, inputPartitions) :
+ convertStandbyToActive((StandbyTask) task, inputPartitions);
+ newTask.initializeIfNeeded();
+ stateUpdater.add(newTask);
+ } catch (final RuntimeException e) {
+ final String uncleanMessage = String.format("Failed to recycle task %s cleanly. " +
+ "Attempting to close remaining tasks before re-throwing:", taskId);
+ log.error(uncleanMessage, e);
+
+ if (task.state() != State.CLOSED) {
+ tasksToCloseDirty.add(task);
+ }
+
+ taskExceptions.putIfAbsent(taskId, e);
+ }
+ } else if (tasks.removePendingTaskToClose(task.id())) {
+ try {
+ task.closeClean();
Review Comment:
Should not remove from the `tasks` since it's not in it, hence we could not call `closeClean`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java:
##########
@@ -43,9 +43,7 @@
class Tasks {
private final Logger log;
- // 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<>();
Review Comment:
I tried to convert the types to StreamTask/StandbyTask, but realized that the blocker from doing so is because we have the stub class `StateMachineTask` in other unit tests which are not either. I then looked into if there's a better way to remove that stub class and instead just use a mock StreamTask/StandbyTask but that's actually not very convienent. So I figured unless we'd eventually remove `StateMachineTask` we'd have to keep as is.
In a second thought, I think we'd probably want to remove it eventually, so I'd add those TODOs back.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -1076,59 +1135,51 @@ private Collection<Task> tryCloseCleanActiveTasks(final Collection<Task> activeT
} else {
try {
taskExecutor.commitOffsetsOrTransaction(consumedOffsetsAndMetadataPerTask);
-
- for (final Task task : activeTaskIterable()) {
- try {
- task.postCommit(true);
- } catch (final RuntimeException e) {
- log.error("Exception caught while post-committing task " + task.id(), e);
- maybeWrapAndSetFirstException(firstException, e, task.id());
- tasksToCloseDirty.add(task);
- tasksToCloseClean.remove(task);
- }
- }
- } catch (final TimeoutException timeoutException) {
- firstException.compareAndSet(null, timeoutException);
-
- tasksToCloseClean.removeAll(tasksToCommit);
- tasksToCloseDirty.addAll(tasksToCommit);
- } catch (final TaskCorruptedException taskCorruptedException) {
- firstException.compareAndSet(null, taskCorruptedException);
-
- final Set<TaskId> corruptedTaskIds = taskCorruptedException.corruptedTasks();
- final Set<Task> corruptedTasks = tasksToCommit
+ } catch (final RuntimeException e) {
+ log.error("Exception caught while committing tasks " + consumedOffsetsAndMetadataPerTask.keySet(), e);
+ // TODO: should record the task ids when handling this exception
+ maybeSetFirstException(false, e, firstException);
+
+ if (e instanceof TaskCorruptedException) {
+ final TaskCorruptedException taskCorruptedException = (TaskCorruptedException) e;
+ final Set<TaskId> corruptedTaskIds = taskCorruptedException.corruptedTasks();
+ final Set<Task> corruptedTasks = tasksToCommit
.stream()
.filter(task -> corruptedTaskIds.contains(task.id()))
.collect(Collectors.toSet());
+ tasksToCloseClean.removeAll(corruptedTasks);
+ tasksToCloseDirty.addAll(corruptedTasks);
+ } else {
+ // If the commit fails, everyone who participated in it must be closed dirty
+ tasksToCloseClean.removeAll(tasksToCommit);
+ tasksToCloseDirty.addAll(tasksToCommit);
+ }
+ }
- tasksToCloseClean.removeAll(corruptedTasks);
- tasksToCloseDirty.addAll(corruptedTasks);
- } catch (final RuntimeException e) {
- log.error("Exception caught while committing tasks during shutdown", e);
- firstException.compareAndSet(null, e);
-
- // If the commit fails, everyone who participated in it must be closed dirty
- tasksToCloseClean.removeAll(tasksToCommit);
- tasksToCloseDirty.addAll(tasksToCommit);
+ for (final Task task : activeTaskIterable()) {
+ try {
+ task.postCommit(true);
+ } catch (final RuntimeException e) {
+ log.error("Exception caught while post-committing task " + task.id(), e);
+ maybeSetFirstException(false, maybeWrapTaskException(e, task.id()), firstException);
+ tasksToCloseDirty.add(task);
+ tasksToCloseClean.remove(task);
+ }
}
}
for (final Task task : tasksToCloseClean) {
try {
task.suspend();
- final RuntimeException exception = completeTaskCloseClean(task);
- if (exception != null) {
- firstException.compareAndSet(null, exception);
- }
- } catch (final StreamsException e) {
- log.error("Exception caught while clean-closing task " + task.id(), e);
- e.setTaskId(task.id());
- firstException.compareAndSet(null, e);
- tasksToCloseDirty.add(task);
+ closeTaskClean(task);
Review Comment:
This is the minor cleanup for exception wrapping / handling.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -298,13 +298,12 @@ public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
logPrefix
);
- 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 Set<Task> tasksToCloseClean = new TreeSet<>(Comparator.comparing(Task::id));
- tasks.purgePendingTasks(activeTasks.keySet(), standbyTasks.keySet());
+ tasks.purgePendingTasksToCreate(activeTasks.keySet(), standbyTasks.keySet());
Review Comment:
You're right! Will do so.
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/ActiveTaskCreatorTest.java:
##########
@@ -71,6 +72,8 @@ public class ActiveTaskCreatorTest {
private StateDirectory stateDirectory;
@Mock(type = MockType.NICE)
private ChangelogReader changeLogReader;
+ @Mock(type = MockType.NICE)
+ private ProcessorStateManager stateManager;
Review Comment:
Sorry, forget to remove it; it's for another work to address comments for the test coverage.
--
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