[flink] 03/03: [FLINK-20033] Ensure that stopping a JobMaster will suspend the running job

[tr...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 32e23052af547bb1b99274778c666f3d814201a8
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Sat Nov 7 17:22:52 2020 +0100

    [FLINK-20033] Ensure that stopping a JobMaster will suspend the running job
    
    This commit adds a test which ensures that stopping a JobMaster will suspend the running job.
    
    This closes #13979.
---
 .../flink/runtime/jobmaster/JobMasterTest.java     | 52 ++++++++++++++++++++++
 1 file changed, 52 insertions(+)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index 83016f8..5a28fad 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -1961,6 +1961,58 @@ public class JobMasterTest extends TestLogger {
 		);
 	}
 
+	/**
+	 * Tests that the job gets suspended when the JobMaster stops. See FLINK-20033.
+	 */
+	@Test
+	public void testJobSuspensionWhenJobMasterStops() throws Exception {
+		final JobMasterBuilder.TestingOnCompletionActions onCompletionActions = new JobMasterBuilder.TestingOnCompletionActions();
+
+		final JobMaster jobMaster = createJobMaster(
+			configuration,
+			JobGraphTestUtils.createSingleVertexJobGraph(),
+			haServices,
+			new TestingJobManagerSharedServicesBuilder().build(),
+			heartbeatServices,
+			onCompletionActions);
+
+		CompletableFuture<Acknowledge> startFuture = jobMaster.start(jobMasterId);
+
+		try {
+			// wait for the start of the JobMaster
+			startFuture.get();
+
+			final CompletableFuture<TaskDeploymentDescriptor> tddFuture = new CompletableFuture<>();
+			final TestingTaskExecutorGateway testingTaskExecutorGateway = new TestingTaskExecutorGatewayBuilder()
+				.setSubmitTaskConsumer((taskDeploymentDescriptor, jobMasterId) -> {
+					tddFuture.complete(taskDeploymentDescriptor);
+					return CompletableFuture.completedFuture(Acknowledge.get());
+				})
+				.createTestingTaskExecutorGateway();
+			final LocalUnresolvedTaskManagerLocation taskManagerLocation = new LocalUnresolvedTaskManagerLocation();
+			final JobMasterGateway jobMasterGateway = jobMaster.getSelfGateway(JobMasterGateway.class);
+
+			registerSlotsAtJobMaster(1, jobMasterGateway, testingTaskExecutorGateway, taskManagerLocation);
+
+			// wait for deployment
+			tddFuture.get();
+
+			// trigger termination of JobMaster and also the job
+			jobMaster.close();
+
+			try {
+				final ArchivedExecutionGraph archivedExecutionGraph = onCompletionActions
+					.getJobReachedGloballyTerminalStateFuture()
+					.get(50, TimeUnit.MILLISECONDS);
+				fail(String.format("Job must not reach the globally terminal state %s when stopping the JobMaster.", archivedExecutionGraph.getState()));
+			} catch (TimeoutException expected) {
+				// expected
+			}
+		} finally {
+			RpcUtils.terminateRpcEndpoint(jobMaster, testingTimeout);
+		}
+	}
+
 	private void runJobFailureWhenTaskExecutorTerminatesTest(
 			HeartbeatServices heartbeatServices,
 			BiConsumer<LocalUnresolvedTaskManagerLocation, JobMasterGateway> jobReachedRunningState,