@Override
    public void registerStore(final StateStore store,
                              final StateRestoreCallback stateRestoreCallback,
                              final CommitCallback commitCallback) {
        final String storeName = store.name();

        // TODO (KAFKA-12887): we should not trigger user's exception handler for illegal-argument but always
        // fail-crash; in this case we would not need to immediately close the state store before throwing
        if (CHECKPOINT_FILE_NAME.equals(storeName)) {
            store.close();
            throw new IllegalArgumentException(format("%sIllegal store name: %s, which collides with the pre-defined " +
                "checkpoint file name", logPrefix, storeName));
        }

        if (stores.containsKey(storeName)) {
            store.close();
            throw new IllegalArgumentException(format("%sStore %s has already been registered.", logPrefix, storeName));
        }

        if (stateRestoreCallback instanceof StateRestoreListener) {
            log.warn("The registered state restore callback is also implementing the state restore listener interface, " +
                    "which is not expected and would be ignored");
        }

        final StateStoreMetadata storeMetadata = isLoggingEnabled(storeName) ?
            new StateStoreMetadata(
                store,
                getStorePartition(storeName),
                stateRestoreCallback,
                commitCallback,
                converterForStore(store)) :
            new StateStoreMetadata(store, commitCallback);

        // register the store first, so that if later an exception is thrown then eventually while we call `close`
        // on the state manager this state store would be closed as well
        stores.put(storeName, storeMetadata);

        if (!stateUpdaterEnabled) {
            maybeRegisterStoreWithChangelogReader(storeName);
        }

        log.debug("Registered state store {} to its state manager", storeName);
    }
@Test
    public void shouldThrowProcessorStateExceptionOnCloseIfStoreThrowsAnException() {
        final RuntimeException exception = new RuntimeException("KABOOM!");
        final ProcessorStateManager stateManager = getStateManager(Task.TaskType.ACTIVE);
        final MockKeyValueStore stateStore = new MockKeyValueStore(persistentStoreName, true) {
            @Override
            public void close() {
                throw exception;
            }
        };
        stateManager.registerStore(stateStore, stateStore.stateRestoreCallback, null);

        final ProcessorStateException thrown = assertThrows(ProcessorStateException.class, stateManager::close);
        assertEquals(exception, thrown.getCause());
    }