@Override
    public Collection<JobID> getJobIds() throws Exception {
        LOG.debug("Retrieving all stored job ids from {}.", jobGraphStateHandleStore);

        final Collection<String> names;
        try {
            names = jobGraphStateHandleStore.getAllHandles();
        } catch (Exception e) {
            throw new Exception(
                    "Failed to retrieve all job ids from " + jobGraphStateHandleStore + ".", e);
        }

        final List<JobID> jobIds = new ArrayList<>(names.size());

        for (String name : names) {
            try {
                jobIds.add(jobGraphStoreUtil.nameToJobID(name));
            } catch (Exception exception) {
                LOG.warn(
                        "Could not parse job id from {}. This indicates a malformed name.",
                        name,
                        exception);
            }
        }

        LOG.info("Retrieved job ids {} from {}", jobIds, jobGraphStateHandleStore);

        return jobIds;
    }
@Test
    public void testGetJobIds() throws Exception {
        final List<JobID> existingJobIds = Arrays.asList(new JobID(0, 0), new JobID(0, 1));
        final TestingStateHandleStore<JobGraph> stateHandleStore =
                builder.setGetAllHandlesSupplier(
                                () ->
                                        existingJobIds.stream()
                                                .map(AbstractID::toString)
                                                .collect(Collectors.toList()))
                        .build();

        final JobGraphStore jobGraphStore = createAndStartJobGraphStore(stateHandleStore);
        final Collection<JobID> jobIds = jobGraphStore.getJobIds();
        assertThat(jobIds, contains(existingJobIds.toArray()));
    }