@Override
    protected VertexFlameGraph handleRequest(
            HandlerRequest<EmptyRequestBody> request, AccessExecutionJobVertex jobVertex)
            throws RestHandlerException {

        @Nullable Integer subtaskIndex = getSubtaskIndex(request, jobVertex);
        if (isTerminated(jobVertex, subtaskIndex)) {
            return VertexFlameGraph.terminated();
        }

        final Optional<VertexThreadInfoStats> threadInfoSample;
        if (subtaskIndex == null) {
            threadInfoSample =
                    threadInfoOperatorTracker.getJobVertexStats(
                            request.getPathParameter(JobIDPathParameter.class), jobVertex);
        } else {
            threadInfoSample =
                    threadInfoOperatorTracker.getExecutionVertexStats(
                            request.getPathParameter(JobIDPathParameter.class),
                            jobVertex,
                            subtaskIndex);
        }

        final FlameGraphTypeQueryParameter.Type flameGraphType = getFlameGraphType(request);

        final Optional<VertexFlameGraph> operatorFlameGraph;

        switch (flameGraphType) {
            case FULL:
                operatorFlameGraph =
                        threadInfoSample.map(VertexFlameGraphFactory::createFullFlameGraphFrom);
                break;
            case ON_CPU:
                operatorFlameGraph =
                        threadInfoSample.map(VertexFlameGraphFactory::createOnCpuFlameGraph);
                break;
            case OFF_CPU:
                operatorFlameGraph =
                        threadInfoSample.map(VertexFlameGraphFactory::createOffCpuFlameGraph);
                break;
            default:
                throw new RestHandlerException(
                        "Unknown Flame Graph type " + flameGraphType + '.',
                        HttpResponseStatus.BAD_REQUEST);
        }

        return operatorFlameGraph.orElse(VertexFlameGraph.waiting());
    }
@Test
    void testHandleFinishedJobVertex() throws Exception {
        final ArchivedExecutionJobVertex archivedExecutionJobVertex =
                new ArchivedExecutionJobVertex(
                        new ArchivedExecutionVertex[] {
                            generateExecutionVertex(0, ExecutionState.FINISHED),
                            generateExecutionVertex(1, ExecutionState.FINISHED)
                        },
                        JOB_VERTEX_ID,
                        "test",
                        2,
                        2,
                        new SlotSharingGroup(),
                        ResourceProfile.UNKNOWN,
                        new StringifiedAccumulatorResult[0]);

        HandlerRequest<EmptyRequestBody> request = generateJobVertexFlameGraphParameters(null);
        VertexFlameGraph jobVertexFlameGraph =
                handler.handleRequest(request, archivedExecutionJobVertex);
        assertThat(jobVertexFlameGraph.getEndTime())
                .isEqualTo(VertexFlameGraph.terminated().getEndTime());
    }