[FLINK-4314] [tests] Fix test instability in JobManagerHAJobGraphReco…

…veryITCase

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.

flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAJobGraphRecoveryITCase.java

@@ -25,6 +25,8 @@
 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 void testJobPersistencyWhenJobManagerShutdown() throws Exception {
 
 			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