docs: 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;
  			}
  		}
  	}