@Override
    public List<PipelineState> getAllCheckpoints(String jobId) throws CheckpointStorageException {
        File filePath = new File(getStorageParentDirectory() + jobId);
        if (!filePath.exists()) {
            return new ArrayList<>();
        }

        Collection<File> fileList;
        try {
            fileList = FileUtils.listFiles(filePath, FILE_EXTENSIONS, true);
        } catch (Exception e) {
            throw new CheckpointStorageException(
                    "Failed to get all checkpoints for job " + jobId, e);
        }
        if (fileList.isEmpty()) {
            log.info("No checkpoint found for this job, the job id is: " + jobId);
            return new ArrayList<>();
        }
        List<PipelineState> states = new ArrayList<>();
        fileList.forEach(
                file -> {
                    try {
                        byte[] data = FileUtils.readFileToByteArray(file);
                        states.add(deserializeCheckPointData(data));
                    } catch (IOException e) {
                        log.error(
                                "Failed to read checkpoint data from file "
                                        + file.getAbsolutePath(),
                                e);
                    }
                });
        return states;
    }
@Test
    public void testGetAllCheckpoints() throws CheckpointStorageException {

        List<PipelineState> pipelineStates = STORAGE.getAllCheckpoints(JOB_ID);
        Assertions.assertEquals(3, pipelineStates.size());
    }