zhuzhurk commented on a change in pull request #9904: [FLINK-14232][runtime] Support global failure handling in NG scheduling URL: https://github.com/apache/flink/pull/9904#discussion_r338062348
########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/failover/flip1/ExecutionFailureHandlerTest.java ########## @@ -40,24 +42,38 @@ */ public class ExecutionFailureHandlerTest extends TestLogger { + private static final long restartDelayMs = 1234L; + + private FailoverTopology failoverTopology; + + private TestFailoverStrategy failoverStrategy; + + private TestRestartBackoffTimeStrategy restartStrategy; + + private ExecutionFailureHandler executionFailureHandler; + + @Before + public void setUp() { + TestFailoverTopology.Builder topologyBuilder = new TestFailoverTopology.Builder(); + topologyBuilder.newVertex(); + failoverTopology = topologyBuilder.build(); + + failoverStrategy = new TestFailoverStrategy(); + restartStrategy = new TestRestartBackoffTimeStrategy(true, restartDelayMs); + executionFailureHandler = new ExecutionFailureHandler(failoverTopology, failoverStrategy, restartStrategy); + } + /** * Tests the case that task restarting is accepted. */ @Test public void testNormalFailureHandling() { - // failover strategy which always suggests restarting the given tasks - Set<ExecutionVertexID> tasksToRestart = new HashSet<>(); - tasksToRestart.add(new ExecutionVertexID(new JobVertexID(), 0)); - FailoverStrategy failoverStrategy = new TestFailoverStrategy(tasksToRestart); - - // restart strategy which accepts restarting - boolean canRestart = true; - long restartDelayMs = 1234; - RestartBackoffTimeStrategy restartStrategy = new TestRestartBackoffTimeStrategy(canRestart, restartDelayMs); - ExecutionFailureHandler executionFailureHandler = new ExecutionFailureHandler(failoverStrategy, restartStrategy); + final Set<ExecutionVertexID> tasksToRestart = Collections.singleton( + new ExecutionVertexID(new JobVertexID(), 0)); + failoverStrategy.setTasksToRestart(tasksToRestart); // trigger a task failure - FailureHandlingResult result = executionFailureHandler.getFailureHandlingResult( + final FailureHandlingResult result = executionFailureHandler.getFailureHandlingResult( new ExecutionVertexID(new JobVertexID(), 0), Review comment: It's by design to be different so that we are sure it's really returning the result from failoverStrategy `failoverStrategy.setTasksToRestart`. This case is to verify task failure handling and is using a `TestFailoverStrategy` so it does not rely on the `FailoverTopology`. ---------------------------------------------------------------- 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: us...@infra.apache.org With regards, Apache Git Services