public void flushCache() {
        RuntimeException firstException = null;
        // attempting to flush the stores
        if (!stores.isEmpty()) {
            log.debug("Flushing all store caches registered in the state manager: {}", stores);
            for (final StateStoreMetadata metadata : stores.values()) {
                final StateStore store = metadata.stateStore;

                try {
                    // buffer should be flushed to send all records to changelog
                    if (store instanceof TimeOrderedKeyValueBuffer) {
                        store.flush();
                    } else if (store instanceof CachedStateStore) {
                        ((CachedStateStore) store).flushCache();
                    }
                    log.trace("Flushed cache or buffer {}", store.name());
                } catch (final RuntimeException exception) {
                    if (firstException == null) {
                        // do NOT wrap the error if it is actually caused by Streams itself
                        if (exception instanceof StreamsException) {
                            firstException = exception;
                        } else {
                            firstException = new ProcessorStateException(
                                format("%sFailed to flush cache of store %s", logPrefix, store.name()),
                                exception
                            );
                        }
                    }
                    log.error("Failed to flush cache of store {}: ", store.name(), exception);
                }
            }
        }

        if (firstException != null) {
            throw firstException;
        }
    }
@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/ProcessorStateManager.java