lucasbru commented on code in PR #20767:
URL: https://github.com/apache/kafka/pull/20767#discussion_r2533210125
##########
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:
The question is
- Where does the `TaskMigratedException` come from. Can it happen, or
should be update the javadoc?
- Why does the comment indicate that the thread should die on an I/O error?
@mjsax any idea? If the comment is correct, we need to call
handleStreamsExceptionWithoutTask to handle the exception. If not, we should
change the comment.
--
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]
