lucasbru commented on code in PR #20767:
URL: https://github.com/apache/kafka/pull/20767#discussion_r2528358146
##########
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:
Hmm, the comment of maybeCheckpoint mentions this:
```
* @throws TaskMigratedException recoverable error sending changelog
records that would cause the task to be removed
* @throws StreamsException fatal error when flushing the state store,
for example sending changelog records failed
* or flushing state store get IO errors; such
error should cause the thread to die
```
That would actually indicate that we have to treat streams exception fatally
here. But I haven't yet tracked down where they come from and if they can
happen in the state updater. Maybe that comment should be made more precise?
But going by the commen, we actually need to fail here.
--
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]
