Nikita-Shupletsov commented on code in PR #20767:
URL: https://github.com/apache/kafka/pull/20767#discussion_r2528399818
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java:
##########
@@ -606,14 +609,19 @@ private boolean removeFailedTask(final TaskId taskId,
final CompletableFuture<Re
private void pauseTask(final Task task) {
final TaskId taskId = task.id();
// do not need to unregister changelog partitions for paused tasks
- measureCheckpointLatency(() -> task.maybeCheckpoint(true));
- pausedTasks.put(taskId, task);
- updatingTasks.remove(taskId);
- if (task.isActive()) {
- transitToUpdateStandbysIfOnlyStandbysLeft();
+ try {
+ measureCheckpointLatency(() -> task.maybeCheckpoint(true));
+ pausedTasks.put(taskId, task);
+ updatingTasks.remove(taskId);
+ if (task.isActive()) {
+ transitToUpdateStandbysIfOnlyStandbysLeft();
+ }
+ log.info((task.isActive() ? "Active" : "Standby")
+ + " task " + task.id() + " was paused from the updating
tasks and added to the paused tasks.");
+
+ } catch (final StreamsException streamsException) {
Review Comment:
>But I haven't yet tracked down where they come from and if they can happen
in the state updater
the problem I am trying to address by this PR is pretty much that
`maybeCheckpoint` threw an exception when was called by the state updater.
Which caused the state updater thread to die and the stream thread shutdown to
hang.
even though it happened during `maybeCheckpointTasks` not during
`pauseTasks`, but it may as well.
`maybeCheckpoint` threw an exception because the RockDB failed to flush(I
assume due to some issues with the file system)
--
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]
