This is an automated email from the ASF dual-hosted git repository. trohrmann pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit e2943006aec1febcfd1c33eef1b1f44367019a98 Author: wangyang0918 <danrtsey...@alibaba-inc.com> AuthorDate: Thu Apr 1 17:24:50 2021 +0800 [FLINK-21008][coordination] Register ClusterEntrypoint#closeAsync as shutdown hook for the cleanup This closes #15396. --- .../runtime/entrypoint/ClusterEntrypoint.java | 43 ++- .../DispatcherResourceManagerComponent.java | 28 +- .../runtime/entrypoint/ClusterEntrypointTest.java | 387 ++++++++++++++++++++- .../TestingHighAvailabilityServices.java | 18 +- .../TestingHighAvailabilityServicesBuilder.java | 21 ++ .../flink/runtime/testutils/TestJvmProcess.java | 19 + .../testutils/TestingClusterEntrypointProcess.java | 136 ++++++++ .../flink/runtime/util/BlockingShutdownTest.java | 11 +- .../recovery/ProcessFailureCancelingITCase.java | 2 +- 9 files changed, 625 insertions(+), 40 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java index aab1c7a..69b090d 100755 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java @@ -165,7 +165,7 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro shutDownHook = ShutdownHookUtil.addShutdownHook( - this::cleanupDirectories, getClass().getSimpleName(), LOG); + () -> this.closeAsync().join(), getClass().getSimpleName(), LOG); } public CompletableFuture<ApplicationStatus> getTerminationFuture() { @@ -198,6 +198,7 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro // clean up any partial state shutDownAsync( ApplicationStatus.FAILED, + ShutdownBehaviour.STOP_APPLICATION, ExceptionUtils.stringifyException(strippedThrowable), false) .get( @@ -266,13 +267,18 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro if (throwable != null) { shutDownAsync( ApplicationStatus.UNKNOWN, + ShutdownBehaviour.STOP_APPLICATION, ExceptionUtils.stringifyException(throwable), false); } else { // This is the general shutdown path. If a separate more // specific shutdown was // already triggered, this will do nothing - shutDownAsync(applicationStatus, null, true); + shutDownAsync( + applicationStatus, + ShutdownBehaviour.STOP_APPLICATION, + null, + true); } }); } @@ -363,10 +369,13 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro @Override public CompletableFuture<Void> closeAsync() { + ShutdownHookUtil.removeShutdownHook(shutDownHook, getClass().getSimpleName(), LOG); + return shutDownAsync( ApplicationStatus.UNKNOWN, + ShutdownBehaviour.STOP_PROCESS, "Cluster entrypoint has been closed externally.", - true) + false) .thenAccept(ignored -> {}); } @@ -465,6 +474,7 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro private CompletableFuture<ApplicationStatus> shutDownAsync( ApplicationStatus applicationStatus, + ShutdownBehaviour shutdownBehaviour, @Nullable String diagnostics, boolean cleanupHaData) { if (isShutDown.compareAndSet(false, true)) { @@ -475,7 +485,7 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro diagnostics); final CompletableFuture<Void> shutDownApplicationFuture = - closeClusterComponent(applicationStatus, diagnostics); + closeClusterComponent(applicationStatus, shutdownBehaviour, diagnostics); final CompletableFuture<Void> serviceShutdownFuture = FutureUtils.composeAfterwards( @@ -498,19 +508,27 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro } /** - * Deregister the Flink application from the resource management system by signalling the {@link - * ResourceManager}. + * Close cluster components and deregister the Flink application from the resource management + * system by signalling the {@link ResourceManager}. * * @param applicationStatus to terminate the application with + * @param shutdownBehaviour shutdown behaviour * @param diagnostics additional information about the shut down, can be {@code null} * @return Future which is completed once the shut down */ private CompletableFuture<Void> closeClusterComponent( - ApplicationStatus applicationStatus, @Nullable String diagnostics) { + ApplicationStatus applicationStatus, + ShutdownBehaviour shutdownBehaviour, + @Nullable String diagnostics) { synchronized (lock) { if (clusterComponent != null) { - return clusterComponent.deregisterApplicationAndClose( - applicationStatus, diagnostics); + switch (shutdownBehaviour) { + case STOP_APPLICATION: + return clusterComponent.stopApplication(applicationStatus, diagnostics); + case STOP_PROCESS: + default: + return clusterComponent.stopProcess(); + } } else { return CompletableFuture.completedFuture(null); } @@ -523,8 +541,6 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro * @throws IOException if the temporary directories could not be cleaned up */ protected void cleanupDirectories() throws IOException { - ShutdownHookUtil.removeShutdownHook(shutDownHook, getClass().getSimpleName(), LOG); - final String webTmpDir = configuration.getString(WebOptions.TMP_DIR); FileUtils.deleteDirectory(new File(webTmpDir)); @@ -615,4 +631,9 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro /** Directly stops after the job has finished. */ DETACHED } + + private enum ShutdownBehaviour { + STOP_APPLICATION, + STOP_PROCESS, + } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java index bb4ba04..477ac9a 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/component/DispatcherResourceManagerComponent.java @@ -41,6 +41,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Supplier; /** * Component which starts a {@link Dispatcher}, {@link ResourceManager} and {@link @@ -113,23 +114,36 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync { /** * Deregister the Flink application from the resource management system by signalling the {@link - * ResourceManager}. + * ResourceManager} and also stop the process. * * @param applicationStatus to terminate the application with * @param diagnostics additional information about the shut down, can be {@code null} * @return Future which is completed once the shut down */ - public CompletableFuture<Void> deregisterApplicationAndClose( + public CompletableFuture<Void> stopApplication( final ApplicationStatus applicationStatus, final @Nullable String diagnostics) { + return internalShutdown(() -> deregisterApplication(applicationStatus, diagnostics)); + } + + /** + * Close the web monitor and cluster components. This method will not deregister the Flink + * application from the resource management and only stop the process. + * + * @return Future which is completed once the shut down + */ + public CompletableFuture<Void> stopProcess() { + return internalShutdown(FutureUtils::completedVoidFuture); + } + private CompletableFuture<Void> internalShutdown( + final Supplier<CompletableFuture<?>> additionalShutdownAction) { if (isRunning.compareAndSet(true, false)) { - final CompletableFuture<Void> closeWebMonitorAndDeregisterAppFuture = + final CompletableFuture<Void> closeWebMonitorAndAdditionalShutdownActionFuture = FutureUtils.composeAfterwards( - webMonitorEndpoint.closeAsync(), - () -> deregisterApplication(applicationStatus, diagnostics)); + webMonitorEndpoint.closeAsync(), additionalShutdownAction); return FutureUtils.composeAfterwards( - closeWebMonitorAndDeregisterAppFuture, this::closeAsyncInternal); + closeWebMonitorAndAdditionalShutdownActionFuture, this::closeAsyncInternal); } else { return terminationFuture; } @@ -188,7 +202,7 @@ public class DispatcherResourceManagerComponent implements AutoCloseableAsync { @Override public CompletableFuture<Void> closeAsync() { - return deregisterApplicationAndClose( + return stopApplication( ApplicationStatus.CANCELED, "DispatcherResourceManagerComponent has been closed."); } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterEntrypointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterEntrypointTest.java index 12b5b97..db497bd 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterEntrypointTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/entrypoint/ClusterEntrypointTest.java @@ -22,50 +22,417 @@ import org.apache.flink.configuration.Configuration; import org.apache.flink.configuration.IllegalConfigurationException; import org.apache.flink.configuration.JobManagerOptions; import org.apache.flink.configuration.SchedulerExecutionMode; +import org.apache.flink.runtime.clusterframework.ApplicationStatus; +import org.apache.flink.runtime.clusterframework.types.ResourceID; +import org.apache.flink.runtime.concurrent.FutureUtils; import org.apache.flink.runtime.concurrent.ScheduledExecutor; import org.apache.flink.runtime.dispatcher.ExecutionGraphInfoStore; +import org.apache.flink.runtime.dispatcher.MemoryExecutionGraphInfoStore; +import org.apache.flink.runtime.dispatcher.PartialDispatcherServices; +import org.apache.flink.runtime.dispatcher.runner.DispatcherRunner; +import org.apache.flink.runtime.dispatcher.runner.DispatcherRunnerFactory; +import org.apache.flink.runtime.dispatcher.runner.TestingDispatcherRunner; +import org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory; import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory; +import org.apache.flink.runtime.heartbeat.HeartbeatServices; +import org.apache.flink.runtime.highavailability.HighAvailabilityServices; +import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServicesBuilder; +import org.apache.flink.runtime.io.network.partition.NoOpResourceManagerPartitionTracker; +import org.apache.flink.runtime.jobmanager.JobGraphStoreFactory; +import org.apache.flink.runtime.leaderelection.LeaderElectionService; +import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup; +import org.apache.flink.runtime.resourcemanager.ArbitraryWorkerResourceSpecFactory; +import org.apache.flink.runtime.resourcemanager.JobLeaderIdService; +import org.apache.flink.runtime.resourcemanager.ResourceManager; +import org.apache.flink.runtime.resourcemanager.ResourceManagerFactory; +import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServices; +import org.apache.flink.runtime.resourcemanager.ResourceManagerRuntimeServicesConfiguration; +import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory; +import org.apache.flink.runtime.resourcemanager.TestingJobLeaderIdService; +import org.apache.flink.runtime.resourcemanager.TestingResourceManager; +import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager; +import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerBuilder; +import org.apache.flink.runtime.rest.SessionRestEndpointFactory; +import org.apache.flink.runtime.rpc.FatalErrorHandler; +import org.apache.flink.runtime.rpc.RpcService; +import org.apache.flink.runtime.testutils.TestJvmProcess; +import org.apache.flink.runtime.testutils.TestingClusterEntrypointProcess; +import org.apache.flink.runtime.util.SignalHandler; +import org.apache.flink.testutils.executor.TestExecutorResource; +import org.apache.flink.util.ConfigurationException; +import org.apache.flink.util.OperatingSystem; +import org.apache.flink.util.TestLogger; +import org.junit.Before; +import org.junit.ClassRule; import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import javax.annotation.Nullable; + +import java.io.File; import java.io.IOException; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.function.BiConsumer; +import static org.hamcrest.CoreMatchers.is; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import static org.junit.Assume.assumeTrue; /** Tests for the {@link ClusterEntrypoint}. */ -public class ClusterEntrypointTest { +public class ClusterEntrypointTest extends TestLogger { + + private static final long TIMEOUT_MS = 3000; + + private Configuration flinkConfig; + + @ClassRule + public static final TestExecutorResource<?> TEST_EXECUTOR_RESOURCE = + new TestExecutorResource<>(Executors::newSingleThreadExecutor); + + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + + @Before + public void before() { + flinkConfig = new Configuration(); + } @Test(expected = IllegalConfigurationException.class) public void testStandaloneSessionClusterEntrypointDeniedInReactiveMode() { - Configuration configuration = new Configuration(); - configuration.set(JobManagerOptions.SCHEDULER_MODE, SchedulerExecutionMode.REACTIVE); - new MockEntryPoint(configuration); + flinkConfig.set(JobManagerOptions.SCHEDULER_MODE, SchedulerExecutionMode.REACTIVE); + new TestingEntryPoint.Builder().setConfiguration(flinkConfig).build(); fail("Entrypoint initialization is supposed to fail"); } - private static class MockEntryPoint extends ClusterEntrypoint { + @Test + public void testCloseAsyncShouldNotCleanUpHAData() throws Exception { + final CompletableFuture<Void> closeFuture = new CompletableFuture<>(); + final CompletableFuture<Void> closeAndCleanupAllDataFuture = new CompletableFuture<>(); + final HighAvailabilityServices testingHaService = + new TestingHighAvailabilityServicesBuilder() + .setCloseFuture(closeFuture) + .setCloseAndCleanupAllDataFuture(closeAndCleanupAllDataFuture) + .build(); + final TestingEntryPoint testingEntryPoint = + new TestingEntryPoint.Builder() + .setConfiguration(flinkConfig) + .setHighAvailabilityServices(testingHaService) + .build(); + + final CompletableFuture<ApplicationStatus> appStatusFuture = + startClusterEntrypoint(testingEntryPoint); + + testingEntryPoint.closeAsync(); + assertThat( + appStatusFuture.get(TIMEOUT_MS, TimeUnit.MILLISECONDS), + is(ApplicationStatus.UNKNOWN)); + assertThat(closeFuture.isDone(), is(true)); + assertThat(closeAndCleanupAllDataFuture.isDone(), is(false)); + } + + @Test + public void testCloseAsyncShouldNotDeregisterApp() throws Exception { + final CompletableFuture<Void> deregisterFuture = new CompletableFuture<>(); + final TestingResourceManagerFactory testingResourceManagerFactory = + new TestingResourceManagerFactory.Builder() + .setInternalDeregisterApplicationConsumer( + (ignored1, ignored2) -> 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)); + } + + @Test + public void testClusterFinishedNormallyShouldDeregisterAppAndCleanupHAData() throws Exception { + final CompletableFuture<Void> deregisterFuture = new CompletableFuture<>(); + final CompletableFuture<Void> closeAndCleanupAllDataFuture = new CompletableFuture<>(); + final CompletableFuture<ApplicationStatus> dispatcherShutDownFuture = + new CompletableFuture<>(); + + final HighAvailabilityServices testingHaService = + new TestingHighAvailabilityServicesBuilder() + .setCloseAndCleanupAllDataFuture(closeAndCleanupAllDataFuture) + .build(); + final TestingResourceManagerFactory testingResourceManagerFactory = + new TestingResourceManagerFactory.Builder() + .setInternalDeregisterApplicationConsumer( + (ignored1, ignored2) -> deregisterFuture.complete(null)) + .build(); + final TestingDispatcherRunnerFactory testingDispatcherRunnerFactory = + new TestingDispatcherRunnerFactory.Builder() + .setShutDownFuture(dispatcherShutDownFuture) + .build(); + + final TestingEntryPoint testingEntryPoint = + new TestingEntryPoint.Builder() + .setConfiguration(flinkConfig) + .setResourceManagerFactory(testingResourceManagerFactory) + .setDispatcherRunnerFactory(testingDispatcherRunnerFactory) + .setHighAvailabilityServices(testingHaService) + .build(); + + final CompletableFuture<ApplicationStatus> appStatusFuture = + startClusterEntrypoint(testingEntryPoint); + + dispatcherShutDownFuture.complete(ApplicationStatus.SUCCEEDED); - protected MockEntryPoint(Configuration configuration) { + assertThat( + appStatusFuture.get(TIMEOUT_MS, TimeUnit.MILLISECONDS), + is(ApplicationStatus.SUCCEEDED)); + assertThat(deregisterFuture.isDone(), is(true)); + assertThat(closeAndCleanupAllDataFuture.isDone(), is(true)); + } + + @Test + public void testCloseAsyncShouldBeExecutedInShutdownHook() throws Exception { + // This test only works on Linux and Mac OS because we are sending SIGTERM to a + // JAVA process via "kill {pid}" + assumeTrue(OperatingSystem.isLinux() || OperatingSystem.isMac()); + final File markerFile = new File(TEMPORARY_FOLDER.getRoot(), UUID.randomUUID() + ".marker"); + final TestingClusterEntrypointProcess clusterEntrypointProcess = + new TestingClusterEntrypointProcess(markerFile); + try { + clusterEntrypointProcess.startProcess(); + final long pid = clusterEntrypointProcess.getProcessId(); + assertTrue("Cannot determine process ID", pid != -1); + + // wait for the marker file to appear, which means the process is up properly + TestJvmProcess.waitForMarkerFile(markerFile, 30000); + + TestJvmProcess.killProcessWithSigTerm(pid); + + final boolean exited = + clusterEntrypointProcess.waitFor(TIMEOUT_MS, TimeUnit.MILLISECONDS); + assertThat( + String.format("Process %s does not exit within %s ms", pid, TIMEOUT_MS), + exited, + is(true)); + assertThat( + "markerFile should be deleted in closeAsync shutdownHook", + markerFile.exists(), + is(false)); + } finally { + clusterEntrypointProcess.destroy(); + } + } + + private CompletableFuture<ApplicationStatus> startClusterEntrypoint( + TestingEntryPoint testingEntryPoint) throws Exception { + testingEntryPoint.startCluster(); + return FutureUtils.supplyAsync( + () -> testingEntryPoint.getTerminationFuture().get(), + TEST_EXECUTOR_RESOURCE.getExecutor()); + } + + private static class TestingEntryPoint extends ClusterEntrypoint { + + private final HighAvailabilityServices haService; + + private final ResourceManagerFactory<ResourceID> resourceManagerFactory; + + private final DispatcherRunnerFactory dispatcherRunnerFactory; + + private TestingEntryPoint( + Configuration configuration, + HighAvailabilityServices haService, + ResourceManagerFactory<ResourceID> resourceManagerFactory, + DispatcherRunnerFactory dispatcherRunnerFactory) { super(configuration); + SignalHandler.register(LOG); + this.haService = haService; + this.resourceManagerFactory = resourceManagerFactory; + this.dispatcherRunnerFactory = dispatcherRunnerFactory; } @Override protected DispatcherResourceManagerComponentFactory createDispatcherResourceManagerComponentFactory(Configuration configuration) throws IOException { - throw new UnsupportedOperationException("Not needed for this test"); + return new DefaultDispatcherResourceManagerComponentFactory( + dispatcherRunnerFactory, + resourceManagerFactory, + SessionRestEndpointFactory.INSTANCE); } @Override protected ExecutionGraphInfoStore createSerializableExecutionGraphStore( - Configuration configuration, ScheduledExecutor scheduledExecutor) - throws IOException { - throw new UnsupportedOperationException("Not needed for this test"); + Configuration configuration, ScheduledExecutor scheduledExecutor) { + return new MemoryExecutionGraphInfoStore(); + } + + @Override + protected HighAvailabilityServices createHaServices( + Configuration configuration, Executor executor) { + return haService; } @Override protected boolean supportsReactiveMode() { return false; } + + public static final class Builder { + private HighAvailabilityServices haService = + new TestingHighAvailabilityServicesBuilder().build(); + + private ResourceManagerFactory<ResourceID> resourceManagerFactory = + StandaloneResourceManagerFactory.getInstance(); + + private DispatcherRunnerFactory dispatcherRunnerFactory = + new TestingDispatcherRunnerFactory.Builder().build(); + + private Configuration configuration = new Configuration(); + + public Builder setHighAvailabilityServices(HighAvailabilityServices haService) { + this.haService = haService; + return this; + } + + public Builder setResourceManagerFactory( + ResourceManagerFactory<ResourceID> resourceManagerFactory) { + this.resourceManagerFactory = resourceManagerFactory; + return this; + } + + public Builder setConfiguration(Configuration configuration) { + this.configuration = configuration; + return this; + } + + public Builder setDispatcherRunnerFactory( + DispatcherRunnerFactory dispatcherRunnerFactory) { + this.dispatcherRunnerFactory = dispatcherRunnerFactory; + return this; + } + + public TestingEntryPoint build() { + return new TestingEntryPoint( + configuration, haService, resourceManagerFactory, dispatcherRunnerFactory); + } + } + } + + private static class TestingDispatcherRunnerFactory implements DispatcherRunnerFactory { + + private final CompletableFuture<ApplicationStatus> shutDownFuture; + + private TestingDispatcherRunnerFactory( + CompletableFuture<ApplicationStatus> shutDownFuture) { + this.shutDownFuture = shutDownFuture; + } + + @Override + public DispatcherRunner createDispatcherRunner( + LeaderElectionService leaderElectionService, + FatalErrorHandler fatalErrorHandler, + JobGraphStoreFactory jobGraphStoreFactory, + Executor ioExecutor, + RpcService rpcService, + PartialDispatcherServices partialDispatcherServices) + throws Exception { + return TestingDispatcherRunner.newBuilder().setShutDownFuture(shutDownFuture).build(); + } + + public static final class Builder { + private CompletableFuture<ApplicationStatus> shutDownFuture = new CompletableFuture<>(); + + public Builder setShutDownFuture(CompletableFuture<ApplicationStatus> shutDownFuture) { + this.shutDownFuture = shutDownFuture; + return this; + } + + public TestingDispatcherRunnerFactory build() { + return new TestingDispatcherRunnerFactory(shutDownFuture); + } + } + } + + private static class TestingResourceManagerFactory extends ResourceManagerFactory<ResourceID> { + + private final BiConsumer<ApplicationStatus, String> deregisterAppConsumer; + + private TestingResourceManagerFactory( + BiConsumer<ApplicationStatus, String> deregisterAppConsumer) { + this.deregisterAppConsumer = deregisterAppConsumer; + } + + @Override + protected ResourceManager<ResourceID> createResourceManager( + Configuration configuration, + ResourceID resourceId, + RpcService rpcService, + HighAvailabilityServices highAvailabilityServices, + HeartbeatServices heartbeatServices, + FatalErrorHandler fatalErrorHandler, + ClusterInformation clusterInformation, + @Nullable String webInterfaceUrl, + ResourceManagerMetricGroup resourceManagerMetricGroup, + ResourceManagerRuntimeServices resourceManagerRuntimeServices, + Executor ioExecutor) + throws Exception { + final SlotManager slotManager = + SlotManagerBuilder.newBuilder() + .setScheduledExecutor(rpcService.getScheduledExecutor()) + .build(); + final JobLeaderIdService jobLeaderIdService = + new TestingJobLeaderIdService.Builder().build(); + return new TestingResourceManager( + rpcService, + resourceId, + highAvailabilityServices, + heartbeatServices, + slotManager, + NoOpResourceManagerPartitionTracker::get, + jobLeaderIdService, + fatalErrorHandler, + resourceManagerMetricGroup) { + @Override + protected void internalDeregisterApplication( + ApplicationStatus finalStatus, @Nullable String diagnostics) { + deregisterAppConsumer.accept(finalStatus, diagnostics); + } + }; + } + + @Override + protected ResourceManagerRuntimeServicesConfiguration + createResourceManagerRuntimeServicesConfiguration(Configuration configuration) + throws ConfigurationException { + return ResourceManagerRuntimeServicesConfiguration.fromConfiguration( + configuration, ArbitraryWorkerResourceSpecFactory.INSTANCE); + } + + public static final class Builder { + private BiConsumer<ApplicationStatus, String> deregisterAppConsumer = + (ignore1, ignore2) -> {}; + + public Builder setInternalDeregisterApplicationConsumer( + BiConsumer<ApplicationStatus, String> biConsumer) { + this.deregisterAppConsumer = biConsumer; + return this; + } + + public TestingResourceManagerFactory build() { + return new TestingResourceManagerFactory(deregisterAppConsumer); + } + } } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java index d805f4e..0afc981 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServices.java @@ -28,6 +28,7 @@ import org.apache.flink.runtime.leaderelection.LeaderElectionService; import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService; import java.io.IOException; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; @@ -67,6 +68,10 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices private volatile RunningJobsRegistry runningJobsRegistry = new StandaloneRunningJobsRegistry(); + private CompletableFuture<Void> closeFuture = new CompletableFuture<>(); + + private CompletableFuture<Void> closeAndCleanupAllDataFuture = new CompletableFuture<>(); + // ------------------------------------------------------------------------ // Setters for mock / testing implementations // ------------------------------------------------------------------------ @@ -131,6 +136,15 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices this.jobMasterLeaderRetrieverFunction = jobMasterLeaderRetrieverFunction; } + public void setCloseFuture(CompletableFuture<Void> closeFuture) { + this.closeFuture = closeFuture; + } + + public void setCloseAndCleanupAllDataFuture( + CompletableFuture<Void> closeAndCleanupAllDataFuture) { + this.closeAndCleanupAllDataFuture = closeAndCleanupAllDataFuture; + } + // ------------------------------------------------------------------------ // HA Services Methods // ------------------------------------------------------------------------ @@ -256,11 +270,11 @@ public class TestingHighAvailabilityServices implements HighAvailabilityServices @Override public void close() throws Exception { - // nothing to do + closeFuture.complete(null); } @Override public void closeAndCleanupAllData() throws Exception { - // nothing to do + closeAndCleanupAllDataFuture.complete(null); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServicesBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServicesBuilder.java index 64e2b07..dbf5269 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServicesBuilder.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/highavailability/TestingHighAvailabilityServicesBuilder.java @@ -29,6 +29,7 @@ import org.apache.flink.runtime.leaderelection.StandaloneLeaderElectionService; import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService; import org.apache.flink.runtime.leaderretrieval.StandaloneLeaderRetrievalService; +import java.util.concurrent.CompletableFuture; import java.util.function.Function; /** Builder for the {@link TestingHighAvailabilityServices}. */ @@ -70,6 +71,10 @@ public class TestingHighAvailabilityServicesBuilder { private RunningJobsRegistry runningJobsRegistry = new StandaloneRunningJobsRegistry(); + private CompletableFuture<Void> closeFuture = new CompletableFuture<>(); + + private CompletableFuture<Void> closeAndCleanupAllDataFuture = new CompletableFuture<>(); + public TestingHighAvailabilityServices build() { final TestingHighAvailabilityServices testingHighAvailabilityServices = new TestingHighAvailabilityServices(); @@ -96,6 +101,10 @@ public class TestingHighAvailabilityServicesBuilder { testingHighAvailabilityServices.setJobGraphStore(jobGraphStore); testingHighAvailabilityServices.setRunningJobsRegistry(runningJobsRegistry); + testingHighAvailabilityServices.setCloseFuture(closeFuture); + testingHighAvailabilityServices.setCloseAndCleanupAllDataFuture( + closeAndCleanupAllDataFuture); + return testingHighAvailabilityServices; } @@ -163,4 +172,16 @@ public class TestingHighAvailabilityServicesBuilder { this.runningJobsRegistry = runningJobsRegistry; return this; } + + public TestingHighAvailabilityServicesBuilder setCloseFuture( + CompletableFuture<Void> closeFuture) { + this.closeFuture = closeFuture; + return this; + } + + public TestingHighAvailabilityServicesBuilder setCloseAndCleanupAllDataFuture( + CompletableFuture<Void> closeAndCleanupAllDataFuture) { + this.closeAndCleanupAllDataFuture = closeAndCleanupAllDataFuture; + return this; + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java index 65e2c33..8ecf6db 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestJvmProcess.java @@ -31,6 +31,7 @@ import java.io.StringWriter; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.util.Arrays; +import java.util.concurrent.TimeUnit; import static org.apache.flink.runtime.testutils.CommonTestUtils.createTemporaryLog4JProperties; import static org.apache.flink.runtime.testutils.CommonTestUtils.getCurrentClasspath; @@ -288,6 +289,15 @@ public abstract class TestJvmProcess { } } + public boolean waitFor(long timeout, TimeUnit unit) throws InterruptedException { + final Process process = this.process; + if (process != null) { + return process.waitFor(timeout, unit); + } else { + throw new IllegalStateException("process not started"); + } + } + public int exitCode() { Process process = this.process; if (process != null) { @@ -324,6 +334,15 @@ public abstract class TestJvmProcess { } } + public static void killProcessWithSigTerm(long pid) throws Exception { + // send it a regular kill command (SIG_TERM) + final Process kill = Runtime.getRuntime().exec("kill " + pid); + kill.waitFor(); + if (kill.exitValue() != 0) { + fail("failed to send SIG_TERM to process " + pid); + } + } + public static void waitForMarkerFiles(File basedir, String prefix, int num, long timeout) { long now = System.currentTimeMillis(); final long deadline = now + timeout; diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestingClusterEntrypointProcess.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestingClusterEntrypointProcess.java new file mode 100644 index 0000000..9591e35 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/TestingClusterEntrypointProcess.java @@ -0,0 +1,136 @@ +/* + * 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.runtime.testutils; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.JobManagerOptions; +import org.apache.flink.configuration.RestOptions; +import org.apache.flink.runtime.concurrent.ScheduledExecutor; +import org.apache.flink.runtime.dispatcher.ExecutionGraphInfoStore; +import org.apache.flink.runtime.dispatcher.MemoryExecutionGraphInfoStore; +import org.apache.flink.runtime.entrypoint.ClusterEntrypoint; +import org.apache.flink.runtime.entrypoint.component.DefaultDispatcherResourceManagerComponentFactory; +import org.apache.flink.runtime.entrypoint.component.DispatcherResourceManagerComponentFactory; +import org.apache.flink.runtime.resourcemanager.StandaloneResourceManagerFactory; +import org.apache.flink.runtime.util.SignalHandler; +import org.apache.flink.util.IOUtils; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.concurrent.CompletableFuture; + +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** A testing {@link ClusterEntrypoint} instance running in a separate JVM. */ +public class TestingClusterEntrypointProcess extends TestJvmProcess { + + private final File markerFile; + + public TestingClusterEntrypointProcess(File markerFile) throws Exception { + this.markerFile = checkNotNull(markerFile, "marker file"); + } + + @Override + public String getName() { + return getClass().getCanonicalName(); + } + + @Override + public String[] getJvmArgs() { + return new String[] {markerFile.getAbsolutePath()}; + } + + @Override + public String getEntryPointClassName() { + return TestingClusterEntrypointProcessEntryPoint.class.getName(); + } + + @Override + public String toString() { + return getClass().getCanonicalName(); + } + + /** Entrypoint for the testing cluster entrypoint process. */ + public static class TestingClusterEntrypointProcessEntryPoint { + + private static final Logger LOG = + LoggerFactory.getLogger(TestingClusterEntrypointProcessEntryPoint.class); + + public static void main(String[] args) { + try { + final File markerFile = new File(args[0]); + + final Configuration config = new Configuration(); + config.setInteger(JobManagerOptions.PORT, 0); + config.setString(RestOptions.BIND_PORT, "0"); + + final TestingClusterEntrypoint clusterEntrypoint = + new TestingClusterEntrypoint(config, markerFile); + + SignalHandler.register(LOG); + clusterEntrypoint.startCluster(); + TestJvmProcess.touchFile(markerFile); + final int returnCode = + clusterEntrypoint.getTerminationFuture().get().processExitCode(); + System.exit(returnCode); + } catch (Throwable t) { + LOG.error("Failed to start TestingClusterEntrypoint process", t); + System.exit(1); + } + } + } + + private static class TestingClusterEntrypoint extends ClusterEntrypoint { + + private final File markerFile; + + protected TestingClusterEntrypoint(Configuration configuration, File markerFile) { + super(configuration); + this.markerFile = markerFile; + } + + @Override + protected DispatcherResourceManagerComponentFactory + createDispatcherResourceManagerComponentFactory(Configuration configuration) + throws IOException { + return DefaultDispatcherResourceManagerComponentFactory.createSessionComponentFactory( + StandaloneResourceManagerFactory.getInstance()); + } + + @Override + protected ExecutionGraphInfoStore createSerializableExecutionGraphStore( + Configuration configuration, ScheduledExecutor scheduledExecutor) + throws IOException { + return new MemoryExecutionGraphInfoStore(); + } + + @Override + public CompletableFuture<Void> closeAsync() { + return super.closeAsync() + .thenRun( + () -> { + LOG.info("Deleting markerFile {}", markerFile); + IOUtils.deleteFileQuietly(markerFile.toPath()); + }); + } + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/BlockingShutdownTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/BlockingShutdownTest.java index 8489a99..b4083fe 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/BlockingShutdownTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/BlockingShutdownTest.java @@ -28,7 +28,6 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.util.UUID; -import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.junit.Assume.assumeTrue; @@ -60,10 +59,7 @@ public class BlockingShutdownTest { // wait for the marker file to appear, which means the process is up properly TestJvmProcess.waitForMarkerFile(markerFile, 30000); - // send it a regular kill command (SIG_TERM) - Process kill = Runtime.getRuntime().exec("kill " + pid); - kill.waitFor(); - assertEquals("failed to send SIG_TERM to process", 0, kill.exitValue()); + TestJvmProcess.killProcessWithSigTerm(pid); // minimal delay until the Java process object notices that the process is gone // this will not let the test fail predictably if the process is actually in fact going @@ -105,10 +101,7 @@ public class BlockingShutdownTest { // wait for the marker file to appear, which means the process is up properly TestJvmProcess.waitForMarkerFile(markerFile, 30000); - // send it a regular kill command (SIG_TERM) - Process kill = Runtime.getRuntime().exec("kill " + pid); - kill.waitFor(); - assertEquals("failed to send SIG_TERM to process", 0, kill.exitValue()); + TestJvmProcess.killProcessWithSigTerm(pid); // the process should eventually go away final long deadline = System.nanoTime() + 30_000_000_000L; // 30 secs in nanos diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java index 5c2d5b9..6923706 100644 --- a/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java +++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/ProcessFailureCancelingITCase.java @@ -219,7 +219,7 @@ public class ProcessFailureCancelingITCase extends TestLogger { taskManagerProcess.destroy(); } if (dispatcherResourceManagerComponent != null) { - dispatcherResourceManagerComponent.deregisterApplicationAndClose( + dispatcherResourceManagerComponent.stopApplication( ApplicationStatus.SUCCEEDED, null); }