[FLINK-29330] Provide better logs of MiniCluster shutdown procedure

• Type: Technical Debt
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Fix Version/s: 2.0.0
• Component/s: Tests
• Assignee: Chesnay Schepler
• Reporter: Chesnay Schepler
• Created: 19/Sep/22 09:13
• Updated: 10/Jul/24 06:57

Description

I recently ran into an issue where the shutdown of a MiniCluster timed out. The logs weren't helpful at all and I had to go in and check every asynchronously component for whether that component was the cause.

The main issues were that various components don't log anything at all, or that when they did it wasn't clear who owned that component.

I'd like to add a util that makes it easier for us log the start/stop of a shutdown procedure,

public class ShutdownLog {
    /**
     * Logs the beginning and end of the shutdown procedure for the given component.
     *
     * <p>This method accepts a {@link Supplier} instead of a {@link CompletableFuture} because the
     * latter usually required implies the shutdown to already have begun.
     *
     * @param log Logger of owning component
     * @param component component that will be shut down
     * @param shutdownTrigger component shutdown trigger
     * @return termination future of the component
     */
    public static <C> CompletableFuture<C> logShutdown(
            Logger log, String component, Supplier<CompletableFuture<C>> shutdownTrigger) {
        log.debug("Starting shutdown of {}.", component);
        return FutureUtils.logCompletion(log, "shutdown of " + component, shutdownTrigger.get());
    }
}

public class FutureUtils {
    public static <T> CompletableFuture<T> logCompletion(
            Logger log, String action, CompletableFuture<T> future) {
        future.handle(
                (t, throwable) -> {
                    if (throwable == null) {
                        log.debug("Completed {}.", action);
                    } else {
                        log.debug("Failed {}.", action, throwable);
                    }
                    return null;
                });
        return future;
    }
...

and extend the AutoCloseableAsync interface for an easy opt-in and customized logging:

    default CompletableFuture<Void> closeAsync(Logger log) {
        return ShutdownLog.logShutdown(log, getClass().getSimpleName(), this::closeAsync);
    }

MiniCluster example usages:

-terminationFutures.add(dispatcherResourceManagerComponent.closeAsync())
+terminationFutures.add(dispatcherResourceManagerComponent.closeAsync(LOG))

-return ExecutorUtils.nonBlockingShutdown(
-        executorShutdownTimeoutMillis, TimeUnit.MILLISECONDS, ioExecutor);
+return ShutdownLog.logShutdown(
+        LOG,
+        "ioExecutor",
+        () ->
+                ExecutorUtils.nonBlockingShutdown(
+                        executorShutdownTimeoutMillis,
+                        TimeUnit.MILLISECONDS,
+                        ioExecutor));

mapohl I'm interested what you think about this.