@Override
    public CompletableFuture<Void> closeAsync() {
        return stopApplication(
                ApplicationStatus.CANCELED, "DispatcherResourceManagerComponent has been closed.");
    }
@Test
    public void unexpectedResourceManagerTermination_ifNotRunning_doesNotFailFatally() {
        final TestingFatalErrorHandler fatalErrorHandler = new TestingFatalErrorHandler();
        final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
        final TestingResourceManagerService resourceManagerService =
                TestingResourceManagerService.newBuilder()
                        .setTerminationFuture(terminationFuture)
                        .withManualTerminationFutureCompletion()
                        .build();

        final DispatcherResourceManagerComponent dispatcherResourceManagerComponent =
                createDispatcherResourceManagerComponent(fatalErrorHandler, resourceManagerService);

        dispatcherResourceManagerComponent.closeAsync();

        final FlinkException expectedException = new FlinkException("Expected test exception.");
        terminationFuture.completeExceptionally(expectedException);

        final CompletableFuture<Throwable> errorFuture = fatalErrorHandler.getErrorFuture();
        assertThat(errorFuture, willNotComplete(Duration.ofMillis(10L)));
    }