@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 testConfigurableDelimiterForReportersInGroup() throws Exception {
        String name = "C";
        MetricConfig config1 = new MetricConfig();
        config1.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "_");

        MetricConfig config2 = new MetricConfig();
        config2.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "-");

        MetricConfig config3 = new MetricConfig();
        config3.setProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key(), "AA");

        Configuration config = new Configuration();
        config.set(MetricOptions.SCOPE_NAMING_TM, "A.B");

        List<ReporterSetup> reporterConfigurations =
                Arrays.asList(
                        ReporterSetup.forReporter(
                                "test1", config1, new CollectingMetricsReporter()),
                        ReporterSetup.forReporter(
                                "test2", config2, new CollectingMetricsReporter()),
                        ReporterSetup.forReporter(
                                "test3", config3, new CollectingMetricsReporter()),
                        ReporterSetup.forReporter("test4", new CollectingMetricsReporter()));
        MetricRegistryImpl registry =
                new MetricRegistryImpl(
                        MetricRegistryTestUtils.fromConfiguration(config), reporterConfigurations);

        TaskManagerMetricGroup group =
                TaskManagerMetricGroup.createTaskManagerMetricGroup(
                        registry, "host", new ResourceID("id"));
        group.counter(name);
        group.close();
        registry.closeAsync().get();

        for (ReporterSetup cfg : reporterConfigurations) {
            String delimiter =
                    cfg.getConfiguration()
                            .getProperty(MetricOptions.REPORTER_SCOPE_DELIMITER.key());
            if (delimiter == null || delimiter.equals("AA")) {
                // test3 reporter: 'AA' - not correct
                // for test4 reporter use global delimiter
                delimiter = String.valueOf(GLOBAL_DEFAULT_DELIMITER);
            }
            String expected =
                    (config.get(MetricOptions.SCOPE_NAMING_TM) + ".C").replaceAll("\\.", delimiter);
            CollectingMetricsReporter reporter = (CollectingMetricsReporter) cfg.getReporter();

            for (MetricGroupAndName groupAndName :
                    Arrays.asList(reporter.findAdded(name), reporter.findRemoved(name))) {
                assertThat(groupAndName.group.getMetricIdentifier(name)).isEqualTo(expected);
                assertThat(groupAndName.group.getMetricIdentifier(name, reporter))
                        .isEqualTo(expected);
            }
        }
    }