Copilot commented on code in PR #29379: URL: https://github.com/apache/flink/pull/29379#discussion_r4279744602
########## flink-container/src/main/java/org/apache/flink/container/entrypoint/ApplicationModeMiniCluster.java: ########## @@ -0,0 +1,226 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.container.entrypoint; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.client.deployment.application.ApplicationDispatcherLeaderProcessFactoryFactory; +import org.apache.flink.client.program.PackagedProgram; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.failure.FailureEnricher; +import org.apache.flink.core.plugin.PluginManager; +import org.apache.flink.core.security.FlinkSecurityManager; +import org.apache.flink.runtime.blob.BlobServer; +import org.apache.flink.runtime.clusterframework.ApplicationStatus; +import org.apache.flink.runtime.dispatcher.SessionDispatcherFactory; +import org.apache.flink.runtime.dispatcher.runner.DefaultDispatcherRunnerFactory; +import org.apache.flink.runtime.entrypoint.ClusterEntrypoint; +import org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory; +import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponent; +import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory; +import org.apache.flink.runtime.externalresource.ExternalResourceInfoProvider; +import org.apache.flink.runtime.externalresource.ExternalResourceUtils; +import org.apache.flink.runtime.failure.FailureEnricherUtils; +import org.apache.flink.runtime.heartbeat.HeartbeatServices; +import org.apache.flink.runtime.metrics.MetricRegistry; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; +import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory; +import org.apache.flink.runtime.rest.ApplicationRestEndpointFactory; +import org.apache.flink.runtime.rpc.FatalErrorHandler; +import org.apache.flink.runtime.rpc.RpcService; +import org.apache.flink.runtime.rpc.RpcSystem; +import org.apache.flink.runtime.security.token.DelegationTokenManager; +import org.apache.flink.runtime.webmonitor.retriever.MetricQueryServiceRetriever; +import org.apache.flink.util.ExceptionUtils; +import org.apache.flink.util.concurrent.FutureUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.util.Collection; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.stream.Collectors; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * A {@link MiniCluster} that runs a single application in the same way as a distributed application + * cluster. The dispatcher is built with the user's program and starts the application itself once + * it has leadership, recovering from high availability metadata where available. + */ +@Internal +final class ApplicationModeMiniCluster extends MiniCluster { + + private static final Logger LOG = LoggerFactory.getLogger(ApplicationModeMiniCluster.class); + + private final Configuration configuration; + private final PackagedProgram program; + @Nullable private final PluginManager pluginManager; + + private final CompletableFuture<ApplicationStatus> applicationShutDownFuture = + new CompletableFuture<>(); + private final CompletableFuture<Void> closeRequestedFuture = new CompletableFuture<>(); + private final Collection<DispatcherResourceManagerComponent> + dispatcherResourceManagerComponents = new CopyOnWriteArrayList<>(); + + ApplicationModeMiniCluster( + final MiniClusterConfiguration miniClusterConfiguration, + final PackagedProgram program) { + super(miniClusterConfiguration); + this.configuration = new Configuration(miniClusterConfiguration.getConfiguration()); + this.program = checkNotNull(program); + this.pluginManager = miniClusterConfiguration.getPluginManager(); + } + + CompletableFuture<ApplicationStatus> getApplicationShutDownFuture() { + return applicationShutDownFuture; + } + + CompletableFuture<Void> getCloseRequestedFuture() { + return closeRequestedFuture; + } + + @Override + protected DispatcherResourceManagerComponentFactory + createDispatcherResourceManagerComponentFactory() { + return new DefaultDispatcherResourceManagerComponentFactory( + new DefaultDispatcherRunnerFactory( + ApplicationDispatcherLeaderProcessFactoryFactory.create( + configuration, SessionDispatcherFactory.INSTANCE, program)), + StandaloneResourceManagerFactory.getInstance(), + ApplicationRestEndpointFactory.INSTANCE); + } + + @Override + protected Collection<? extends DispatcherResourceManagerComponent> + createDispatcherResourceManagerComponents( + Configuration configuration, + RpcServiceFactory rpcServiceFactory, + BlobServer blobServer, + HeartbeatServices heartbeatServices, + DelegationTokenManager delegationTokenManager, + MetricRegistry metricRegistry, + MetricQueryServiceRetriever metricQueryServiceRetriever, + FatalErrorHandler fatalErrorHandler) + throws Exception { + final Collection<? extends DispatcherResourceManagerComponent> components = + super.createDispatcherResourceManagerComponents( + configuration, + rpcServiceFactory, + blobServer, + heartbeatServices, + delegationTokenManager, + metricRegistry, + metricQueryServiceRetriever, + fatalErrorHandler); + for (DispatcherResourceManagerComponent component : components) { + FutureUtils.forward(component.getShutDownFuture(), applicationShutDownFuture); + } + dispatcherResourceManagerComponents.addAll(components); + return components; + } + + @Override + protected FatalErrorHandler createJobManagerFatalErrorHandler() { + return createProcessFatalErrorHandler("JobManager"); + } + + @Override + protected FatalErrorHandler createTaskManagerFatalErrorHandler(int index) { + return createProcessFatalErrorHandler("TaskManager #" + index); + } + + /** + * Nothing replaces a JobManager or TaskManager that fails, so the whole process has to fail + * instead, like their processes do in a distributed cluster. That leaves the restart to the + * process supervisor, with high availability recovering the job. + */ + private FatalErrorHandler createProcessFatalErrorHandler(String component) { + return exception -> { + if (closeRequestedFuture.isDone()) { + LOG.debug("Ignoring {} error during shutdown.", component, exception); + return; + } + LOG.error("{} failed. Shutting the MiniCluster down.", component, exception); + if (ExceptionUtils.isJvmFatalOrOutOfMemoryError(exception)) { + LOG.error( + "Terminating the process with exit code {}.", + ClusterEntrypoint.RUNTIME_FAILURE_RETURN_CODE); + FlinkSecurityManager.forceProcessExit( + ClusterEntrypoint.RUNTIME_FAILURE_RETURN_CODE); + } else { + closeAsync(); + } + }; + } + + @Override + protected Collection<FailureEnricher> createFailureEnrichers(Configuration configuration) { + return FailureEnricherUtils.getFailureEnrichers(configuration); + } + + @Override + protected ExternalResourceInfoProvider createExternalResourceInfoProvider( + Configuration configuration) { + return ExternalResourceUtils.createStaticExternalResourceInfoProviderFromConfig( + configuration, pluginManager); + } + + /** Sizes the RPC thread pool from the {@code pekko.fork-join-executor} options. */ + @Override + protected RpcService createLocalRpcService(Configuration configuration, RpcSystem rpcSystem) + throws Exception { + return rpcSystem.localServiceBuilder(configuration).createAndStart(); + } + + @Override + public CompletableFuture<Void> closeAsync() { + closeRequestedFuture.complete(null); + final boolean cleanupHighAvailabilityData = isDispatcherShutDown(); + return FutureUtils.composeAfterwards( + FutureUtils.completeAll( + // stop the dispatcher before the MiniCluster stops TaskManagers so it + // suspends jobs first + dispatcherResourceManagerComponents.stream() + .map(DispatcherResourceManagerComponent::stopProcess) + .collect(Collectors.toList())), + // cluster high availability data is only removed if the dispatcher has shut down + // (which happens when the application finishes is set if + // execution.shutdown-on-application-finish is set) + // Otherwise, any data is kept to allow a restarted process to recover. + cleanupHighAvailabilityData + ? super::closeAsync + : super::closeAsyncWithoutCleaningHighAvailabilityData); Review Comment: Failed startup can deadlock here when dispatcher shutdown completes asynchronously. `MiniCluster.start()` calls blocking `close()` from its catch block while holding `lock` (MiniCluster.java:499–502). This deferred superclass close then runs on the component-shutdown thread and needs that same lock in `closeInternal()`. Startup waits for shutdown, while shutdown waits for startup to release the lock, so the process never reaches startup failure exit code 1. Move failed-start cleanup outside the startup monitor, or provide a startup-safe close path, and test failure after dispatcher component creation. ########## flink-container/src/main/java/org/apache/flink/container/entrypoint/MiniClusterJobEntryPoint.java: ########## @@ -0,0 +1,213 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.container.entrypoint; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.client.deployment.application.ApplicationClusterEntryPoint; +import org.apache.flink.client.program.PackagedProgram; +import org.apache.flink.configuration.ClusterOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JMXServerOptions; +import org.apache.flink.configuration.RpcOptions; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.plugin.PluginManager; +import org.apache.flink.core.plugin.PluginUtils; +import org.apache.flink.core.security.FlinkSecurityManager; +import org.apache.flink.management.jmx.JMXService; +import org.apache.flink.runtime.clusterframework.ApplicationStatus; +import org.apache.flink.runtime.entrypoint.ClusterEntrypoint; +import org.apache.flink.runtime.entrypoint.ClusterEntrypointUtils; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; +import org.apache.flink.runtime.security.contexts.SecurityContext; +import org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader; +import org.apache.flink.runtime.util.EnvironmentInformation; +import org.apache.flink.runtime.util.JvmShutdownSafeguard; +import org.apache.flink.runtime.util.SignalHandler; +import org.apache.flink.util.ShutdownHookUtil; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Entry point that runs a single application in a {@link MiniCluster}. It takes the same command + * line parameters as {@link StandaloneApplicationClusterEntryPoint}, and the MiniCluster is built + * with the same application-mode dispatcher. + */ +@Internal +public final class MiniClusterJobEntryPoint { + + private static final Logger LOG = LoggerFactory.getLogger(MiniClusterJobEntryPoint.class); + + private static final Duration SHUTDOWN_TIMEOUT = Duration.ofSeconds(30); + + /** The safeguard's default, as used by the other cluster entrypoints. */ + private static final Duration JVM_SHUTDOWN_SAFEGUARD_DELAY = Duration.ofSeconds(5); + + /** Shorter than the safeguard delay, so that a slow close is logged before the JVM halts. */ + private static final Duration SHUTDOWN_HOOK_TIMEOUT = Duration.ofSeconds(4); + + private static final AtomicBoolean STOPPING = new AtomicBoolean(); + + /** Whether the JVM is exiting for another reason, which decides its exit code. */ + private static volatile boolean closedByShutdownHook; + + public static void main(String[] args) { + EnvironmentInformation.logEnvironmentInfo( + LOG, MiniClusterJobEntryPoint.class.getSimpleName(), args); + SignalHandler.register(LOG); + JvmShutdownSafeguard.installAsShutdownHook(LOG, JVM_SHUTDOWN_SAFEGUARD_DELAY.toMillis()); + + final StandaloneApplicationClusterConfiguration clusterConfiguration = + ClusterEntrypointUtils.parseParametersOrExit( + args, + new StandaloneApplicationClusterConfigurationParserFactory(), + MiniClusterJobEntryPoint.class); + + int exitCode; + try { + final Configuration configuration = + ApplicationClusterEntryPointUtils.loadConfiguration(clusterConfiguration); + FlinkSecurityManager.setFromConfiguration(configuration); + final PluginManager pluginManager = + PluginUtils.createPluginManagerFromRootFolder(configuration); + FileSystem.initialize(configuration, pluginManager); + StateChangelogStorageLoader.initialize(pluginManager); + + final SecurityContext securityContext = + ClusterEntrypoint.installSecurityContext(configuration); + ClusterEntrypointUtils.configureUncaughtExceptionHandler(configuration); + exitCode = + securityContext.runSecured( + () -> + runApplication( + clusterConfiguration, configuration, pluginManager)); + } catch (Throwable t) { + LOG.error("Could not run the application on a MiniCluster.", t); + exitCode = ClusterEntrypoint.STARTUP_FAILURE_RETURN_CODE; + } + if (closedByShutdownHook) { + return; + } + LOG.info( + "Terminating {} with exit code {}.", + MiniClusterJobEntryPoint.class.getSimpleName(), + exitCode); + System.exit(exitCode); + } + + private static int runApplication( + final StandaloneApplicationClusterConfiguration clusterConfiguration, + final Configuration configuration, + final PluginManager pluginManager) + throws Exception { + final PackagedProgram program = + ApplicationClusterEntryPointUtils.getPackagedProgram( + clusterConfiguration, configuration); + final ApplicationModeMiniCluster miniCluster; + try { + ApplicationClusterEntryPoint.configureExecution(configuration, program); + + miniCluster = + new ApplicationModeMiniCluster( + buildMiniClusterConfiguration(configuration, pluginManager), program); + final String serviceName = MiniClusterJobEntryPoint.class.getSimpleName(); + ShutdownHookUtil.addShutdownHookThread( + new Thread( + () -> { + if (STOPPING.compareAndSet(false, true)) { + closedByShutdownHook = true; + closeQuietly(miniCluster, SHUTDOWN_HOOK_TIMEOUT); + program.close(); + } + }, + serviceName + " shutdown hook"), + serviceName, + LOG); + JMXService.startInstance(configuration.get(JMXServerOptions.JMX_SERVER_PORT)); + miniCluster.start(); + } catch (Throwable t) { + if (STOPPING.compareAndSet(false, true)) { + program.close(); + } + throw t; + } + + CompletableFuture.anyOf( + miniCluster.getApplicationShutDownFuture(), + miniCluster.getCloseRequestedFuture()) + .handle((ignored, throwable) -> null) + .join(); + + // only one of this and the shutdown hook closes the MiniCluster, so that a close that + // times out here isn't waited for again during JVM shutdown + if (!STOPPING.compareAndSet(false, true)) { + return 0; + } + + closeQuietly(miniCluster, SHUTDOWN_TIMEOUT); + program.close(); Review Comment: If shutdown fails or times out after a successful application, `closeQuietly()` only logs the error and this path still returns exit code 0. For example, an exception while removing HA metadata is lost. `ClusterEntrypoint` returns runtime failure code 2 when termination fails. Propagate the shutdown outcome here, keeping best-effort handling only for the shutdown hook, so supervisors can distinguish successful termination from failed cleanup. ########## flink-container/src/main/java/org/apache/flink/container/entrypoint/MiniClusterJobEntryPoint.java: ########## @@ -0,0 +1,213 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.container.entrypoint; + +import org.apache.flink.annotation.Internal; +import org.apache.flink.client.deployment.application.ApplicationClusterEntryPoint; +import org.apache.flink.client.program.PackagedProgram; +import org.apache.flink.configuration.ClusterOptions; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JMXServerOptions; +import org.apache.flink.configuration.RpcOptions; +import org.apache.flink.configuration.TaskManagerOptions; +import org.apache.flink.core.fs.FileSystem; +import org.apache.flink.core.plugin.PluginManager; +import org.apache.flink.core.plugin.PluginUtils; +import org.apache.flink.core.security.FlinkSecurityManager; +import org.apache.flink.management.jmx.JMXService; +import org.apache.flink.runtime.clusterframework.ApplicationStatus; +import org.apache.flink.runtime.entrypoint.ClusterEntrypoint; +import org.apache.flink.runtime.entrypoint.ClusterEntrypointUtils; +import org.apache.flink.runtime.minicluster.MiniCluster; +import org.apache.flink.runtime.minicluster.MiniClusterConfiguration; +import org.apache.flink.runtime.security.contexts.SecurityContext; +import org.apache.flink.runtime.state.changelog.StateChangelogStorageLoader; +import org.apache.flink.runtime.util.EnvironmentInformation; +import org.apache.flink.runtime.util.JvmShutdownSafeguard; +import org.apache.flink.runtime.util.SignalHandler; +import org.apache.flink.util.ShutdownHookUtil; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Entry point that runs a single application in a {@link MiniCluster}. It takes the same command + * line parameters as {@link StandaloneApplicationClusterEntryPoint}, and the MiniCluster is built + * with the same application-mode dispatcher. + */ +@Internal +public final class MiniClusterJobEntryPoint { + + private static final Logger LOG = LoggerFactory.getLogger(MiniClusterJobEntryPoint.class); + + private static final Duration SHUTDOWN_TIMEOUT = Duration.ofSeconds(30); + + /** The safeguard's default, as used by the other cluster entrypoints. */ + private static final Duration JVM_SHUTDOWN_SAFEGUARD_DELAY = Duration.ofSeconds(5); + + /** Shorter than the safeguard delay, so that a slow close is logged before the JVM halts. */ + private static final Duration SHUTDOWN_HOOK_TIMEOUT = Duration.ofSeconds(4); + + private static final AtomicBoolean STOPPING = new AtomicBoolean(); + + /** Whether the JVM is exiting for another reason, which decides its exit code. */ + private static volatile boolean closedByShutdownHook; + + public static void main(String[] args) { + EnvironmentInformation.logEnvironmentInfo( + LOG, MiniClusterJobEntryPoint.class.getSimpleName(), args); + SignalHandler.register(LOG); + JvmShutdownSafeguard.installAsShutdownHook(LOG, JVM_SHUTDOWN_SAFEGUARD_DELAY.toMillis()); + + final StandaloneApplicationClusterConfiguration clusterConfiguration = + ClusterEntrypointUtils.parseParametersOrExit( + args, + new StandaloneApplicationClusterConfigurationParserFactory(), + MiniClusterJobEntryPoint.class); + + int exitCode; + try { + final Configuration configuration = + ApplicationClusterEntryPointUtils.loadConfiguration(clusterConfiguration); + FlinkSecurityManager.setFromConfiguration(configuration); + final PluginManager pluginManager = + PluginUtils.createPluginManagerFromRootFolder(configuration); + FileSystem.initialize(configuration, pluginManager); + StateChangelogStorageLoader.initialize(pluginManager); + + final SecurityContext securityContext = + ClusterEntrypoint.installSecurityContext(configuration); + ClusterEntrypointUtils.configureUncaughtExceptionHandler(configuration); + exitCode = + securityContext.runSecured( + () -> + runApplication( + clusterConfiguration, configuration, pluginManager)); + } catch (Throwable t) { + LOG.error("Could not run the application on a MiniCluster.", t); + exitCode = ClusterEntrypoint.STARTUP_FAILURE_RETURN_CODE; + } + if (closedByShutdownHook) { + return; + } + LOG.info( + "Terminating {} with exit code {}.", + MiniClusterJobEntryPoint.class.getSimpleName(), + exitCode); + System.exit(exitCode); + } + + private static int runApplication( + final StandaloneApplicationClusterConfiguration clusterConfiguration, + final Configuration configuration, + final PluginManager pluginManager) + throws Exception { + final PackagedProgram program = + ApplicationClusterEntryPointUtils.getPackagedProgram( + clusterConfiguration, configuration); + final ApplicationModeMiniCluster miniCluster; + try { + ApplicationClusterEntryPoint.configureExecution(configuration, program); + + miniCluster = + new ApplicationModeMiniCluster( + buildMiniClusterConfiguration(configuration, pluginManager), program); + final String serviceName = MiniClusterJobEntryPoint.class.getSimpleName(); + ShutdownHookUtil.addShutdownHookThread( + new Thread( + () -> { + if (STOPPING.compareAndSet(false, true)) { + closedByShutdownHook = true; + closeQuietly(miniCluster, SHUTDOWN_HOOK_TIMEOUT); + program.close(); + } + }, + serviceName + " shutdown hook"), + serviceName, + LOG); + JMXService.startInstance(configuration.get(JMXServerOptions.JMX_SERVER_PORT)); + miniCluster.start(); + } catch (Throwable t) { + if (STOPPING.compareAndSet(false, true)) { + program.close(); + } + throw t; + } + + CompletableFuture.anyOf( + miniCluster.getApplicationShutDownFuture(), + miniCluster.getCloseRequestedFuture()) + .handle((ignored, throwable) -> null) + .join(); + + // only one of this and the shutdown hook closes the MiniCluster, so that a close that + // times out here isn't waited for again during JVM shutdown + if (!STOPPING.compareAndSet(false, true)) { + return 0; + } + + closeQuietly(miniCluster, SHUTDOWN_TIMEOUT); + program.close(); + + final CompletableFuture<ApplicationStatus> applicationShutDown = + miniCluster.getApplicationShutDownFuture(); + if (!applicationShutDown.isDone() || applicationShutDown.isCompletedExceptionally()) { + LOG.error("MiniCluster shut down before the application finished."); + return ClusterEntrypoint.RUNTIME_FAILURE_RETURN_CODE; + } + final ApplicationStatus applicationStatus = applicationShutDown.join(); + LOG.info("Application finished with status {}.", applicationStatus); + return applicationStatus.processExitCode(); + } + + private static MiniClusterConfiguration buildMiniClusterConfiguration( + final Configuration configuration, final PluginManager pluginManager) { + final Configuration clusterConfiguration = new Configuration(configuration); + if (!clusterConfiguration.contains(RpcOptions.ASK_TIMEOUT_DURATION)) { + clusterConfiguration.set( + RpcOptions.ASK_TIMEOUT_DURATION, + RpcOptions.ASK_TIMEOUT_DURATION.defaultValue()); + } + if (!clusterConfiguration.contains(ClusterOptions.CLUSTER_IO_EXECUTOR_POOL_SIZE)) { + clusterConfiguration.set( + ClusterOptions.CLUSTER_IO_EXECUTOR_POOL_SIZE, + ClusterEntrypointUtils.getPoolSize(clusterConfiguration)); + } + return new MiniClusterConfiguration.Builder() + .setConfiguration(clusterConfiguration) + .setNumSlotsPerTaskManager(configuration.get(TaskManagerOptions.NUM_TASK_SLOTS)) + .setPluginManager(pluginManager) + .build(); + } + + private static void closeQuietly(final MiniCluster miniCluster, final Duration timeout) { + try { + miniCluster.closeAsync().get(timeout.toMillis(), TimeUnit.MILLISECONDS); Review Comment: The timeout bounds only `get()`, not the preceding `closeAsync()` invocation. Although the shutdown hook is now registered before startup, SIGTERM before dispatcher creation can make this call synchronously enter `MiniCluster.closeInternal()` and wait for the lock held by `start()`. If startup remains blocked, the four-second timeout cannot fire; the five-second safeguard instead halts with exit code 239 on Unix, without the intended timeout log. Run the close invocation on a dedicated daemon worker and bound the entire operation, preserving the installed security context. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
