@Test public void testNormalFailureHandling() { Set<ExecutionVertexID> tasksToRestart = new HashSet<>(); tasksToRestart.add(new ExecutionVertexID(new JobVertexID(), 0)); FailoverStrategy failoverStrategy = new TestFailoverStrategy(tasksToRestart); boolean canRestart = true; long restartDelayMs = 1234; RestartBackoffTimeStrategy restartStrategy = new TestRestartBackoffTimeStrategy(canRestart, restartDelayMs); ExecutionFailureHandler executionFailureHandler = new ExecutionFailureHandler(failoverStrategy, restartStrategy); FailureHandlingResult result = executionFailureHandler.getFailureHandlingResult( new ExecutionVertexID(new JobVertexID(), 0), new Exception("test failure")); assertTrue(result.canRestart()); assertEquals(restartDelayMs, result.getRestartDelayMS()); assertEquals(tasksToRestart, result.getVerticesToRestart()); try { result.getError(); fail("Cannot get error when the restarting is accepted"); } catch (IllegalStateException ex) { } }
public FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause) { if (isUnrecoverableError(cause)) { return FailureHandlingResult.unrecoverable(new JobException("The failure is not recoverable", cause)); } restartBackoffTimeStrategy.notifyFailure(cause); if (restartBackoffTimeStrategy.canRestart()) { return FailureHandlingResult.restartable( failoverStrategy.getTasksNeedingRestart(failedTask, cause), restartBackoffTimeStrategy.getBackoffTime()); } else { return FailureHandlingResult.unrecoverable( new JobException("Failed task restarting is suppressed by " + restartBackoffTimeStrategy, cause)); } }
ExecutionFailureHandler { public FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause) { if (isUnrecoverableError(cause)) { return FailureHandlingResult.unrecoverable(new JobException("The failure is not recoverable", cause)); } restartBackoffTimeStrategy.notifyFailure(cause); if (restartBackoffTimeStrategy.canRestart()) { return FailureHandlingResult.restartable( failoverStrategy.getTasksNeedingRestart(failedTask, cause), restartBackoffTimeStrategy.getBackoffTime()); } else { return FailureHandlingResult.unrecoverable( new JobException("Failed task restarting is suppressed by " + restartBackoffTimeStrategy, cause)); } } }
ExecutionFailureHandler { public FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause) { if (isUnrecoverableError(cause)) { return FailureHandlingResult.unrecoverable(new JobException("The failure is not recoverable", cause)); } restartBackoffTimeStrategy.notifyFailure(cause); if (restartBackoffTimeStrategy.canRestart()) { return FailureHandlingResult.restartable( failoverStrategy.getTasksNeedingRestart(failedTask, cause), restartBackoffTimeStrategy.getBackoffTime()); } else { return FailureHandlingResult.unrecoverable( new JobException("Failed task restarting is suppressed by " + restartBackoffTimeStrategy, cause)); } } ExecutionFailureHandler(
		FailoverStrategy failoverStrategy,
		RestartBackoffTimeStrategy restartBackoffTimeStrategy); }
ExecutionFailureHandler { public FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause) { if (isUnrecoverableError(cause)) { return FailureHandlingResult.unrecoverable(new JobException("The failure is not recoverable", cause)); } restartBackoffTimeStrategy.notifyFailure(cause); if (restartBackoffTimeStrategy.canRestart()) { return FailureHandlingResult.restartable( failoverStrategy.getTasksNeedingRestart(failedTask, cause), restartBackoffTimeStrategy.getBackoffTime()); } else { return FailureHandlingResult.unrecoverable( new JobException("Failed task restarting is suppressed by " + restartBackoffTimeStrategy, cause)); } } ExecutionFailureHandler(
		FailoverStrategy failoverStrategy,
		RestartBackoffTimeStrategy restartBackoffTimeStrategy); FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause); }
ExecutionFailureHandler { public FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause) { if (isUnrecoverableError(cause)) { return FailureHandlingResult.unrecoverable(new JobException("The failure is not recoverable", cause)); } restartBackoffTimeStrategy.notifyFailure(cause); if (restartBackoffTimeStrategy.canRestart()) { return FailureHandlingResult.restartable( failoverStrategy.getTasksNeedingRestart(failedTask, cause), restartBackoffTimeStrategy.getBackoffTime()); } else { return FailureHandlingResult.unrecoverable( new JobException("Failed task restarting is suppressed by " + restartBackoffTimeStrategy, cause)); } } ExecutionFailureHandler(
		FailoverStrategy failoverStrategy,
		RestartBackoffTimeStrategy restartBackoffTimeStrategy); FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause);  }