@Override
    public CompletableFuture<Void> closeAsync() {
        ShutdownHookUtil.removeShutdownHook(shutDownHook, getClass().getSimpleName(), LOG);

        return shutDownAsync(
                        ApplicationStatus.UNKNOWN,
                        ShutdownBehaviour.PROCESS_FAILURE,
                        "Cluster entrypoint has been closed externally.",
                        false)
                .thenAccept(ignored -> {});
    }
@Test
    public void testCloseAsyncShouldNotDeregisterApp() throws Exception {
        final CompletableFuture<Void> deregisterFuture = new CompletableFuture<>();
        final TestingResourceManagerFactory testingResourceManagerFactory =
                new TestingResourceManagerFactory.Builder()
                        .setInternalDeregisterApplicationConsumer(
                                (ignored1, ignored2, ignore3) -> deregisterFuture.complete(null))
                        .build();
        final TestingEntryPoint testingEntryPoint =
                new TestingEntryPoint.Builder()
                        .setConfiguration(flinkConfig)
                        .setResourceManagerFactory(testingResourceManagerFactory)
                        .build();

        final CompletableFuture<ApplicationStatus> appStatusFuture =
                startClusterEntrypoint(testingEntryPoint);

        testingEntryPoint.closeAsync();
        assertThat(
                appStatusFuture.get(TIMEOUT_MS, TimeUnit.MILLISECONDS),
                is(ApplicationStatus.UNKNOWN));
        assertThat(deregisterFuture.isDone(), is(false));
    }