@Override
    public CompletableFuture<Void> closeAsync() {
        synchronized (lock) {
            if (isShutdown) {
                return terminationFuture;
            } else {
                isShutdown = true;
                final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3);
                final Time gracePeriod = Time.seconds(1L);

                if (metricQueryServiceRpcService != null) {
                    final CompletableFuture<Void> metricQueryServiceRpcServiceTerminationFuture =
                            metricQueryServiceRpcService.closeAsync();
                    terminationFutures.add(metricQueryServiceRpcServiceTerminationFuture);
                }

                Throwable throwable = null;
                for (ReporterAndSettings reporterAndSettings : reporters) {
                    try {
                        reporterAndSettings.getReporter().close();
                    } catch (Throwable t) {
                        throwable = ExceptionUtils.firstOrSuppressed(t, throwable);
                    }
                }
                reporters.clear();

                if (throwable != null) {
                    terminationFutures.add(
                            FutureUtils.completedExceptionally(
                                    new FlinkException(
                                            "Could not shut down the metric reporters properly.",
                                            throwable)));
                }

                final CompletableFuture<Void> reporterExecutorShutdownFuture =
                        ExecutorUtils.nonBlockingShutdown(
                                gracePeriod.toMilliseconds(),
                                TimeUnit.MILLISECONDS,
                                reporterScheduledExecutor);
                terminationFutures.add(reporterExecutorShutdownFuture);

                final CompletableFuture<Void> viewUpdaterExecutorShutdownFuture =
                        ExecutorUtils.nonBlockingShutdown(
                                gracePeriod.toMilliseconds(),
                                TimeUnit.MILLISECONDS,
                                viewUpdaterScheduledExecutor);

                terminationFutures.add(viewUpdaterExecutorShutdownFuture);

                FutureUtils.completeAll(terminationFutures)
                        .whenComplete(
                                (Void ignored, Throwable error) -> {
                                    if (error != null) {
                                        terminationFuture.completeExceptionally(error);
                                    } else {
                                        terminationFuture.complete(null);
                                    }
                                });

                return terminationFuture;
            }
        }
    }
@Test
    void testReporterScheduling() throws Exception {
        MetricConfig config = new MetricConfig();
        config.setProperty("arg1", "hello");
        config.setProperty(MetricOptions.REPORTER_INTERVAL.key(), "50 MILLISECONDS");

        final ReportCountingReporter reporter = new ReportCountingReporter();

        MetricRegistryImpl registry =
                new MetricRegistryImpl(
                        MetricRegistryTestUtils.defaultMetricRegistryConfiguration(),
                        Collections.singletonList(
                                ReporterSetup.forReporter("test", config, reporter)));

        long start = System.currentTimeMillis();

        // only start counting from now on
        reporter.resetCount();

        for (int x = 0; x < 10; x++) {
            Thread.sleep(100);
            int reportCount = reporter.getReportCount();
            long curT = System.currentTimeMillis();
            /**
             * Within a given time-frame T only T/500 reports may be triggered due to the interval
             * between reports. This value however does not not take the first triggered report into
             * account (=> +1). Furthermore we have to account for the mis-alignment between reports
             * being triggered and our time measurement (=> +1); for T=200 a total of 4-6 reports
             * may have been triggered depending on whether the end of the interval for the first
             * reports ends before or after T=50.
             */
            long maxAllowedReports = (curT - start) / 50 + 2;
            assertThat(maxAllowedReports)
                    .as("Too many reports were triggered.")
                    .isGreaterThanOrEqualTo(reportCount);
        }
        assertThat(reporter.getReportCount()).as("No report was triggered.").isGreaterThan(0);

        registry.closeAsync().get();
    }