### Question:
TaskStateManagerImpl implements TaskStateManager { @Nonnull @Override public LocalRecoveryConfig createLocalRecoveryConfig() { return localStateStore.getLocalRecoveryConfig(); } TaskStateManagerImpl(
		@Nonnull JobID jobId,
		@Nonnull ExecutionAttemptID executionAttemptID,
		@Nonnull TaskLocalStateStore localStateStore,
		@Nullable JobManagerTaskRestore jobManagerTaskRestore,
		@Nonnull CheckpointResponder checkpointResponder); @Override void reportTaskStateSnapshots(
		@Nonnull CheckpointMetaData checkpointMetaData,
		@Nonnull CheckpointMetrics checkpointMetrics,
		@Nullable TaskStateSnapshot acknowledgedState,
		@Nullable TaskStateSnapshot localState); @Nonnull @Override PrioritizedOperatorSubtaskState prioritizedOperatorState(OperatorID operatorID); @Nonnull @Override LocalRecoveryConfig createLocalRecoveryConfig(); @Override void notifyCheckpointComplete(long checkpointId);  }### Answer:
@Test public void testForwardingSubtaskLocalStateBaseDirFromLocalStateStore() throws IOException { JobID jobID = new JobID(42L, 43L); AllocationID allocationID = new AllocationID(4711L, 23L); JobVertexID jobVertexID = new JobVertexID(12L, 34L); ExecutionAttemptID executionAttemptID = new ExecutionAttemptID(23L, 24L); TestCheckpointResponder checkpointResponderMock = new TestCheckpointResponder(); Executor directExecutor = Executors.directExecutor(); TemporaryFolder tmpFolder = new TemporaryFolder(); try { tmpFolder.create(); File[] allocBaseDirs = new File[]{tmpFolder.newFolder(), tmpFolder.newFolder(), tmpFolder.newFolder()}; LocalRecoveryDirectoryProviderImpl directoryProvider = new LocalRecoveryDirectoryProviderImpl(allocBaseDirs, jobID, jobVertexID, 0); LocalRecoveryConfig localRecoveryConfig = new LocalRecoveryConfig(true, directoryProvider); TaskLocalStateStore taskLocalStateStore = new TaskLocalStateStoreImpl(jobID, allocationID, jobVertexID, 13, localRecoveryConfig, directExecutor); TaskStateManager taskStateManager = taskStateManager( jobID, executionAttemptID, checkpointResponderMock, null, taskLocalStateStore); LocalRecoveryConfig localRecoveryConfFromTaskLocalStateStore = taskLocalStateStore.getLocalRecoveryConfig(); LocalRecoveryConfig localRecoveryConfFromTaskStateManager = taskStateManager.createLocalRecoveryConfig(); for (int i = 0; i < 10; ++i) { Assert.assertEquals(allocBaseDirs[i % allocBaseDirs.length], localRecoveryConfFromTaskLocalStateStore.getLocalStateDirectoryProvider().allocationBaseDirectory(i)); Assert.assertEquals(allocBaseDirs[i % allocBaseDirs.length], localRecoveryConfFromTaskStateManager.getLocalStateDirectoryProvider().allocationBaseDirectory(i)); } Assert.assertEquals( localRecoveryConfFromTaskLocalStateStore.isLocalRecoveryEnabled(), localRecoveryConfFromTaskStateManager.isLocalRecoveryEnabled()); } finally { tmpFolder.delete(); } }