@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 testReporterIntervalParsingErrorFallsBackToDefaultValue() throws Exception {
        MetricConfig config = new MetricConfig();
        // in a prior implementation the time amount was applied even if the time unit was invalid
        // in this case this would imply using 1 SECOND as the interval (seconds is the default)
        config.setProperty(MetricOptions.REPORTER_INTERVAL.key(), "1 UNICORN");

        final ManuallyTriggeredScheduledExecutorService manuallyTriggeredScheduledExecutorService =
                new ManuallyTriggeredScheduledExecutorService();

        MetricRegistryImpl registry =
                new MetricRegistryImpl(
                        MetricRegistryTestUtils.defaultMetricRegistryConfiguration(),
                        Collections.singletonList(
                                ReporterSetup.forReporter(
                                        "test", config, new ReportCountingReporter())),
                        manuallyTriggeredScheduledExecutorService);
        try {
            Collection<ScheduledFuture<?>> scheduledTasks =
                    manuallyTriggeredScheduledExecutorService.getActiveScheduledTasks();
            ScheduledFuture<?> reportTask = Iterators.getOnlyElement(scheduledTasks.iterator());
            assertThat(reportTask.getDelay(TimeUnit.SECONDS))
                    .isEqualTo(MetricOptions.REPORTER_INTERVAL.defaultValue().getSeconds());
        } finally {
            registry.closeAsync().get();
        }
    }