private void maybeThrowTaskExceptions(final Map<TaskId, RuntimeException> taskExceptions) {
        if (!taskExceptions.isEmpty()) {
            log.error("Get exceptions for the following tasks: {}", taskExceptions);

            final Set<TaskId> aggregatedCorruptedTaskIds = new HashSet<>();
            StreamsException lastFatal = null;
            TaskMigratedException lastTaskMigrated = null;
            for (final Map.Entry<TaskId, RuntimeException> entry : taskExceptions.entrySet()) {
                final TaskId taskId = entry.getKey();
                final RuntimeException exception = entry.getValue();

                if (exception instanceof StreamsException) {
                    if (exception instanceof TaskMigratedException) {
                        lastTaskMigrated = (TaskMigratedException) exception;
                    } else if (exception instanceof TaskCorruptedException) {
                        log.warn("Encounter corrupted task " + taskId + ", will group it with other corrupted tasks " +
                            "and handle together", exception);
                        aggregatedCorruptedTaskIds.add(taskId);
                    } else {
                        ((StreamsException) exception).setTaskId(taskId);
                        lastFatal = (StreamsException) exception;
                    }
                } else if (exception instanceof KafkaException) {
                    lastFatal = new StreamsException(exception, taskId);
                } else {
                    lastFatal = new StreamsException("Encounter unexpected fatal error for task " + taskId, exception, taskId);
                }
            }

            if (lastFatal != null) {
                throw lastFatal;
            } else if (lastTaskMigrated != null) {
                throw lastTaskMigrated;
            } else {
                throw new TaskCorruptedException(aggregatedCorruptedTaskIds);
            }
        }
    }
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws StreamsException fatal error while creating / initializing the task

public for upgrade testing only
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java