@Override
    public List<StreamStateHandle> duplicate(List<StreamStateHandle> stateHandles)
            throws IOException {

        final List<CopyRequest> requests = new ArrayList<>();
        for (StreamStateHandle handle : stateHandles) {
            if (!(handle instanceof FileStateHandle)) {
                throw new IllegalArgumentException("We can duplicate only FileStateHandles.");
            }
            final Path srcPath = ((FileStateHandle) handle).getFilePath();
            requests.add(CopyRequest.of(srcPath, getNewDstPath(srcPath.getName())));
        }
        fs.copyFiles(requests, new CloseableRegistry());

        return IntStream.range(0, stateHandles.size())
                .mapToObj(
                        idx -> {
                            final StreamStateHandle originalHandle = stateHandles.get(idx);
                            final Path dst = requests.get(idx).getDestination();
                            if (originalHandle instanceof RelativeFileStateHandle) {
                                return new RelativeFileStateHandle(
                                        dst, dst.getName(), originalHandle.getStateSize());
                            } else {
                                return new FileStateHandle(dst, originalHandle.getStateSize());
                            }
                        })
                .collect(Collectors.toList());
    }
@Test
    void testDuplicating() throws IOException {
        final TestDuplicatingFileSystem fs = new TestDuplicatingFileSystem();
        final FsCheckpointStateToolset stateToolset =
                new FsCheckpointStateToolset(new Path("test-path"), fs);

        final List<StreamStateHandle> duplicated =
                stateToolset.duplicate(
                        Arrays.asList(
                                new FileStateHandle(new Path("old-test-path", "test-file1"), 0),
                                new FileStateHandle(new Path("old-test-path", "test-file2"), 0),
                                new RelativeFileStateHandle(
                                        new Path("old-test-path", "test-file3"), "test-file3", 0)));

        assertThat(duplicated)
                .containsExactly(
                        new FileStateHandle(new Path("test-path", "test-file1"), 0),
                        new FileStateHandle(new Path("test-path", "test-file2"), 0),
                        new RelativeFileStateHandle(
                                new Path("test-path", "test-file3"), "test-file3", 0));
    }