@Override
    public void maybeCheckpoint(final boolean enforceCheckpoint) {
        // commitNeeded indicates we may have processed some records since last commit
        // and hence we need to refresh checkpointable offsets regardless whether we should checkpoint or not
        if (commitNeeded || enforceCheckpoint) {
            stateMgr.updateChangelogOffsets(checkpointableOffsets());
        }

        super.maybeCheckpoint(enforceCheckpoint);
    }
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