private void validateClean() {
        // It may be that we failed to commit a task during handleRevocation, but "forgot" this and tried to
        // closeClean in handleAssignment. We should throw if we detect this to force the TaskManager to closeDirty
        if (commitNeeded) {
            log.debug("Tried to close clean but there was pending uncommitted data, this means we failed to"
                          + " commit and should close as dirty instead");
            throw new TaskMigratedException("Tried to close dirty task as clean");
        }
    }
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/StreamTask.java