@Override
    public void putJobGraph(JobGraph jobGraph) throws Exception {
        checkNotNull(jobGraph, "Job graph");

        final JobID jobID = jobGraph.getJobID();
        final String name = jobGraphStoreUtil.jobIDToName(jobID);

        LOG.debug("Adding job graph {} to {}.", jobID, jobGraphStateHandleStore);

        boolean success = false;

        while (!success) {
            synchronized (lock) {
                verifyIsRunning();

                final R currentVersion = jobGraphStateHandleStore.exists(name);

                if (!currentVersion.isExisting()) {
                    try {
                        jobGraphStateHandleStore.addAndLock(name, jobGraph);

                        addedJobGraphs.add(jobID);

                        success = true;
                    } catch (StateHandleStore.AlreadyExistException ignored) {
                        LOG.warn("{} already exists in {}.", jobGraph, jobGraphStateHandleStore);
                    }
                } else if (addedJobGraphs.contains(jobID)) {
                    try {
                        jobGraphStateHandleStore.replace(name, currentVersion, jobGraph);
                        LOG.info("Updated {} in {}.", jobGraph, getClass().getSimpleName());

                        success = true;
                    } catch (StateHandleStore.NotExistException ignored) {
                        LOG.warn("{} does not exists in {}.", jobGraph, jobGraphStateHandleStore);
                    }
                } else {
                    throw new IllegalStateException(
                            "Trying to update a graph you didn't "
                                    + "#getAllSubmittedJobGraphs() or #putJobGraph() yourself before.");
                }
            }
        }

        LOG.info("Added {} to {}.", jobGraph, jobGraphStateHandleStore);
    }
@Test
    public void testOnRemovedJobGraphShouldOnlyProcessKnownJobGraphs() throws Exception {
        final TestingStateHandleStore<JobGraph> stateHandleStore =
                builder.setAddFunction((ignore, state) -> jobGraphStorageHelper.store(state))
                        .build();
        final JobGraphStore jobGraphStore = createAndStartJobGraphStore(stateHandleStore);
        jobGraphStore.putJobGraph(testingJobGraph);

        // Unknown job
        testingJobGraphStoreWatcher.removeJobGraph(JobID.generate());
        // Known job
        testingJobGraphStoreWatcher.removeJobGraph(testingJobGraph.getJobID());
        assertThat(testingJobGraphListener.getRemovedJobGraphs().size(), is(1));
        assertThat(
                testingJobGraphListener.getRemovedJobGraphs(),
                contains(testingJobGraph.getJobID()));
    }