### Question:
FsCheckpointStorage extends AbstractFsCheckpointStorage { @Override public CheckpointStateOutputStream createTaskOwnedStateStream() throws IOException { return new FsCheckpointStateOutputStream( taskOwnedStateDirectory, fileSystem, FsCheckpointStreamFactory.DEFAULT_WRITE_BUFFER_SIZE, fileSizeThreshold); } FsCheckpointStorage(
			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();  }### Answer:
@Test public void testTaskOwnedStateStream() throws Exception { final List<String> state = Arrays.asList("Flopsy", "Mopsy", "Cotton Tail", "Peter"); final FsCheckpointStorage storage = new FsCheckpointStorage( Path.fromLocalFile(tmp.newFolder()), null, new JobID(), 10); final StreamStateHandle stateHandle; try (CheckpointStateOutputStream stream = storage.createTaskOwnedStateStream()) { assertTrue(stream instanceof FsCheckpointStateOutputStream); new ObjectOutputStream(stream).writeObject(state); stateHandle = stream.closeAndGetHandle(); } FileStateHandle fileStateHandle = (FileStateHandle) stateHandle; String parentDirName = fileStateHandle.getFilePath().getParent().getName(); assertEquals(FsCheckpointStorage.CHECKPOINT_TASK_OWNED_STATE_DIR, parentDirName); try (ObjectInputStream in = new ObjectInputStream(stateHandle.openInputStream())) { assertEquals(state, in.readObject()); } }