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:
[email protected]
With regards,
Apache Git Services
