class MetricRegistryImpl implements MetricRegistry {

	public CompletableFuture<Void> shutdown() {
		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.stopService();
					terminationFutures.add(metricQueryServiceRpcServiceTerminationFuture);
				}

				Throwable throwable = null;
				for (MetricReporter reporter : reporters) {
					try {
						reporter.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> executorShutdownFuture = ExecutorUtils.nonBlockingShutdown(
					gracePeriod.toMilliseconds(),
					TimeUnit.MILLISECONDS,
					executor);

				terminationFutures.add(executorShutdownFuture);

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

				return terminationFuture;
			}
		}
	}

	public  MetricRegistryImpl(MetricRegistryConfiguration config);
	public  MetricRegistryImpl(MetricRegistryConfiguration config, Collection<ReporterSetup> reporterConfigurations);

}