@Override
    public void maybeInitTaskTimeoutOrThrow(final long currentWallClockMs,
                                            final Exception cause) {
        if (deadlineMs == NO_DEADLINE) {
            deadlineMs = currentWallClockMs + config.taskTimeoutMs;
        } else if (currentWallClockMs > deadlineMs) {
            final String errorMessage = String.format(
                "Task %s did not make progress within %d ms. Adjust `%s` if needed.",
                id,
                currentWallClockMs - deadlineMs + config.taskTimeoutMs,
                StreamsConfig.TASK_TIMEOUT_MS_CONFIG
            );

            if (cause != null) {
                throw new StreamsException(new TimeoutException(errorMessage, cause), id);
            } else {
                throw new StreamsException(new TimeoutException(errorMessage), id);
            }
        }

        if (cause != null) {
            log.debug(
                String.format(
                    "Task did not make progress. Remaining time to deadline %d; retrying.",
                    deadlineMs - currentWallClockMs
                ),
                cause
            );
        } else {
            log.debug(
                "Task did not make progress. Remaining time to deadline {}; retrying.",
                deadlineMs - currentWallClockMs
            );
        }

    }
The following exceptions maybe thrown from the state manager flushing call

@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
streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java