public void flushState() {
        // this could theoretically throw a ProcessorStateException caused by a ProducerFencedException,
        // but in practice this shouldn't happen for global state update tasks, since the stores are not
        // logged and there are no downstream operators after global stores.
        stateMgr.flush();
        stateMgr.updateChangelogOffsets(offsets);
        stateMgr.checkpoint();
    }
@throws IllegalStateException If store gets registered after initialized is already finished
@throws StreamsException      If the store's change log does not contain the partition
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateUpdateTask.java