@Override
    public void shutDown() {
        synchronized (lock) {
            if (!shutDown) {
                jobVertexStatsCache.invalidateAll();
                pendingJobVertexStats.clear();

                executionVertexStatsCache.invalidateAll();
                pendingExecutionVertexStats.clear();

                shutDown = true;
            }
        }
    }
@TestTemplate
    void testShutDown() throws Exception {
        final VertexThreadInfoTracker tracker = createThreadInfoTracker();
        doInitialJobVertexRequestAndVerifyResult(tracker);

        // shutdown directly
        tracker.shutDown();

        // verify that the previous cached result is invalid and trigger another request
        assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
        assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, 0)).isNotPresent();
        // verify no response after shutdown
        assertThat(tracker.getJobVertexStats(JOB_ID, executionJobVertex)).isNotPresent();
        assertThat(tracker.getExecutionVertexStats(JOB_ID, executionJobVertex, 0)).isNotPresent();
    }