public void shutdown() {
        HashMap<JobID, Optional<StateChangelogStorage<?>>> toReleaseStorage;
        HashMap<JobID, StateChangelogStorageView<ChangelogStateHandleStreamImpl>>
                toReleaseStorageView;
        synchronized (lock) {
            if (closed) {
                return;
            }
            closed = true;

            toReleaseStorage = new HashMap<>(changelogStoragesByJobId);
            toReleaseStorageView = new HashMap<>(changelogStorageViewsByJobId);
            changelogStoragesByJobId.clear();
            changelogStorageViewsByJobId.clear();
        }

        ShutdownHookUtil.removeShutdownHook(shutdownHook, getClass().getSimpleName(), LOG);

        LOG.info("Shutting down TaskExecutorStateChangelogStoragesManager.");

        for (Map.Entry<JobID, Optional<StateChangelogStorage<?>>> entry :
                toReleaseStorage.entrySet()) {
            entry.getValue().ifPresent(this::doRelease);
        }
        for (Map.Entry<JobID, StateChangelogStorageView<ChangelogStateHandleStreamImpl>> entry :
                toReleaseStorageView.entrySet()) {
            doRelease(entry.getValue());
        }
    }
@Test
    void testShutdown() throws IOException {
        StateChangelogStorageLoader.initialize(TestStateChangelogStorageFactory.pluginManager);
        TaskExecutorStateChangelogStoragesManager manager =
                new TaskExecutorStateChangelogStoragesManager();
        Configuration configuration = new Configuration();
        configuration.set(
                StateChangelogOptions.STATE_CHANGE_LOG_STORAGE,
                TestStateChangelogStorageFactory.identifier);
        JobID jobId1 = new JobID(1L, 1L);
        StateChangelogStorage<?> storage1 =
                manager.stateChangelogStorageForJob(
                        jobId1,
                        configuration,
                        createUnregisteredTaskManagerJobMetricGroup(),
                        TestLocalRecoveryConfig.disabled());
        assertThat(storage1).isInstanceOf(TestStateChangelogStorage.class);
        assertThat(((TestStateChangelogStorage) storage1).closed).isFalse();

        JobID jobId2 = new JobID(1L, 2L);
        StateChangelogStorage<?> storage2 =
                manager.stateChangelogStorageForJob(
                        jobId1,
                        configuration,
                        createUnregisteredTaskManagerJobMetricGroup(),
                        TestLocalRecoveryConfig.disabled());
        assertThat(storage2).isInstanceOf(TestStateChangelogStorage.class);
        assertThat(((TestStateChangelogStorage) storage2).closed).isFalse();

        manager.shutdown();
        assertThat(((TestStateChangelogStorage) storage1).closed).isTrue();
        assertThat(((TestStateChangelogStorage) storage2).closed).isTrue();

        StateChangelogStorageLoader.initialize(null);
    }