public static void shutdownExecutorServiceQuietly(ExecutorService executorService,
                                                      long timeout, TimeUnit timeUnit) {
        if (executorService == null) {
            return;
        }
        executorService.shutdown(); // Disable new tasks from being submitted
        try {
            // Wait a while for existing tasks to terminate
            if (!executorService.awaitTermination(timeout, timeUnit)) {
                executorService.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if (!executorService.awaitTermination(timeout, timeUnit)) {
                    log.error("Executor {} did not terminate in time", executorService);
                }
            }
        } catch (InterruptedException e) {
            // (Re-)Cancel if current thread also interrupted
            executorService.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }
Shuts down an executor service in two phases, first by calling shutdown to reject incoming tasks,
and then calling shutdownNow, if necessary, to cancel any lingering tasks.
After the timeout/on interrupt, the service is forcefully closed.
@param executorService The service to shut down.
@param timeout The timeout of the shutdown.
@param timeUnit The time unit of the shutdown timeout.
clients/src/main/java/org/apache/kafka/common/utils/ThreadUtils.java