Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdater.java (Java)

Code before:
            final Set<Task> corruptedTasks = new HashSet<>();
            final Set<TopicPartition> changelogsOfCorruptedTasks = new HashSet<>();
            for (final TaskId taskId : corruptedTaskIds) {
                final Task corruptedTask = updatingTasks.get(taskId);
                if (corruptedTask == null) {
                    throw new IllegalStateException("Task " + taskId + " is corrupted but is not updating. " + BUG_ERROR_MESSAGE);
                }
                corruptedTasks.add(corruptedTask);
                removeCheckpointForCorruptedTask(corruptedTask);
                changelogsOfCorruptedTasks.addAll(corruptedTask.changelogPartitions());
            }
            changelogReader.unregister(changelogsOfCorruptedTasks);
            corruptedTasks.forEach(
                task -> addToExceptionsAndFailedTasksThenRemoveFromUpdatingTasks(new ExceptionAndTask(taskCorruptedException, task))
            );
        }

        // TODO: we can let the exception encode the actual corrupted changelog partitions and only
        //       mark those instead of marking all changelogs
        private void removeCheckpointForCorruptedTask(final Task task) {
            try {
                task.markChangelogAsCorrupted(task.changelogPartitions());

                // we need to enforce a checkpoint that removes the corrupted partitions
                measureCheckpointLatency(() -> task.maybeCheckpoint(true));
            } catch (final StreamsException e) {
                log.warn("Checkpoint failed for corrupted task {}", task.id(), e);
            }
        }

        private void handleStreamsException(final StreamsException streamsException) {
            log.info("Encountered streams exception: ", streamsException);
            if (streamsException.taskId().isPresent()) {
                handleStreamsExceptionWithTask(streamsException, streamsException.taskId().get());
            } else {
                handleStreamsExceptionWithoutTask(streamsException);
            }
        }

        private void handleStreamsExceptionWithTask(final StreamsException streamsException, final TaskId failedTaskId) {
            if (updatingTasks.containsKey(failedTaskId)) {
                addToExceptionsAndFailedTasksThenRemoveFromUpdatingTasks(
                    new ExceptionAndTask(streamsException, updatingTasks.get(failedTaskId))
                );
            } else if (pausedTasks.containsKey(failedTaskId)) {
                addToExceptionsAndFailedTasksThenRemoveFromPausedTasks(
                    new ExceptionAndTask(streamsException, pausedTasks.get(failedTaskId))
                );
            } else {
                throw new IllegalStateException("Task " + failedTaskId + " failed but is not updating or paused. " + BUG_ERROR_MESSAGE);
            }

Review comment (suggestion):
In general, we treat `StreamsException` as fatal, and should never swallow it. What is the reason for this change?

Also: maybe rename `e` -> `swallow` to make clear that we swallow intentionally, and not by mistake.

Code after:
            final Set<Task> corruptedTasks = new HashSet<>();
            final Set<TopicPartition> changelogsOfCorruptedTasks = new HashSet<>();
            for (final TaskId taskId : corruptedTaskIds) {
                final Task corruptedTask = updatingTasks.get(taskId);
                if (corruptedTask == null) {
                    throw new IllegalStateException("Task " + taskId + " is corrupted but is not updating. " + BUG_ERROR_MESSAGE);
                }
                corruptedTasks.add(corruptedTask);
                removeCheckpointForCorruptedTask(corruptedTask);
                changelogsOfCorruptedTasks.addAll(corruptedTask.changelogPartitions());
            }
            changelogReader.unregister(changelogsOfCorruptedTasks);
            corruptedTasks.forEach(
                task -> addToExceptionsAndFailedTasksThenRemoveFromUpdatingTasks(new ExceptionAndTask(taskCorruptedException, task))
            );
        }

        // TODO: we can let the exception encode the actual corrupted changelog partitions and only
        //       mark those instead of marking all changelogs
        private void removeCheckpointForCorruptedTask(final Task task) {
            try {
                task.markChangelogAsCorrupted(task.changelogPartitions());

                // we need to enforce a checkpoint that removes the corrupted partitions
                measureCheckpointLatency(() -> task.maybeCheckpoint(true));
            } catch (final StreamsException swallow) {
                log.warn("Checkpoint failed for corrupted task {}", task.id(), swallow);
            }
        }

        private void handleStreamsException(final StreamsException streamsException) {
            log.info("Encountered streams exception: ", streamsException);
            if (streamsException.taskId().isPresent()) {
                handleStreamsExceptionWithTask(streamsException, streamsException.taskId().get());
            } else {
                handleStreamsExceptionWithoutTask(streamsException);
            }
        }

        private void handleStreamsExceptionWithTask(final StreamsException streamsException, final TaskId failedTaskId) {
            if (updatingTasks.containsKey(failedTaskId)) {
                addToExceptionsAndFailedTasksThenRemoveFromUpdatingTasks(
                    new ExceptionAndTask(streamsException, updatingTasks.get(failedTaskId))
                );
            } else if (pausedTasks.containsKey(failedTaskId)) {
                addToExceptionsAndFailedTasksThenRemoveFromPausedTasks(
                    new ExceptionAndTask(streamsException, pausedTasks.get(failedTaskId))
                );
            } else {
                throw new IllegalStateException("Task " + failedTaskId + " failed but is not updating or paused. " + BUG_ERROR_MESSAGE);
            }