tillrohrmann commented on a change in pull request #9832: [FLINK-11843] Bind lifespan of Dispatcher to leader session URL: https://github.com/apache/flink/pull/9832#discussion_r337408843
########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/runner/DefaultDispatcherRunnerTest.java ########## @@ -0,0 +1,415 @@ +/* + * 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.dispatcher.runner; + +import org.apache.flink.runtime.clusterframework.ApplicationStatus; +import org.apache.flink.runtime.concurrent.FutureUtils; +import org.apache.flink.runtime.dispatcher.DispatcherGateway; +import org.apache.flink.runtime.dispatcher.DispatcherId; +import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService; +import org.apache.flink.runtime.util.LeaderConnectionInfo; +import org.apache.flink.runtime.util.TestingFatalErrorHandler; +import org.apache.flink.runtime.webmonitor.TestingDispatcherGateway; +import org.apache.flink.util.TestLogger; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.fail; + +/** + * Tests for the {@link DefaultDispatcherRunner}. + */ +public class DefaultDispatcherRunnerTest extends TestLogger { + + private TestingLeaderElectionService testingLeaderElectionService; + private TestingFatalErrorHandler testingFatalErrorHandler; + private TestingDispatcherLeaderProcessFactory testingDispatcherLeaderProcessFactory; + + @Before + public void setup() { + testingLeaderElectionService = new TestingLeaderElectionService(); + testingFatalErrorHandler = new TestingFatalErrorHandler(); + testingDispatcherLeaderProcessFactory = TestingDispatcherLeaderProcessFactory.defaultValue(); + } + + @After + public void teardown() throws Exception { + if (testingLeaderElectionService != null) { + testingLeaderElectionService.stop(); + testingLeaderElectionService = null; + } + + if (testingFatalErrorHandler != null) { + testingFatalErrorHandler.rethrowError(); + testingFatalErrorHandler = null; + } + } + + @Test + public void closeAsync_doesNotCompleteUncompletedShutDownFuture() throws Exception { + final DefaultDispatcherRunner dispatcherRunner = createDispatcherRunner(); + + final CompletableFuture<Void> terminationFuture = dispatcherRunner.closeAsync(); + terminationFuture.get(); + + final CompletableFuture<ApplicationStatus> shutDownFuture = dispatcherRunner.getShutDownFuture(); + assertThat(shutDownFuture.isDone(), is(false)); + } + + @Test + public void getDispatcherGateway_beforeDispatcherLeaderProcessCompletes_returnsDispatcherGateway() throws Exception { + final UUID leaderSessionId = UUID.randomUUID(); + final TestingDispatcherGateway expectedDispatcherGateway = createDispatcherGateway(leaderSessionId); + final TestingDispatcherLeaderProcess testingDispatcherLeaderProcess = TestingDispatcherLeaderProcess.newBuilder(leaderSessionId) + .setDispatcherGatewayFuture(CompletableFuture.completedFuture(expectedDispatcherGateway)) + .build(); + + testingDispatcherLeaderProcessFactory = TestingDispatcherLeaderProcessFactory.from(testingDispatcherLeaderProcess); + try (final DefaultDispatcherRunner dispatcherRunner = createDispatcherRunner()) { + + final CompletableFuture<DispatcherGateway> dispatcherGatewayFuture = dispatcherRunner.getDispatcherGateway(); + + assertThat(dispatcherGatewayFuture.isDone(), is(false)); + + testingLeaderElectionService.isLeader(leaderSessionId); + + assertThat(dispatcherGatewayFuture.get(), is(expectedDispatcherGateway)); + } + } + + @Test + public void getDispatcherGateway_withChangingLeaders_returnsLeadingDispatcherGateway() throws Exception { + final UUID firstLeaderSessionId = UUID.randomUUID(); + final UUID secondLeaderSessionId = UUID.randomUUID(); + final TestingDispatcherGateway firstDispatcherGateway = createDispatcherGateway(firstLeaderSessionId); + final TestingDispatcherGateway secondDispatcherGateway = createDispatcherGateway(secondLeaderSessionId); + + final TestingDispatcherLeaderProcess firstDispatcherLeaderProcess = TestingDispatcherLeaderProcess.newBuilder(firstLeaderSessionId) + .setDispatcherGatewayFuture(CompletableFuture.completedFuture(firstDispatcherGateway)) + .build(); + final TestingDispatcherLeaderProcess secondDispatcherLeaderProcess = TestingDispatcherLeaderProcess.newBuilder(secondLeaderSessionId) + .setDispatcherGatewayFuture(CompletableFuture.completedFuture(secondDispatcherGateway)) + .build(); + + testingDispatcherLeaderProcessFactory = TestingDispatcherLeaderProcessFactory.from( + firstDispatcherLeaderProcess, + secondDispatcherLeaderProcess); + + try (final DefaultDispatcherRunner dispatcherRunner = createDispatcherRunner()) { + testingLeaderElectionService.isLeader(firstLeaderSessionId); + + final CompletableFuture<DispatcherGateway> firstDispatcherGatewayFuture = dispatcherRunner.getDispatcherGateway(); + + testingLeaderElectionService.notLeader(); + testingLeaderElectionService.isLeader(secondLeaderSessionId); + + final CompletableFuture<DispatcherGateway> secondDispatcherGatewayFuture = dispatcherRunner.getDispatcherGateway(); + + assertThat(firstDispatcherGatewayFuture.get(), is(firstDispatcherGateway)); + assertThat(secondDispatcherGatewayFuture.get(), is(secondDispatcherGateway)); + } + } + + @Test + public void getShutDownFuture_whileRunning_forwardsDispatcherLeaderProcessShutDownRequest() throws Exception { + final UUID leaderSessionId = UUID.randomUUID(); + final CompletableFuture<ApplicationStatus> shutDownFuture = new CompletableFuture<>(); + final TestingDispatcherLeaderProcess testingDispatcherLeaderProcess = TestingDispatcherLeaderProcess.newBuilder(leaderSessionId) + .setShutDownFuture(shutDownFuture) + .build(); + testingDispatcherLeaderProcessFactory = TestingDispatcherLeaderProcessFactory.from(testingDispatcherLeaderProcess); + + try (final DefaultDispatcherRunner dispatcherRunner = createDispatcherRunner()) { + testingLeaderElectionService.isLeader(leaderSessionId); + + final CompletableFuture<ApplicationStatus> dispatcherShutDownFuture = dispatcherRunner.getShutDownFuture(); + + assertFalse(dispatcherShutDownFuture.isDone()); + + final ApplicationStatus finalApplicationStatus = ApplicationStatus.UNKNOWN; + shutDownFuture.complete(finalApplicationStatus); + + assertThat(dispatcherShutDownFuture.get(), is(finalApplicationStatus)); + } + } + + @Test + public void getShutDownFuture_afterClose_ignoresDispatcherLeaderProcessShutDownRequest() throws Exception { + final UUID leaderSessionId = UUID.randomUUID(); + final CompletableFuture<ApplicationStatus> shutDownFuture = new CompletableFuture<>(); + final TestingDispatcherLeaderProcess testingDispatcherLeaderProcess = TestingDispatcherLeaderProcess.newBuilder(leaderSessionId) + .setShutDownFuture(shutDownFuture) + .build(); + testingDispatcherLeaderProcessFactory = TestingDispatcherLeaderProcessFactory.from(testingDispatcherLeaderProcess); + + try (final DefaultDispatcherRunner dispatcherRunner = createDispatcherRunner()) { + testingLeaderElectionService.isLeader(leaderSessionId); + + final CompletableFuture<ApplicationStatus> dispatcherShutDownFuture = dispatcherRunner.getShutDownFuture(); + + assertFalse(dispatcherShutDownFuture.isDone()); + + dispatcherRunner.closeAsync(); + + final ApplicationStatus finalApplicationStatus = ApplicationStatus.UNKNOWN; + shutDownFuture.complete(finalApplicationStatus); + + try { + dispatcherShutDownFuture.get(10L, TimeUnit.MILLISECONDS); + fail("The dispatcher runner should no longer react to the dispatcher leader process's shut down request if it has been terminated."); + } catch (TimeoutException expected) {} + } + } + + @Test + public void getShutDownFuture_newLeader_ignoresOldDispatcherLeaderProcessShutDownRequest() throws Exception { + final UUID firstLeaderSessionId = UUID.randomUUID(); + final UUID secondLeaderSessionId = UUID.randomUUID(); + final CompletableFuture<ApplicationStatus> shutDownFuture = new CompletableFuture<>(); + final TestingDispatcherLeaderProcess firstTestingDispatcherLeaderProcess = TestingDispatcherLeaderProcess.newBuilder(firstLeaderSessionId) + .setShutDownFuture(shutDownFuture) + .build(); + final TestingDispatcherLeaderProcess secondTestingDispatcherLeaderProcess = TestingDispatcherLeaderProcess.newBuilder(secondLeaderSessionId) + .build(); + testingDispatcherLeaderProcessFactory = TestingDispatcherLeaderProcessFactory.from( + firstTestingDispatcherLeaderProcess, + secondTestingDispatcherLeaderProcess); + + try (final DefaultDispatcherRunner dispatcherRunner = createDispatcherRunner()) { + testingLeaderElectionService.isLeader(firstLeaderSessionId); + + final CompletableFuture<ApplicationStatus> dispatcherShutDownFuture = dispatcherRunner.getShutDownFuture(); + + assertFalse(dispatcherShutDownFuture.isDone()); + + testingLeaderElectionService.isLeader(secondLeaderSessionId); + + final ApplicationStatus finalApplicationStatus = ApplicationStatus.UNKNOWN; + shutDownFuture.complete(finalApplicationStatus); + + try { + dispatcherShutDownFuture.get(10L, TimeUnit.MILLISECONDS); + fail("The dispatcher runner should no longer react to the dispatcher leader process's shut down request if it has been terminated."); + } catch (TimeoutException expected) {} + } + } + + @Test + public void revokeLeadership_withExistingLeader_stopsLeaderProcess() throws Exception { + final UUID leaderSessionId = UUID.randomUUID(); + + final CompletableFuture<Void> startFuture = new CompletableFuture<>(); Review comment: True, even though it does not make a big difference imo. Will change it. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services
