[flink] 07/12: [FLINK-10406] (Part 7) testResourceManagerConnection

[tr...@apache.org]

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 015150e4505f08fb0eee705206244ecebc1a37e2
Author: tison <wa...@gmail.com>
AuthorDate: Wed Sep 26 19:30:53 2018 +0800

    [FLINK-10406] (Part 7) testResourceManagerConnection
    
    testResourceManagerConnection is not applicable to FLIP-6 new code base,
    FLIP-6 has its own reconnect logic between JM and RM and the mechanism
    should be guarded by
    
    JobMasterTest#testReconnectionAfterDisconnect
    JobMasterTest#testResourceManagerConnectionAfterRegainingLeadership
    JobMasterTest#testCloseUnestablishedResourceManagerConnection
---
 .../flink/runtime/jobmanager/JobManagerTest.java   | 84 ----------------------
 1 file changed, 84 deletions(-)

diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
index c1bdcc1..b6385b0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/JobManagerTest.java
@@ -355,88 +355,4 @@ public class JobManagerTest extends TestLogger {
 		assertEquals(JobStatus.FAILED, jobStatus.state());
 
 	}
-
-	/**
-	 * This tests makes sure that triggering a reconnection from the ResourceManager will stop after a new
-	 * ResourceManager has connected. Furthermore it makes sure that there is not endless loop of reconnection
-	 * commands (see FLINK-6341).
-	 */
-	@Test
-	public void testResourceManagerConnection() throws TimeoutException, InterruptedException {
-		FiniteDuration testTimeout = new FiniteDuration(30L, TimeUnit.SECONDS);
-		final long reconnectionInterval = 200L;
-
-		final Configuration configuration = new Configuration();
-		configuration.setLong(JobManagerOptions.RESOURCE_MANAGER_RECONNECT_INTERVAL, reconnectionInterval);
-
-		final ActorSystem actorSystem = AkkaUtils.createLocalActorSystem(configuration);
-
-		try {
-			final ActorGateway jmGateway = TestingUtils.createJobManager(
-				actorSystem,
-				TestingUtils.defaultExecutor(),
-				TestingUtils.defaultExecutor(),
-				configuration,
-				highAvailabilityServices);
-
-			final TestProbe probe = TestProbe.apply(actorSystem);
-			final AkkaActorGateway rmGateway = new AkkaActorGateway(probe.ref(), HighAvailabilityServices.DEFAULT_LEADER_ID);
-
-			// wait for the JobManager to become the leader
-			Future<?> leaderFuture = jmGateway.ask(TestingJobManagerMessages.getNotifyWhenLeader(), testTimeout);
-			Await.ready(leaderFuture, testTimeout);
-
-			jmGateway.tell(new RegisterResourceManager(probe.ref()), rmGateway);
-
-			LeaderSessionMessage leaderSessionMessage = probe.expectMsgClass(LeaderSessionMessage.class);
-
-			assertEquals(jmGateway.leaderSessionID(), leaderSessionMessage.leaderSessionID());
-			assertTrue(leaderSessionMessage.message() instanceof RegisterResourceManagerSuccessful);
-
-			jmGateway.tell(
-				new RegistrationMessages.RegisterTaskManager(
-					ResourceID.generate(),
-					mock(TaskManagerLocation.class),
-					new HardwareDescription(1, 1L, 1L, 1L),
-					1));
-			leaderSessionMessage = probe.expectMsgClass(LeaderSessionMessage.class);
-
-			assertTrue(leaderSessionMessage.message() instanceof NotifyResourceStarted);
-
-			// fail the NotifyResourceStarted so that we trigger the reconnection process on the JobManager's side
-			probe.lastSender().tell(new Status.Failure(new Exception("Test exception")), ActorRef.noSender());
-
-			Deadline reconnectionDeadline = new FiniteDuration(5L * reconnectionInterval, TimeUnit.MILLISECONDS).fromNow();
-			boolean registered = false;
-
-			while (reconnectionDeadline.hasTimeLeft()) {
-				try {
-					leaderSessionMessage = probe.expectMsgClass(reconnectionDeadline.timeLeft(), LeaderSessionMessage.class);
-				} catch (AssertionError ignored) {
-					// expected timeout after the reconnectionDeadline has been exceeded
-					continue;
-				}
-
-				if (leaderSessionMessage.message() instanceof TriggerRegistrationAtJobManager) {
-					if (registered) {
-						fail("A successful registration should not be followed by another TriggerRegistrationAtJobManager message.");
-					}
-
-					jmGateway.tell(new RegisterResourceManager(probe.ref()), rmGateway);
-				} else if (leaderSessionMessage.message() instanceof RegisterResourceManagerSuccessful) {
-					// now we should no longer receive TriggerRegistrationAtJobManager messages
-					registered = true;
-				} else {
-					fail("Received unknown message: " + leaderSessionMessage.message() + '.');
-				}
-			}
-
-			assertTrue(registered);
-
-		} finally {
-			// cleanup the actor system and with it all of the started actors if not already terminated
-			actorSystem.terminate();
-			Await.ready(actorSystem.whenTerminated(), Duration.Inf());
-		}
-	}
 }