@Nullable
    @Override
    public JobGraph recoverJobGraph(JobID jobId) throws Exception {
        checkNotNull(jobId, "Job ID");

        LOG.debug("Recovering job graph {} from {}.", jobId, jobGraphStateHandleStore);

        final String name = jobGraphStoreUtil.jobIDToName(jobId);

        synchronized (lock) {
            verifyIsRunning();

            boolean success = false;

            RetrievableStateHandle<JobGraph> jobGraphRetrievableStateHandle;

            try {
                try {
                    jobGraphRetrievableStateHandle = jobGraphStateHandleStore.getAndLock(name);
                } catch (StateHandleStore.NotExistException ignored) {
                    success = true;
                    return null;
                } catch (Exception e) {
                    throw new FlinkException(
                            "Could not retrieve the submitted job graph state handle "
                                    + "for "
                                    + name
                                    + " from the submitted job graph store.",
                            e);
                }

                JobGraph jobGraph;
                try {
                    jobGraph = jobGraphRetrievableStateHandle.retrieveState();
                } catch (ClassNotFoundException cnfe) {
                    throw new FlinkException(
                            "Could not retrieve submitted JobGraph from state handle under "
                                    + name
                                    + ". This indicates that you are trying to recover from state written by an "
                                    + "older Flink version which is not compatible. Try cleaning the state handle store.",
                            cnfe);
                } catch (IOException ioe) {
                    throw new FlinkException(
                            "Could not retrieve submitted JobGraph from state handle under "
                                    + name
                                    + ". This indicates that the retrieved state handle is broken. Try cleaning the state handle "
                                    + "store.",
                            ioe);
                }

                addedJobGraphs.add(jobGraph.getJobID());

                LOG.info("Recovered {}.", jobGraph);

                success = true;
                return jobGraph;
            } finally {
                if (!success) {
                    jobGraphStateHandleStore.release(name);
                }
            }
        }
    }
@Test
    public void testRecoverJobGraphFailedShouldReleaseHandle() throws Exception {
        final CompletableFuture<String> releaseFuture = new CompletableFuture<>();
        final FlinkException testException = new FlinkException("Test exception.");
        final TestingStateHandleStore<JobGraph> stateHandleStore =
                builder.setGetFunction(
                                ignore -> {
                                    throw testException;
                                })
                        .setReleaseConsumer(releaseFuture::complete)
                        .build();

        final JobGraphStore jobGraphStore = createAndStartJobGraphStore(stateHandleStore);

        try {
            jobGraphStore.recoverJobGraph(testingJobGraph.getJobID());
            fail(
                    "recoverJobGraph should fail when there is exception in getting the state handle.");
        } catch (Exception ex) {
            assertThat(ex, FlinkMatchers.containsCause(testException));
            String actual = releaseFuture.get(timeout, TimeUnit.MILLISECONDS);
            assertThat(actual, is(testingJobGraph.getJobID().toString()));
        }
    }