@Override
    public PipelineState getLatestCheckpointByJobIdAndPipelineId(String jobId, String pipelineId)
            throws CheckpointStorageException {

        String parentPath = getStorageParentDirectory() + jobId;
        Collection<File> fileList = new ArrayList<>();
        try {
            fileList = FileUtils.listFiles(new File(parentPath), FILE_EXTENSIONS, false);
        } catch (Exception e) {
            if (!(e.getCause() instanceof NoSuchFileException)) {
                throw new CheckpointStorageException(ExceptionUtils.getMessage(e));
            }
        }
        if (fileList.isEmpty()) {
            log.info("No checkpoint found for job, job id is: " + jobId);
            return null;
        }
        List<String> fileNames = fileList.stream().map(File::getName).collect(Collectors.toList());

        String latestFileName =
                getLatestCheckpointFileNameByJobIdAndPipelineId(fileNames, pipelineId);

        AtomicReference<PipelineState> latestFile = new AtomicReference<>(null);
        fileList.forEach(
                file -> {
                    String fileName = file.getName();
                    if (fileName.equals(latestFileName)) {
                        try {
                            byte[] data = FileUtils.readFileToByteArray(file);
                            latestFile.set(deserializeCheckPointData(data));
                        } catch (IOException e) {
                            log.error(
                                    "read checkpoint data from file " + file.getAbsolutePath(), e);
                        }
                    }
                });

        if (latestFile.get() == null) {
            log.info(
                    "No checkpoint found for this job, the job id is: "
                            + jobId
                            + ", pipeline id is: "
                            + pipelineId);
            return null;
        }
        return latestFile.get();
    }
@Test
    public void testGetLatestCheckpointByJobIdAndPipelineId() throws CheckpointStorageException {
        PipelineState state = STORAGE.getLatestCheckpointByJobIdAndPipelineId(JOB_ID, "1");
        Assertions.assertEquals(2, state.getCheckpointId());
    }