[GitHub] eaglewatcherwb closed pull request #7473: [FLINK-11295][test] add retry test

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

eaglewatcherwb closed pull request #7473: [FLINK-11295][test] add retry test
URL: https://github.com/apache/flink/pull/7473
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
index 2882fc7e908..f4a8b74f02e 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
@@ -18,7 +18,9 @@
 
 package org.apache.flink.examples.java.wordcount;
 
+import org.apache.flink.api.common.ExecutionMode;
 import org.apache.flink.api.common.functions.FlatMapFunction;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.tuple.Tuple2;
@@ -55,10 +57,15 @@ public static void main(String[] args) throws Exception {
 
 		// set up the execution environment
 		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+		env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, 1000L));
 
 		// make parameters available in the web interface
 		env.getConfig().setGlobalJobParameters(params);
 
+		if (params.has("executionMode")) {
+			env.getConfig().setExecutionMode(ExecutionMode.valueOf(params.get("executionMode")));
+		}
+
 		// get input data
 		DataSet<String> text;
 		if (params.has("input")) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/RestartPipelinedRegionStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/RestartPipelinedRegionStrategy.java
index b8f69641fd7..cb0a4d63682 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/RestartPipelinedRegionStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/failover/RestartPipelinedRegionStrategy.java
@@ -206,7 +206,7 @@ else if (predecessor != null) {
 
 		// now that we have all regions, create the failover region objects 
 		LOG.info("Creating {} individual failover regions for job {} ({})",
-				executionGraph.getJobName(), executionGraph.getJobID());
+			distinctRegions.size(), executionGraph.getJobName(), executionGraph.getJobID());
 
 		for (List<ExecutionVertex> region : distinctRegions.keySet()) {
 			final FailoverRegion failoverRegion = new FailoverRegion(executionGraph, executor, region);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
index 9abf92f76fd..b4bc0e542ed 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java
@@ -709,6 +709,13 @@ else if (current == ExecutionState.CANCELING) {
 				throw new CancelTaskException();
 			}
 
+			//*** demo code begin
+			if (taskInfo.getTaskName().contains("Reduce") && taskInfo.getIndexOfThisSubtask() == 0
+				&& taskInfo.getAttemptNumber() == 0) {
+				throw new RuntimeException("crash");
+			}
+			//*** demo code end
+
 			// ----------------------------------------------------------------
 			//  finalization of a successful execution
 			// ----------------------------------------------------------------


 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on 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