@Override
    public CheckpointStateOutputStream createTaskOwnedStateStream() {
        // as the comment of CheckpointStorageWorkerView#createTaskOwnedStateStream said we may
        // change into shared state,
        // so we use CheckpointedStateScope.SHARED here.
        return new FsCheckpointStateOutputStream(
                taskOwnedStateDirectory, fileSystem, writeBufferSize, fileSizeThreshold);
    }
@Test
    void testTaskOwnedStateStream() throws Exception {
        final List<String> state = Arrays.asList("Flopsy", "Mopsy", "Cotton Tail", "Peter");

        // we chose a small size threshold here to force the state to disk
        final FsCheckpointStorageAccess storage =
                new FsCheckpointStorageAccess(
                        Path.fromLocalFile(TempDirUtils.newFolder(tmp)),
                        null,
                        true,
                        new JobID(),
                        10,
                        WRITE_BUFFER_SIZE);

        final StreamStateHandle stateHandle;

        try (CheckpointStateOutputStream stream = storage.createTaskOwnedStateStream()) {
            assertThat(stream).isInstanceOf(FsCheckpointStateOutputStream.class);

            new ObjectOutputStream(stream).writeObject(state);
            stateHandle = stream.closeAndGetHandle();
        }

        // the state must have gone to disk
        FileStateHandle fileStateHandle = (FileStateHandle) stateHandle;

        // check that the state is in the correct directory
        String parentDirName = fileStateHandle.getFilePath().getParent().getName();
        assertThat(parentDirName)
                .isEqualTo(FsCheckpointStorageAccess.CHECKPOINT_TASK_OWNED_STATE_DIR);

        // validate the contents
        try (ObjectInputStream in = new ObjectInputStream(stateHandle.openInputStream())) {
            assertThat(in.readObject()).isEqualTo(state);
        }
    }