Repository: apache/flink
File: flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java (Java)

Code before:
                executor);

        this.metricFetcher = Preconditions.checkNotNull(metricFetcher);
    }

    @Override
    protected JobDetailsInfo handleRequest(
            HandlerRequest<EmptyRequestBody> request, AccessExecutionGraph executionGraph)
            throws RestHandlerException {
        metricFetcher.update();
        return createJobDetailsInfo(executionGraph, metricFetcher.getMetricStore().getJobs());
    }

    @Override
    public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
            throws IOException {
        ResponseBody json = createJobDetailsInfo(graph, null);
        String path =
                getMessageHeaders()
                        .getTargetRestEndpointURL()
                        .replace(':' + JobIDPathParameter.KEY, graph.getJobID().toString());
        return Collections.singleton(new ArchivedJson(path, json));
    }

    private static JobDetailsInfo createJobDetailsInfo(
            AccessExecutionGraph executionGraph, @Nullable MetricStore.MetricStoreJobs jobs) {
        final long now = System.currentTimeMillis();
        final long startTime = executionGraph.getStatusTimestamp(JobStatus.INITIALIZING);
        final long endTime =
                executionGraph.getState().isGloballyTerminalState()
                        ? executionGraph.getStatusTimestamp(executionGraph.getState())
                        : -1L;
        final long duration = (endTime > 0L ? endTime : now) - startTime;

        final Map<JobStatus, Long> timestamps =
                CollectionUtil.newHashMapWithExpectedSize(JobStatus.values().length);

        for (JobStatus jobStatus : JobStatus.values()) {
            timestamps.put(jobStatus, executionGraph.getStatusTimestamp(jobStatus));
        }

        Collection<JobDetailsInfo.JobVertexDetailsInfo> jobVertexInfos =
                new ArrayList<>(executionGraph.getAllVertices().size());
        int[] jobVerticesPerState = new int[ExecutionState.values().length];

        for (AccessExecutionJobVertex accessExecutionJobVertex :
                executionGraph.getVerticesTopologically()) {
            final JobDetailsInfo.JobVertexDetailsInfo vertexDetailsInfo =
                    createJobVertexDetailsInfo(
                            accessExecutionJobVertex, now, executionGraph.getJobID(), jobs);


Review comment (suggestion):
The name `jobs` is too general and may be confusing. The `MetricsStoreJobs` is actually a snapshot of the metrics of all jobs. We may name it `jobMetrics`.

Code after:
                executor);

        this.metricFetcher = Preconditions.checkNotNull(metricFetcher);
    }

    @Override
    protected JobDetailsInfo handleRequest(
            HandlerRequest<EmptyRequestBody> request, AccessExecutionGraph executionGraph)
            throws RestHandlerException {
        metricFetcher.update();
        return createJobDetailsInfo(executionGraph, metricFetcher.getMetricStore().getJobs());
    }

    @Override
    public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
            throws IOException {
        ResponseBody json = createJobDetailsInfo(graph, null);
        String path =
                getMessageHeaders()
                        .getTargetRestEndpointURL()
                        .replace(':' + JobIDPathParameter.KEY, graph.getJobID().toString());
        return Collections.singleton(new ArchivedJson(path, json));
    }

    private static JobDetailsInfo createJobDetailsInfo(
            AccessExecutionGraph executionGraph,
            @Nullable MetricStore.JobMetricStoreSnapshot jobMetrics) {
        final long now = System.currentTimeMillis();
        final long startTime = executionGraph.getStatusTimestamp(JobStatus.INITIALIZING);
        final long endTime =
                executionGraph.getState().isGloballyTerminalState()
                        ? executionGraph.getStatusTimestamp(executionGraph.getState())
                        : -1L;
        final long duration = (endTime > 0L ? endTime : now) - startTime;

        final Map<JobStatus, Long> timestamps =
                CollectionUtil.newHashMapWithExpectedSize(JobStatus.values().length);

        for (JobStatus jobStatus : JobStatus.values()) {
            timestamps.put(jobStatus, executionGraph.getStatusTimestamp(jobStatus));
        }

        Collection<JobDetailsInfo.JobVertexDetailsInfo> jobVertexInfos =
                new ArrayList<>(executionGraph.getAllVertices().size());
        int[] jobVerticesPerState = new int[ExecutionState.values().length];

        for (AccessExecutionJobVertex accessExecutionJobVertex :
                executionGraph.getVerticesTopologically()) {
            final JobDetailsInfo.JobVertexDetailsInfo vertexDetailsInfo =
                    createJobVertexDetailsInfo(
                            accessExecutionJobVertex, now, executionGraph.getJobID(), jobMetrics);