@Test public void testResolveCheckpointStorageLocation() throws Exception { final FileSystem checkpointFileSystem = mock(FileSystem.class); final FsCheckpointStorage storage = new FsCheckpointStorage( new TestingPath("hdfs: null, new JobID(), FILE_SIZE_THRESHOLD); final FsCheckpointStorageLocation checkpointStreamFactory = (FsCheckpointStorageLocation) storage.resolveCheckpointStorageLocation(1L, CheckpointStorageLocationReference.getDefault()); assertEquals(checkpointFileSystem, checkpointStreamFactory.getFileSystem()); final CheckpointStorageLocationReference savepointLocationReference = AbstractFsCheckpointStorage.encodePathAsReference(new Path("file: final FsCheckpointStorageLocation savepointStreamFactory = (FsCheckpointStorageLocation) storage.resolveCheckpointStorageLocation(2L, savepointLocationReference); final FileSystem fileSystem = savepointStreamFactory.getFileSystem(); assertTrue(fileSystem instanceof LocalFileSystem); }
@Override public CheckpointStreamFactory resolveCheckpointStorageLocation( long checkpointId, CheckpointStorageLocationReference reference) throws IOException { if (reference.isDefaultReference()) { final Path checkpointDir = createCheckpointDirectory(checkpointsDirectory, checkpointId); return new FsCheckpointStorageLocation( fileSystem, checkpointDir, sharedStateDirectory, taskOwnedStateDirectory, reference, fileSizeThreshold); } else { final Path path = decodePathFromReference(reference); return new FsCheckpointStorageLocation( path.getFileSystem(), path, path, path, reference, fileSizeThreshold); } }
FsCheckpointStorage extends AbstractFsCheckpointStorage { @Override public CheckpointStreamFactory resolveCheckpointStorageLocation( long checkpointId, CheckpointStorageLocationReference reference) throws IOException { if (reference.isDefaultReference()) { final Path checkpointDir = createCheckpointDirectory(checkpointsDirectory, checkpointId); return new FsCheckpointStorageLocation( fileSystem, checkpointDir, sharedStateDirectory, taskOwnedStateDirectory, reference, fileSizeThreshold); } else { final Path path = decodePathFromReference(reference); return new FsCheckpointStorageLocation( path.getFileSystem(), path, path, path, reference, fileSizeThreshold); } } }
FsCheckpointStorage extends AbstractFsCheckpointStorage { @Override public CheckpointStreamFactory resolveCheckpointStorageLocation( long checkpointId, CheckpointStorageLocationReference reference) throws IOException { if (reference.isDefaultReference()) { final Path checkpointDir = createCheckpointDirectory(checkpointsDirectory, checkpointId); return new FsCheckpointStorageLocation( fileSystem, checkpointDir, sharedStateDirectory, taskOwnedStateDirectory, reference, fileSizeThreshold); } else { final Path path = decodePathFromReference(reference); return new FsCheckpointStorageLocation( path.getFileSystem(), path, path, path, reference, fileSizeThreshold); } } FsCheckpointStorage(
			Path checkpointBaseDirectory,
			@Nullable Path defaultSavepointDirectory,
			JobID jobId,
			int fileSizeThreshold);  FsCheckpointStorage(
			FileSystem fs,
			Path checkpointBaseDirectory,
			@Nullable Path defaultSavepointDirectory,
			JobID jobId,
			int fileSizeThreshold); }
FsCheckpointStorage extends AbstractFsCheckpointStorage { @Override public CheckpointStreamFactory resolveCheckpointStorageLocation( long checkpointId, CheckpointStorageLocationReference reference) throws IOException { if (reference.isDefaultReference()) { final Path checkpointDir = createCheckpointDirectory(checkpointsDirectory, checkpointId); return new FsCheckpointStorageLocation( fileSystem, checkpointDir, sharedStateDirectory, taskOwnedStateDirectory, reference, fileSizeThreshold); } else { final Path path = decodePathFromReference(reference); return new FsCheckpointStorageLocation( path.getFileSystem(), path, path, path, reference, fileSizeThreshold); } } FsCheckpointStorage(
			Path checkpointBaseDirectory,
			@Nullable Path defaultSavepointDirectory,
			JobID jobId,
			int fileSizeThreshold);  FsCheckpointStorage(
			FileSystem fs,
			Path checkpointBaseDirectory,
			@Nullable Path defaultSavepointDirectory,
			JobID jobId,
			int fileSizeThreshold); Path getCheckpointsDirectory(); @Override boolean supportsHighlyAvailableStorage(); @Override CheckpointStorageLocation initializeLocationForCheckpoint(long checkpointId); @Override CheckpointStreamFactory resolveCheckpointStorageLocation(
			long checkpointId,
			CheckpointStorageLocationReference reference); @Override CheckpointStateOutputStream createTaskOwnedStateStream(); }
FsCheckpointStorage extends AbstractFsCheckpointStorage { @Override public CheckpointStreamFactory resolveCheckpointStorageLocation( long checkpointId, CheckpointStorageLocationReference reference) throws IOException { if (reference.isDefaultReference()) { final Path checkpointDir = createCheckpointDirectory(checkpointsDirectory, checkpointId); return new FsCheckpointStorageLocation( fileSystem, checkpointDir, sharedStateDirectory, taskOwnedStateDirectory, reference, fileSizeThreshold); } else { final Path path = decodePathFromReference(reference); return new FsCheckpointStorageLocation( path.getFileSystem(), path, path, path, reference, fileSizeThreshold); } } FsCheckpointStorage(
			Path checkpointBaseDirectory,
			@Nullable Path defaultSavepointDirectory,
			JobID jobId,
			int fileSizeThreshold);  FsCheckpointStorage(
			FileSystem fs,
			Path checkpointBaseDirectory,
			@Nullable Path defaultSavepointDirectory,
			JobID jobId,
			int fileSizeThreshold); Path getCheckpointsDirectory(); @Override boolean supportsHighlyAvailableStorage(); @Override CheckpointStorageLocation initializeLocationForCheckpoint(long checkpointId); @Override CheckpointStreamFactory resolveCheckpointStorageLocation(
			long checkpointId,
			CheckpointStorageLocationReference reference); @Override CheckpointStateOutputStream createTaskOwnedStateStream();  }