@Override
    public void close() throws ProcessorStateException {
        log.debug("Closing its state manager and all the registered state stores: {}", stores);

        if (!stateUpdaterEnabled) {
            changelogReader.unregister(getAllChangelogTopicPartitions());
        }

        RuntimeException firstException = null;
        // attempting to close the stores, just in case they
        // are not closed by a ProcessorNode yet
        if (!stores.isEmpty()) {
            for (final Map.Entry<String, StateStoreMetadata> entry : stores.entrySet()) {
                final StateStore store = entry.getValue().stateStore;
                log.trace("Closing store {}", store.name());
                try {
                    store.close();
                } 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 close state store %s", logPrefix, store.name()), exception);
                    }
                    log.error("Failed to close state store {}: ", store.name(), exception);
                }
            }

            stores.clear();
        }

        if (firstException != null) {
            throw firstException;
        }
    }
{@link StateStore#close() Close} all stores (even in case of failure).
Log all exceptions and re-throw the first exception that occurred at the end.

@throws ProcessorStateException if any error happens when closing the state stores
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java