@Override
    public void stop() throws Exception {
        if (!running) {
            return;
        }
        running = false;

        LOG.info("Stopping ZooKeeperJobGraphStoreWatcher ");
        pathCache.close();
    }
@Test
    void testJobGraphAddedAndRemovedShouldNotifyGraphStoreListener() throws Exception {
        try (final CuratorFrameworkWithUnhandledErrorListener curatorFrameworkWrapper =
                ZooKeeperUtils.startCuratorFramework(
                        configuration, NoOpFatalErrorHandler.INSTANCE)) {
            final CuratorFramework client = curatorFrameworkWrapper.asCuratorFramework();
            final JobGraphStoreWatcher jobGraphStoreWatcher =
                    createAndStartJobGraphStoreWatcher(client);

            final ZooKeeperStateHandleStore<JobGraph> stateHandleStore =
                    createStateHandleStore(client);

            final JobGraph jobGraph = JobGraphTestUtils.emptyJobGraph();
            final JobID jobID = jobGraph.getJobID();
            stateHandleStore.addAndLock("/" + jobID, jobGraph);

            CommonTestUtils.waitUntilCondition(
                    () -> testingJobGraphListener.getAddedJobGraphs().size() > 0);

            assertThat(testingJobGraphListener.getAddedJobGraphs()).containsExactly(jobID);

            stateHandleStore.releaseAndTryRemove("/" + jobID);

            CommonTestUtils.waitUntilCondition(
                    () -> testingJobGraphListener.getRemovedJobGraphs().size() > 0);
            assertThat(testingJobGraphListener.getRemovedJobGraphs()).containsExactly(jobID);

            jobGraphStoreWatcher.stop();
        }
    }