@Override
    public void maybeCheckpoint(final boolean enforceCheckpoint) {
        final Map<TopicPartition, Long> offsetSnapshot = stateMgr.changelogOffsets();
        if (StateManagerUtil.checkpointNeeded(enforceCheckpoint, offsetSnapshotSinceLastFlush, offsetSnapshot)) {
            // the state's current offset would be used to checkpoint
            stateMgr.flush();
            stateMgr.checkpoint();
            offsetSnapshotSinceLastFlush = new HashMap<>(offsetSnapshot);
        }
    }
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