flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From u..@apache.org
Subject flink git commit: [FLINK-4314] [tests] Fix test instability in JobManagerHAJobGraphRecoveryITCase
Date Fri, 05 Aug 2016 09:53:33 GMT
Repository: flink
Updated Branches:
  refs/heads/release-1.1 5363e7ad7 -> 5184d0bb4


[FLINK-4314] [tests] Fix test instability in JobManagerHAJobGraphRecoveryITCase

The test was relying on the JobManager shutting down before the
TaskManager, which is not necessarily the case. If the TaskManager
shuts down before the JobManager, the JobGraph could reach the final
state FAILED, in which case all HA state is removed.

To circumvent this, we add a restart strategy.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/5184d0bb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/5184d0bb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/5184d0bb

Branch: refs/heads/release-1.1
Commit: 5184d0bb4941cdbe09a5288a9a3cc21bbd116c43
Parents: 5363e7a
Author: Ufuk Celebi <uce@apache.org>
Authored: Fri Aug 5 11:49:42 2016 +0200
Committer: Ufuk Celebi <uce@apache.org>
Committed: Fri Aug 5 11:53:25 2016 +0200

----------------------------------------------------------------------
 .../test/recovery/JobManagerHAJobGraphRecoveryITCase.java   | 9 +++++++++
 1 file changed, 9 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/5184d0bb/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java
index 5a10604..eccf971 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java
@@ -25,6 +25,8 @@ import akka.actor.UntypedActor;
 import akka.testkit.TestActorRef;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.filefilter.TrueFileFilter;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
 import org.apache.flink.configuration.ConfigConstants;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.runtime.akka.AkkaUtils;
@@ -140,6 +142,13 @@ public class JobManagerHAJobGraphRecoveryITCase extends TestLogger {
 
 			JobGraph jobGraph = createBlockingJobGraph();
 
+			// Set restart strategy to guard against shut down races.
+			// If the TM fails before the JM, it might happen that the
+			// Job is failed, leading to state removal.
+			ExecutionConfig ec = new ExecutionConfig();
+			ec.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 100));
+			jobGraph.setExecutionConfig(ec);
+
 			ActorGateway jobManager = flink.getLeaderGateway(deadline.timeLeft());
 
 			// Submit the job


Mime
View raw message