[GitHub] [flink] zhuzhurk commented on a change in pull request #9904: [FLINK-14232][runtime] Support global failure handling in NG scheduling

[GitBox <gi...@apache.org>]

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_r337681258
 
 

 ##########
 File path: flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/flip1/ExecutionFailureHandler.java
 ##########
 @@ -62,18 +71,41 @@ public ExecutionFailureHandler(
 	 * @return result of the failure handling
 	 */
 	public FailureHandlingResult getFailureHandlingResult(ExecutionVertexID failedTask, Throwable cause) {
+		return handleFailure(cause, () -> failoverStrategy.getTasksNeedingRestart(failedTask, cause));
+	}
+
+	/**
+	 * Return result of failure handling on a global failure. Can be a set of task vertices to restart
+	 * and a delay of the restarting. Or that the failure is not recoverable and the reason for it.
+	 *
+	 * @param cause of the task failure
+	 * @return result of the failure handling
+	 */
+	public FailureHandlingResult getGlobalFailureHandlingResult(final Throwable cause) {
+		return handleFailure(
+			cause,
+			() -> StreamSupport
+				.stream(failoverTopology.getFailoverVertices().spliterator(), false)
+				.map(FailoverVertex::getExecutionVertexID)
+				.collect(Collectors.toSet()));
+	}
+
+	private FailureHandlingResult handleFailure(
+			final Throwable cause,
+			final Supplier<Set<ExecutionVertexID>> vertexToRestartSupplier) {
 
 Review comment:
   You are right. Done.

----------------------------------------------------------------
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:
users@infra.apache.org


With regards,
Apache Git Services