lucasbru commented on code in PR #20767:
URL: https://github.com/apache/kafka/pull/20767#discussion_r2546130213
##########
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:
Wait, you are saying a `StreamsException` which is not
`TaskMigrationException` should cause a fatal error, no? Because this is not
what the PR is doing now.
--
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]
