@Nonnull
    @Override
    public LocalRecoveryConfig createLocalRecoveryConfig() {
        return localStateStore.getLocalRecoveryConfig();
    }
@Test
    void testForwardingSubtaskLocalStateBaseDirFromLocalStateStore(@TempDir Path tmpFolder)
            throws Exception {
        JobID jobID = new JobID(42L, 43L);
        AllocationID allocationID = new AllocationID(4711L, 23L);
        JobVertexID jobVertexID = new JobVertexID(12L, 34L);
        ExecutionAttemptID executionAttemptID = createExecutionAttemptId(jobVertexID);
        TestCheckpointResponder checkpointResponderMock = new TestCheckpointResponder();

        Executor directExecutor = Executors.directExecutor();

        File[] allocBaseDirs =
                new File[] {
                    TempDirUtils.newFolder(tmpFolder),
                    TempDirUtils.newFolder(tmpFolder),
                    TempDirUtils.newFolder(tmpFolder)
                };

        LocalSnapshotDirectoryProviderImpl directoryProvider =
                new LocalSnapshotDirectoryProviderImpl(allocBaseDirs, jobID, jobVertexID, 0);

        LocalRecoveryConfig localRecoveryConfig =
                LocalRecoveryConfig.backupAndRecoveryEnabled(directoryProvider);

        TaskLocalStateStore taskLocalStateStore =
                new TaskLocalStateStoreImpl(
                        jobID, allocationID, jobVertexID, 13, localRecoveryConfig, directExecutor);

        InMemoryStateChangelogStorage changelogStorage = new InMemoryStateChangelogStorage();

        TaskStateManager taskStateManager =
                taskStateManager(
                        jobID,
                        executionAttemptID,
                        checkpointResponderMock,
                        null,
                        taskLocalStateStore,
                        changelogStorage);

        LocalRecoveryConfig localRecoveryConfFromTaskLocalStateStore =
                taskLocalStateStore.getLocalRecoveryConfig();

        LocalRecoveryConfig localRecoveryConfFromTaskStateManager =
                taskStateManager.createLocalRecoveryConfig();

        for (int i = 0; i < 10; ++i) {
            assertThat(
                            localRecoveryConfFromTaskLocalStateStore
                                    .getLocalStateDirectoryProvider()
                                    .get()
                                    .allocationBaseDirectory(i))
                    .isEqualTo(allocBaseDirs[i % allocBaseDirs.length]);
            assertThat(
                            localRecoveryConfFromTaskStateManager
                                    .getLocalStateDirectoryProvider()
                                    .get()
                                    .allocationBaseDirectory(i))
                    .isEqualTo(allocBaseDirs[i % allocBaseDirs.length]);
        }

        assertThat(localRecoveryConfFromTaskStateManager.isLocalRecoveryEnabled())
                .isEqualTo(localRecoveryConfFromTaskLocalStateStore.isLocalRecoveryEnabled());
    }