@Override
    public void put(ExecutionGraphInfo executionGraphInfo) throws IOException {
        final JobID jobId = executionGraphInfo.getJobId();

        final ArchivedExecutionGraph archivedExecutionGraph =
                executionGraphInfo.getArchivedExecutionGraph();
        final JobStatus jobStatus = archivedExecutionGraph.getState();
        final String jobName = archivedExecutionGraph.getJobName();

        Preconditions.checkArgument(
                jobStatus.isTerminalState(),
                "The job "
                        + jobName
                        + '('
                        + jobId
                        + ") is not in a terminal state. Instead it is in state "
                        + jobStatus
                        + '.');

        switch (jobStatus) {
            case FINISHED:
                numFinishedJobs++;
                break;
            case CANCELED:
                numCanceledJobs++;
                break;
            case FAILED:
                numFailedJobs++;
                break;
            case SUSPENDED:
                break;
            default:
                throw new IllegalStateException(
                        "The job "
                                + jobName
                                + '('
                                + jobId
                                + ") should have been in a known terminal state. "
                                + "Instead it was in state "
                                + jobStatus
                                + '.');
        }

        // write the ArchivedExecutionGraph to disk
        storeExecutionGraphInfo(executionGraphInfo);

        final JobDetails detailsForJob = JobDetails.createDetailsForJob(archivedExecutionGraph);

        jobDetailsCache.put(jobId, detailsForJob);
        executionGraphInfoCache.put(jobId, executionGraphInfo);
    }
@Test
    public void testPut() throws IOException {
        assertPutJobGraphWithStatus(JobStatus.FINISHED);
    }