You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2016/02/15 15:59:53 UTC
[2/3] flink git commit: [FLINK-3187] [restart] Introduce
RestartStrategy to ExecutionGraph
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
index 47a48a2..7cef66a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java
@@ -23,6 +23,9 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.execution.UnrecoverableException;
+import org.apache.flink.runtime.executiongraph.restart.FixedDelayRestartStrategy;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
+import org.apache.flink.runtime.executiongraph.restart.RestartStrategy;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobStatus;
@@ -32,6 +35,7 @@ import org.apache.flink.runtime.jobmanager.scheduler.CoLocationConstraint;
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.util.TestLogger;
import org.junit.Test;
import scala.concurrent.duration.Deadline;
import scala.concurrent.duration.FiniteDuration;
@@ -50,7 +54,7 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
-public class ExecutionGraphRestartTest {
+public class ExecutionGraphRestartTest extends TestLogger {
private final static int NUM_TASKS = 31;
@@ -74,8 +78,8 @@ public class ExecutionGraphRestartTest {
new JobID(),
"test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
- eg.setNumberOfRetriesLeft(0);
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
assertEquals(JobStatus.CREATED, eg.getState());
@@ -128,8 +132,8 @@ public class ExecutionGraphRestartTest {
new JobID(),
"test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
- eg.setNumberOfRetriesLeft(1);
+ AkkaUtils.getDefaultTimeout(),
+ new FixedDelayRestartStrategy(1, 0L));
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
assertEquals(JobStatus.CREATED, eg.getState());
@@ -183,8 +187,8 @@ public class ExecutionGraphRestartTest {
new JobID(),
"Test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
- eg.setNumberOfRetriesLeft(1);
+ AkkaUtils.getDefaultTimeout(),
+ new FixedDelayRestartStrategy(1, 1000));
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
assertEquals(JobStatus.CREATED, eg.getState());
@@ -211,7 +215,9 @@ public class ExecutionGraphRestartTest {
new JobID(),
"TestJob",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ // We want to manually control the restart and delay
+ new FixedDelayRestartStrategy(Integer.MAX_VALUE, Long.MAX_VALUE));
JobVertex jobVertex = new JobVertex("NoOpInvokable");
jobVertex.setInvokableClass(Tasks.NoOpInvokable.class);
@@ -219,9 +225,6 @@ public class ExecutionGraphRestartTest {
JobGraph jobGraph = new JobGraph("TestJob", jobVertex);
- // We want to manually control the restart and delay
- executionGraph.setNumberOfRetriesLeft(Integer.MAX_VALUE);
- executionGraph.setDelayBeforeRetrying(Integer.MAX_VALUE);
executionGraph.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
assertEquals(JobStatus.CREATED, executionGraph.getState());
@@ -270,7 +273,9 @@ public class ExecutionGraphRestartTest {
new JobID(),
"TestJob",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ // We want to manually control the restart and delay
+ new FixedDelayRestartStrategy(Integer.MAX_VALUE, Long.MAX_VALUE));
// Spy on the graph
executionGraph = spy(executionGraph);
@@ -285,9 +290,6 @@ public class ExecutionGraphRestartTest {
JobGraph jobGraph = new JobGraph("TestJob", jobVertex);
- // We want to manually control the restart and delay
- executionGraph.setNumberOfRetriesLeft(Integer.MAX_VALUE);
- executionGraph.setDelayBeforeRetrying(Integer.MAX_VALUE);
executionGraph.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
assertEquals(JobStatus.CREATED, executionGraph.getState());
@@ -348,13 +350,13 @@ public class ExecutionGraphRestartTest {
JobGraph jobGraph = new JobGraph("Pointwise job", sender);
ExecutionGraph eg = spy(new ExecutionGraph(
- TestingUtils.defaultExecutionContext(),
- new JobID(),
- "Test job",
- new Configuration(),
- AkkaUtils.getDefaultTimeout()));
+ TestingUtils.defaultExecutionContext(),
+ new JobID(),
+ "Test job",
+ new Configuration(),
+ AkkaUtils.getDefaultTimeout(),
+ new FixedDelayRestartStrategy(1, 1000)));
- eg.setNumberOfRetriesLeft(1);
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
assertEquals(JobStatus.CREATED, eg.getState());
@@ -385,7 +387,12 @@ public class ExecutionGraphRestartTest {
// No restart
verify(eg, never()).restart();
- assertEquals(1, eg.getNumberOfRetriesLeft());
+
+ RestartStrategy restartStrategy = eg.getRestartStrategy();
+
+ assertTrue(restartStrategy instanceof FixedDelayRestartStrategy);
+
+ assertEquals(0, ((FixedDelayRestartStrategy) restartStrategy).getCurrentRestartAttempt());
}
/**
@@ -417,8 +424,9 @@ public class ExecutionGraphRestartTest {
new JobID(),
"test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
- eg.setNumberOfRetriesLeft(1);
+ AkkaUtils.getDefaultTimeout(),
+ new FixedDelayRestartStrategy(1, 1000));
+
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
assertEquals(JobStatus.CREATED, eg.getState());
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
index ad30b9e..35ab8ac 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphTestUtils.java
@@ -30,6 +30,7 @@ import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.akka.AkkaUtils;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.instance.BaseTestingActorGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
@@ -173,7 +174,8 @@ public class ExecutionGraphTestUtils {
new JobID(),
"test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
ExecutionJobVertex ejv = spy(new ExecutionJobVertex(graph, ajv, 1,
AkkaUtils.getDefaultTimeout()));
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
index 89b82f3..9c520b6 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionStateProgressTest.java
@@ -26,6 +26,7 @@ import java.util.Arrays;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.instance.SimpleSlot;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.api.common.JobID;
@@ -52,7 +53,8 @@ public class ExecutionStateProgressTest {
jid,
"test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
graph.attachJobGraph(Arrays.asList(ajv));
@@ -82,4 +84,4 @@ public class ExecutionStateProgressTest {
fail(e.getMessage());
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
index 2530a53..e0da2c9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/LocalInputSplitsTest.java
@@ -30,6 +30,7 @@ import org.apache.flink.core.io.InputSplitAssigner;
import org.apache.flink.core.io.InputSplitSource;
import org.apache.flink.core.io.LocatableInputSplit;
import org.apache.flink.runtime.JobException;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
import org.apache.flink.runtime.instance.InstanceConnectionInfo;
@@ -270,7 +271,8 @@ public class LocalInputSplitsTest {
jobGraph.getJobID(),
jobGraph.getName(),
jobGraph.getJobConfiguration(),
- TIMEOUT);
+ TIMEOUT,
+ new NoRestartStrategy());
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
eg.setQueuedSchedulingAllowed(false);
@@ -333,7 +335,8 @@ public class LocalInputSplitsTest {
jobGraph.getJobID(),
jobGraph.getName(),
jobGraph.getJobConfiguration(),
- TIMEOUT);
+ TIMEOUT,
+ new NoRestartStrategy());
eg.setQueuedSchedulingAllowed(false);
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources());
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
index 3cedb63..eda9115 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/PointwisePatternTest.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.testingUtils.TestingUtils;
import org.junit.Test;
@@ -62,7 +63,8 @@ public class PointwisePatternTest {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
try {
eg.attachJobGraph(ordered);
}
@@ -102,7 +104,8 @@ public class PointwisePatternTest {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
try {
eg.attachJobGraph(ordered);
}
@@ -143,7 +146,8 @@ public class PointwisePatternTest {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
try {
eg.attachJobGraph(ordered);
}
@@ -185,7 +189,8 @@ public class PointwisePatternTest {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
try {
eg.attachJobGraph(ordered);
}
@@ -225,7 +230,8 @@ public class PointwisePatternTest {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
try {
eg.attachJobGraph(ordered);
}
@@ -285,7 +291,8 @@ public class PointwisePatternTest {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
try {
eg.attachJobGraph(ordered);
}
@@ -336,7 +343,8 @@ public class PointwisePatternTest {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
try {
eg.attachJobGraph(ordered);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
index f42543f..2f0d5e7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/TerminalStateDeadlockTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.executiongraph;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.restart.FixedDelayRestartStrategy;
import org.apache.flink.runtime.instance.DummyActorGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
@@ -124,9 +125,7 @@ public class TerminalStateDeadlockTest {
for (int i = 0; i < 20000; i++) {
final TestExecGraph eg = new TestExecGraph(jobId);
eg.attachJobGraph(vertices);
- eg.setDelayBeforeRetrying(0);
- eg.setNumberOfRetriesLeft(1);
-
+
final Execution e1 = eg.getJobVertex(vid1).getTaskVertices()[0].getCurrentExecutionAttempt();
final Execution e2 = eg.getJobVertex(vid2).getTaskVertices()[0].getCurrentExecutionAttempt();
@@ -181,7 +180,13 @@ public class TerminalStateDeadlockTest {
private volatile boolean done;
TestExecGraph(JobID jobId) {
- super(TestingUtils.defaultExecutionContext(), jobId, "test graph", EMPTY_CONFIG, TIMEOUT);
+ super(
+ TestingUtils.defaultExecutionContext(),
+ jobId,
+ "test graph",
+ EMPTY_CONFIG,
+ TIMEOUT,
+ new FixedDelayRestartStrategy(1, 0));
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
index 8604b63..3670358 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexLocationConstraintTest.java
@@ -26,6 +26,7 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.TimeUnit;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.instance.DummyActorGateway;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.Instance;
@@ -80,7 +81,8 @@ public class VertexLocationConstraintTest {
jg.getJobID(),
jg.getName(),
jg.getJobConfiguration(),
- timeout);
+ timeout,
+ new NoRestartStrategy());
eg.attachJobGraph(Collections.singletonList(jobVertex));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex.getID());
@@ -151,7 +153,8 @@ public class VertexLocationConstraintTest {
jg.getJobID(),
jg.getName(),
jg.getJobConfiguration(),
- timeout);
+ timeout,
+ new NoRestartStrategy());
eg.attachJobGraph(Collections.singletonList(jobVertex));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex.getID());
@@ -226,7 +229,8 @@ public class VertexLocationConstraintTest {
jg.getJobID(),
jg.getName(),
jg.getJobConfiguration(),
- timeout);
+ timeout,
+ new NoRestartStrategy());
eg.attachJobGraph(Arrays.asList(jobVertex1, jobVertex2));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex1.getID());
@@ -292,7 +296,8 @@ public class VertexLocationConstraintTest {
jg.getJobID(),
jg.getName(),
jg.getJobConfiguration(),
- timeout);
+ timeout,
+ new NoRestartStrategy());
eg.attachJobGraph(Collections.singletonList(jobVertex));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex.getID());
@@ -360,7 +365,8 @@ public class VertexLocationConstraintTest {
jg.getJobID(),
jg.getName(),
jg.getJobConfiguration(),
- timeout);
+ timeout,
+ new NoRestartStrategy());
eg.attachJobGraph(Arrays.asList(jobVertex1, jobVertex2));
ExecutionJobVertex ejv = eg.getAllVertices().get(jobVertex1.getID());
@@ -398,7 +404,8 @@ public class VertexLocationConstraintTest {
jg.getJobID(),
jg.getName(),
jg.getJobConfiguration(),
- timeout);
+ timeout,
+ new NoRestartStrategy());
eg.attachJobGraph(Collections.singletonList(vertex));
ExecutionVertex ev = eg.getAllVertices().get(vertex.getID()).getTaskVertices()[0];
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
index d9e422c..5c7297e 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/VertexSlotSharingTest.java
@@ -26,6 +26,7 @@ import java.util.List;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.akka.AkkaUtils;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.api.common.JobID;
@@ -74,7 +75,8 @@ public class VertexSlotSharingTest {
new JobID(),
"test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy());
eg.attachJobGraph(vertices);
// verify that the vertices are all in the same slot sharing group
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategyTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategyTest.java
new file mode 100644
index 0000000..d19060a
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/restart/FixedDelayRestartStrategyTest.java
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.executiongraph.restart;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.junit.Test;
+import org.mockito.Mockito;
+import scala.concurrent.ExecutionContext$;
+
+public class FixedDelayRestartStrategyTest {
+
+ @Test
+ public void testFixedDelayRestartStrategy() {
+ int numberRestarts = 10;
+ long restartDelay = 10;
+
+ FixedDelayRestartStrategy fixedDelayRestartStrategy = new FixedDelayRestartStrategy(
+ numberRestarts,
+ restartDelay);
+
+ ExecutionGraph executionGraph = mock(ExecutionGraph.class);
+ when(executionGraph.getExecutionContext())
+ .thenReturn(ExecutionContext$.MODULE$.fromExecutor(MoreExecutors.directExecutor()));
+
+ while(fixedDelayRestartStrategy.canRestart()) {
+ fixedDelayRestartStrategy.restart(executionGraph);
+ }
+
+ Mockito.verify(executionGraph, Mockito.times(numberRestarts)).restart();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
index 6520066..8750a1a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/partition/PipelinedSubpartitionTest.java
@@ -18,14 +18,11 @@
package org.apache.flink.runtime.io.network.partition;
-import org.apache.flink.api.common.JobID;
import org.apache.flink.core.memory.MemorySegment;
import org.apache.flink.runtime.event.AbstractEvent;
-import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
import org.apache.flink.runtime.io.network.buffer.Buffer;
import org.apache.flink.runtime.io.network.buffer.BufferProvider;
import org.apache.flink.runtime.io.network.partition.consumer.BufferOrEvent;
-import org.apache.flink.runtime.io.network.util.TestBufferFactory;
import org.apache.flink.runtime.io.network.util.TestConsumerCallback;
import org.apache.flink.runtime.io.network.util.TestNotificationListener;
import org.apache.flink.runtime.io.network.util.TestPooledBufferProvider;
@@ -39,7 +36,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import static org.apache.flink.runtime.io.disk.iomanager.IOManager.IOMode.ASYNC;
import static org.apache.flink.runtime.io.network.util.TestBufferFactory.createBuffer;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
index 73c7646..fe35c0d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/leaderelection/JobManagerLeaderElectionTest.java
@@ -35,6 +35,7 @@ import org.apache.flink.runtime.checkpoint.SavepointStore;
import org.apache.flink.runtime.checkpoint.SavepointStoreFactory;
import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory;
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.instance.InstanceManager;
import org.apache.flink.runtime.jobmanager.RecoveryMode;
import org.apache.flink.runtime.jobmanager.StandaloneSubmittedJobGraphStore;
@@ -191,13 +192,13 @@ public class JobManagerLeaderElectionTest extends TestLogger {
new Scheduler(TestingUtils.defaultExecutionContext()),
new BlobLibraryCacheManager(new BlobServer(configuration), 10L),
ActorRef.noSender(),
- 1,
- 1L,
+ new NoRestartStrategy(),
AkkaUtils.getDefaultTimeout(),
leaderElectionService,
submittedJobGraphStore,
checkpointRecoveryFactory,
- savepointStore
+ savepointStore,
+ AkkaUtils.getDefaultTimeout()
);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
index 75569ec..da3c02d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/testutils/ZooKeeperTestUtils.java
@@ -88,7 +88,7 @@ public class ZooKeeperTestUtils {
config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, "6 s");
config.setInteger(ConfigConstants.AKKA_WATCH_THRESHOLD, 9);
config.setString(ConfigConstants.AKKA_ASK_TIMEOUT, "100 s");
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "10 s");
+ config.setString(ConfigConstants.RECOVERY_JOB_DELAY, "10 s");
return config;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
index 177dc85..dbd87d0 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/executiongraph/TaskManagerLossFailsTasksTest.scala
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.JobID
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.akka.AkkaUtils
import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.SimpleActorGateway
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy
import org.apache.flink.runtime.jobgraph.{JobStatus, JobGraph, JobVertex}
import org.apache.flink.runtime.jobmanager.Tasks
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler
@@ -56,8 +57,9 @@ class TaskManagerLossFailsTasksTest extends WordSpecLike with Matchers {
new JobID(),
"test job",
new Configuration(),
- AkkaUtils.getDefaultTimeout)
- eg.setNumberOfRetriesLeft(0)
+ AkkaUtils.getDefaultTimeout,
+ new NoRestartStrategy())
+
eg.attachJobGraph(jobGraph.getVerticesSortedTopologicallyFromSources)
eg.getState should equal(JobStatus.CREATED)
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
index 68923fe..41b6702 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/jobmanager/RecoveryITCase.scala
@@ -33,6 +33,8 @@ import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
import org.scalatest.junit.JUnitRunner
import scala.concurrent.duration._
+import language.postfixOps
+
@RunWith(classOf[JUnitRunner])
class RecoveryITCase(_system: ActorSystem)
extends TestKit(_system)
@@ -57,7 +59,9 @@ class RecoveryITCase(_system: ActorSystem)
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlots)
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, numTaskManagers)
config.setString(ConfigConstants.AKKA_WATCH_HEARTBEAT_PAUSE, heartbeatTimeout)
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, heartbeatTimeout)
+ config.setString(ConfigConstants.RESTART_STRATEGY, "fixeddelay")
+ config.setInteger(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_ATTEMPTS, 1)
+ config.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, heartbeatTimeout)
new TestingCluster(config)
}
@@ -79,7 +83,6 @@ class RecoveryITCase(_system: ActorSystem)
receiver.connectNewDataSetAsInput(sender, DistributionPattern.POINTWISE)
val jobGraph = new JobGraph("Pointwise job", sender, receiver)
- jobGraph.setNumberOfExecutionRetries(1)
val cluster = createTestClusterWithHeartbeatTimeout(2 * NUM_TASKS, 1, "2 s")
cluster.start()
@@ -124,7 +127,6 @@ class RecoveryITCase(_system: ActorSystem)
receiver.setSlotSharingGroup(sharingGroup)
val jobGraph = new JobGraph("Pointwise job", sender, receiver)
- jobGraph.setNumberOfExecutionRetries(1)
val cluster = createTestClusterWithHeartbeatTimeout(NUM_TASKS, 1, "2 s")
cluster.start()
@@ -169,7 +171,6 @@ class RecoveryITCase(_system: ActorSystem)
receiver.setSlotSharingGroup(sharingGroup)
val jobGraph = new JobGraph("Pointwise job", sender, receiver)
- jobGraph.setNumberOfExecutionRetries(1)
val cluster = createTestClusterWithHeartbeatTimeout(NUM_TASKS, 2, "2 s")
cluster.start()
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
index cfb1192..7b7750a 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingCluster.scala
@@ -96,14 +96,14 @@ class TestingCluster(
instanceManager,
scheduler,
libraryCacheManager,
- executionRetries,
- delayBetweenRetries,
+ restartStrategy,
timeout,
archiveCount,
leaderElectionService,
submittedJobsGraphs,
checkpointRecoveryFactory,
- savepointStore) = JobManager.createJobManagerComponents(
+ savepointStore,
+ jobRecoveryTimeout) = JobManager.createJobManagerComponents(
config,
createLeaderElectionService())
@@ -118,13 +118,13 @@ class TestingCluster(
scheduler,
libraryCacheManager,
archive,
- executionRetries,
- delayBetweenRetries,
+ restartStrategy,
timeout,
leaderElectionService,
submittedJobsGraphs,
checkpointRecoveryFactory,
- savepointStore))
+ savepointStore,
+ jobRecoveryTimeout))
val dispatcherJobManagerProps = if (synchronousDispatcher) {
// disable asynchronous futures (e.g. accumulator update in Heartbeat)
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
index 98d8863..53867e0 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingJobManager.scala
@@ -23,6 +23,7 @@ import akka.actor.ActorRef
import org.apache.flink.configuration.Configuration
import org.apache.flink.runtime.checkpoint.{SavepointStore, CheckpointRecoveryFactory}
import org.apache.flink.runtime.execution.librarycache.BlobLibraryCacheManager
+import org.apache.flink.runtime.executiongraph.restart.RestartStrategy
import org.apache.flink.runtime.instance.InstanceManager
import org.apache.flink.runtime.jobmanager.scheduler.Scheduler
import org.apache.flink.runtime.jobmanager.{JobManager, SubmittedJobGraphStore}
@@ -43,13 +44,13 @@ class TestingJobManager(
scheduler: Scheduler,
libraryCacheManager: BlobLibraryCacheManager,
archive: ActorRef,
- defaultExecutionRetries: Int,
- delayBetweenRetries: Long,
+ restartStrategy: RestartStrategy,
timeout: FiniteDuration,
leaderElectionService: LeaderElectionService,
submittedJobGraphs : SubmittedJobGraphStore,
checkpointRecoveryFactory : CheckpointRecoveryFactory,
- savepointStore : SavepointStore)
+ savepointStore : SavepointStore,
+ jobRecoveryTimeout: FiniteDuration)
extends JobManager(
flinkConfiguration,
executorService,
@@ -57,11 +58,11 @@ class TestingJobManager(
scheduler,
libraryCacheManager,
archive,
- defaultExecutionRetries,
- delayBetweenRetries,
+ restartStrategy,
timeout,
leaderElectionService,
submittedJobGraphs,
checkpointRecoveryFactory,
- savepointStore)
+ savepointStore,
+ jobRecoveryTimeout)
with TestingJobManagerLike {}
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
index 6057f65..d682112 100644
--- a/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
+++ b/flink-runtime/src/test/scala/org/apache/flink/runtime/testingUtils/TestingUtils.scala
@@ -323,14 +323,14 @@ object TestingUtils {
instanceManager,
scheduler,
libraryCacheManager,
- executionRetries,
- delayBetweenRetries,
+ restartStrategy,
timeout,
archiveCount,
leaderElectionService,
submittedJobGraphs,
checkpointRecoveryFactory,
- savepointStore) = JobManager.createJobManagerComponents(
+ savepointStore,
+ jobRecoveryTimeout) = JobManager.createJobManagerComponents(
configuration,
None
)
@@ -347,12 +347,12 @@ object TestingUtils {
scheduler,
libraryCacheManager,
archive,
- executionRetries,
- delayBetweenRetries,
+ restartStrategy,
timeout,
leaderElectionService,
submittedJobGraphs,
- checkpointRecoveryFactory)
+ checkpointRecoveryFactory,
+ jobRecoveryTimeout)
val jobManager: ActorRef = actorSystem.actorOf(jobManagerProps, JobManager.JOB_MANAGER_NAME)
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
index b3d2430..5c923cb 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/ExecutionEnvironment.scala
@@ -21,6 +21,7 @@ import com.esotericsoftware.kryo.Serializer
import com.google.common.base.Preconditions
import org.apache.flink.annotation.{PublicEvolving, Public}
import org.apache.flink.api.common.io.{FileInputFormat, InputFormat}
+import org.apache.flink.api.common.restartstrategy.RestartStrategies.RestartStrategyConfiguration
import org.apache.flink.api.common.typeinfo.{BasicTypeInfo, TypeInformation}
import org.apache.flink.api.common.typeutils.CompositeType
import org.apache.flink.api.common.{ExecutionConfig, JobExecutionResult, JobID}
@@ -93,20 +94,49 @@ class ExecutionEnvironment(javaEnv: JavaEnv) {
def getParallelism = javaEnv.getParallelism
/**
- * Sets the number of times that failed tasks are re-executed. A value of zero
- * effectively disables fault tolerance. A value of "-1" indicates that the system
- * default value (as defined in the configuration) should be used.
- */
+ * Sets the restart strategy configuration. The configuration specifies which restart strategy
+ * will be used for the execution graph in case of a restart.
+ *
+ * @param restartStrategyConfiguration Restart strategy configuration to be set
+ */
+ @PublicEvolving
+ def setRestartStrategy(restartStrategyConfiguration: RestartStrategyConfiguration): Unit = {
+ javaEnv.setRestartStrategy(restartStrategyConfiguration)
+ }
+
+ /**
+ * Returns the specified restart strategy configuration.
+ *
+ * @return The restart strategy configuration to be used
+ */
+ @PublicEvolving
+ def getRestartStrategy: RestartStrategyConfiguration = {
+ javaEnv.getRestartStrategy()
+ }
+
+ /**
+ * Sets the number of times that failed tasks are re-executed. A value of zero
+ * effectively disables fault tolerance. A value of "-1" indicates that the system
+ * default value (as defined in the configuration) should be used.
+ *
+ * @deprecated This method will be replaced by [[setRestartStrategy()]]. The
+ * FixedDelayRestartStrategyConfiguration contains the number of execution retries.
+ */
+ @Deprecated
@PublicEvolving
def setNumberOfExecutionRetries(numRetries: Int): Unit = {
javaEnv.setNumberOfExecutionRetries(numRetries)
}
/**
- * Gets the number of times the system will try to re-execute failed tasks. A value
- * of "-1" indicates that the system default value (as defined in the configuration)
- * should be used.
- */
+ * Gets the number of times the system will try to re-execute failed tasks. A value
+ * of "-1" indicates that the system default value (as defined in the configuration)
+ * should be used.
+ *
+ * @deprecated This method will be replaced by [[getRestartStrategy]]. The
+ * FixedDelayRestartStrategyConfiguration contains the number of execution retries.
+ */
+ @Deprecated
@PublicEvolving
def getNumberOfExecutionRetries = javaEnv.getNumberOfExecutionRetries
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java b/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
index a3e815e..a0caadd 100644
--- a/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-0.8/src/test/java/org/apache/flink/streaming/connectors/kafka/Kafka08ITCase.java
@@ -18,22 +18,15 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.curator.framework.CuratorFramework;
-import org.apache.flink.api.java.functions.FlatMapIterator;
-import org.apache.flink.api.java.operators.DataSource;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.internals.ZookeeperOffsetHandler;
import org.junit.Test;
-import java.util.Iterator;
import java.util.Properties;
import static org.junit.Assert.assertTrue;
-
public class Kafka08ITCase extends KafkaConsumerTestBase {
// ------------------------------------------------------------------------
@@ -166,19 +159,19 @@ public class Kafka08ITCase extends KafkaConsumerTestBase {
StreamExecutionEnvironment env1 = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env1.getConfig().disableSysoutLogging();
env1.enableCheckpointing(50);
- env1.setNumberOfExecutionRetries(0);
+ env1.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env1.setParallelism(parallelism);
StreamExecutionEnvironment env2 = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env2.getConfig().disableSysoutLogging();
env2.enableCheckpointing(50);
- env2.setNumberOfExecutionRetries(0);
+ env2.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env2.setParallelism(parallelism);
StreamExecutionEnvironment env3 = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env3.getConfig().disableSysoutLogging();
env3.enableCheckpointing(50);
- env3.setNumberOfExecutionRetries(0);
+ env3.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env3.setParallelism(parallelism);
// write a sequence from 0 to 99 to each of the 3 partitions.
@@ -222,13 +215,13 @@ public class Kafka08ITCase extends KafkaConsumerTestBase {
StreamExecutionEnvironment env1 = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env1.getConfig().disableSysoutLogging();
- env1.setNumberOfExecutionRetries(0);
+ env1.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env1.setParallelism(parallelism);
StreamExecutionEnvironment env2 = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
// NOTE: We are not enabling the checkpointing!
env2.getConfig().disableSysoutLogging();
- env2.setNumberOfExecutionRetries(0);
+ env2.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env2.setParallelism(parallelism);
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
index 680e4ec..7c8cdb2 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaConsumerTestBase.java
@@ -33,6 +33,7 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.TypeSerializer;
@@ -136,7 +137,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment see = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
see.getConfig().disableSysoutLogging();
- see.setNumberOfExecutionRetries(0);
+ see.setRestartStrategy(RestartStrategies.noRestart());
see.setParallelism(1);
// use wrong ports for the consumers
@@ -255,7 +256,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.setParallelism(parallelism);
env.enableCheckpointing(500);
- env.setNumberOfExecutionRetries(0); // fail immediately
+ env.setRestartStrategy(RestartStrategies.noRestart()); // fail immediately
env.getConfig().disableSysoutLogging();
TypeInformation<Tuple2<Long, String>> longStringType = TypeInfoParser.parse("Tuple2<Long, String>");
@@ -385,7 +386,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.enableCheckpointing(500);
env.setParallelism(parallelism);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000));
env.getConfig().disableSysoutLogging();
FlinkKafkaConsumerBase<Integer> kafkaSource = kafkaServer.getConsumer(topic, schema, standardProps);
@@ -430,7 +431,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.enableCheckpointing(500);
env.setParallelism(parallelism);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000));
env.getConfig().disableSysoutLogging();
FlinkKafkaConsumerBase<Integer> kafkaSource = kafkaServer.getConsumer(topic, schema, standardProps);
@@ -475,7 +476,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.enableCheckpointing(500);
env.setParallelism(parallelism);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000));
env.getConfig().disableSysoutLogging();
env.setBufferTimeout(0);
@@ -781,7 +782,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
new TypeInformationSerializationSchema<>(longBytesInfo, new ExecutionConfig());
final StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
- env.setNumberOfExecutionRetries(0);
+ env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
env.enableCheckpointing(100);
env.setParallelism(parallelism);
@@ -895,7 +896,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.setParallelism(parallelism);
env.enableCheckpointing(500);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000));
env.getConfig().disableSysoutLogging();
@@ -923,7 +924,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.setParallelism(1);
- env.setNumberOfExecutionRetries(0);
+ env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
DataStream<Tuple2<Long, PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<Long, PojoValue>>() {
@@ -953,7 +954,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.setParallelism(1);
- env.setNumberOfExecutionRetries(0);
+ env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
@@ -1005,7 +1006,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.setParallelism(1);
- env.setNumberOfExecutionRetries(0);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
DataStream<Tuple2<byte[], PojoValue>> kvStream = env.addSource(new SourceFunction<Tuple2<byte[], PojoValue>>() {
@@ -1038,7 +1039,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.setParallelism(1);
- env.setNumberOfExecutionRetries(0);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
DataStream<Tuple2<byte[], PojoValue>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, schema, standardProps));
@@ -1076,7 +1077,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
// write some data
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env.setParallelism(1);
- env.setNumberOfExecutionRetries(0);
+ env.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
writeSequence(env, topic, ELEMENT_COUNT, 1);
@@ -1084,7 +1085,7 @@ public abstract class KafkaConsumerTestBase extends KafkaTestBase {
// read using custom schema
final StreamExecutionEnvironment env1 = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
env1.setParallelism(1);
- env1.setNumberOfExecutionRetries(0);
+ env1.getConfig().setRestartStrategy(RestartStrategies.noRestart());
env1.getConfig().disableSysoutLogging();
DataStream<Tuple2<Integer, Integer>> fromKafka = env.addSource(kafkaServer.getConsumer(topic, new FixedNumberDeserializationSchema(ELEMENT_COUNT), standardProps));
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
index 228f3ac..e251174 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.connectors.kafka;
import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeInfoParser;
@@ -70,7 +71,7 @@ public abstract class KafkaProducerTestBase extends KafkaTestBase {
TypeInformation<Tuple2<Long, String>> longStringInfo = TypeInfoParser.parse("Tuple2<Long, String>");
StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment("localhost", flinkPort);
- env.setNumberOfExecutionRetries(0);
+ env.setRestartStrategy(RestartStrategies.noRestart());
env.getConfig().disableSysoutLogging();
TypeInformationSerializationSchema<Tuple2<Long, String>> serSchema =
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
index ab1d5b6..1108e2f 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaTestBase.java
@@ -104,7 +104,7 @@ public abstract class KafkaTestBase extends TestLogger {
flinkConfig.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 1);
flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, 8);
flinkConfig.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 16);
- flinkConfig.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "0 s");
+ flinkConfig.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "0 s");
flink = new ForkableFlinkMiniCluster(flinkConfig, false);
flink.start();
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
----------------------------------------------------------------------
diff --git a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
index cc237b6..c489a27 100644
--- a/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
+++ b/flink-streaming-connectors/flink-connector-kafka-base/src/test/java/org/apache/flink/streaming/connectors/kafka/testutils/DataGenerators.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.connectors.kafka.testutils;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -48,7 +49,7 @@ public class DataGenerators {
env.setParallelism(numPartitions);
env.getConfig().disableSysoutLogging();
- env.setNumberOfExecutionRetries(0);
+ env.setRestartStrategy(RestartStrategies.noRestart());
DataStream<Tuple2<Integer, Integer>> stream =env.addSource(
new RichParallelSourceFunction<Tuple2<Integer, Integer>>() {
@@ -90,7 +91,7 @@ public class DataGenerators {
final boolean randomizeOrder) throws Exception {
env.setParallelism(numPartitions);
env.getConfig().disableSysoutLogging();
- env.setNumberOfExecutionRetries(0);
+ env.setRestartStrategy(RestartStrategies.noRestart());
DataStream<Integer> stream = env.addSource(
new RichParallelSourceFunction<Integer>() {
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index 8d829c6..82e536c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.functions.InvalidTypesException;
import org.apache.flink.api.common.io.FileInputFormat;
import org.apache.flink.api.common.io.InputFormat;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -424,6 +425,27 @@ public abstract class StreamExecutionEnvironment {
}
/**
+ * Sets the restart strategy configuration. The configuration specifies which restart strategy
+ * will be used for the execution graph in case of a restart.
+ *
+ * @param restartStrategyConfiguration Restart strategy configuration to be set
+ */
+ @PublicEvolving
+ public void setRestartStrategy(RestartStrategies.RestartStrategyConfiguration restartStrategyConfiguration) {
+ config.setRestartStrategy(restartStrategyConfiguration);
+ }
+
+ /**
+ * Returns the specified restart strategy configuration.
+ *
+ * @return The restart strategy configuration to be used
+ */
+ @PublicEvolving
+ public RestartStrategies.RestartStrategyConfiguration getRestartStrategy() {
+ return config.getRestartStrategy();
+ }
+
+ /**
* Sets the number of times that failed tasks are re-executed. A value of
* zero effectively disables fault tolerance. A value of {@code -1}
* indicates that the system default value (as defined in the configuration)
@@ -431,7 +453,12 @@ public abstract class StreamExecutionEnvironment {
*
* @param numberOfExecutionRetries
* The number of times the system will try to re-execute failed tasks.
+ *
+ * @deprecated This method will be replaced by {@link #setRestartStrategy}. The
+ * {@link RestartStrategies.FixedDelayRestartStrategyConfiguration} contains the number of
+ * execution retries.
*/
+ @Deprecated
@PublicEvolving
public void setNumberOfExecutionRetries(int numberOfExecutionRetries) {
config.setNumberOfExecutionRetries(numberOfExecutionRetries);
@@ -443,7 +470,12 @@ public abstract class StreamExecutionEnvironment {
* in the configuration) should be used.
*
* @return The number of times the system will try to re-execute failed tasks.
+ *
+ * @deprecated This method will be replaced by {@link #getRestartStrategy}. The
+ * {@link RestartStrategies.FixedDelayRestartStrategyConfiguration} contains the number of
+ * execution retries.
*/
+ @Deprecated
@PublicEvolving
public int getNumberOfExecutionRetries() {
return config.getNumberOfExecutionRetries();
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
index 04b19174..a21e3b7 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGenerator.java
@@ -24,6 +24,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.flink.annotation.Internal;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.operators.util.UserCodeObjectWrapper;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
@@ -75,6 +76,12 @@ public class StreamingJobGraphGenerator {
private static final Logger LOG = LoggerFactory.getLogger(StreamingJobGraphGenerator.class);
+ /**
+ * Restart delay used for the FixedDelayRestartStrategy in case checkpointing was enabled but
+ * no restart strategy has been specified.
+ */
+ private static final long DEFAULT_RESTART_DELAY = 10000L;
+
private StreamGraph streamGraph;
private Map<Integer, JobVertex> jobVertices;
@@ -121,9 +128,7 @@ public class StreamingJobGraphGenerator {
configureCheckpointing();
- configureExecutionRetries();
-
- configureExecutionRetryDelay();
+ configureRestartStrategy();
try {
InstantiationUtil.writeObjectToConfig(this.streamGraph.getExecutionConfig(), this.jobGraph.getJobConfiguration(), ExecutionConfig.CONFIG_KEY);
@@ -477,22 +482,17 @@ public class StreamingJobGraphGenerator {
cfg.getMaxConcurrentCheckpoints());
jobGraph.setSnapshotSettings(settings);
- // if the user enabled checkpointing, the default number of exec retries is infinitive.
- int executionRetries = streamGraph.getExecutionConfig().getNumberOfExecutionRetries();
- if(executionRetries == -1) {
- streamGraph.getExecutionConfig().setNumberOfExecutionRetries(Integer.MAX_VALUE);
+ // check if a restart strategy has been set, if not then set the FixedDelayRestartStrategy
+ if (streamGraph.getExecutionConfig().getRestartStrategy() == null) {
+ // if the user enabled checkpointing, the default number of exec retries is infinitive.
+ streamGraph.getExecutionConfig().setRestartStrategy(
+ RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, DEFAULT_RESTART_DELAY));
}
}
}
- private void configureExecutionRetries() {
- int executionRetries = streamGraph.getExecutionConfig().getNumberOfExecutionRetries();
- jobGraph.setNumberOfExecutionRetries(executionRetries);
- }
-
- private void configureExecutionRetryDelay() {
- long executionRetryDelay = streamGraph.getExecutionConfig().getExecutionRetryDelay();
- jobGraph.setExecutionRetryDelay(executionRetryDelay);
+ private void configureRestartStrategy() {
+ jobGraph.setRestartStrategyConfiguration(streamGraph.getExecutionConfig().getRestartStrategy());
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
index bac7fa5..184f87e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/partitioner/RescalePartitionerTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.runtime.executiongraph.ExecutionEdge;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.plugable.SerializationDelegate;
@@ -133,7 +134,11 @@ public class RescalePartitionerTest extends TestLogger {
jobId,
jobName,
cfg,
- AkkaUtils.getDefaultTimeout(),new ArrayList<BlobKey>(), new ArrayList<URL>(), ExecutionGraph.class.getClassLoader());
+ AkkaUtils.getDefaultTimeout(),
+ new NoRestartStrategy(),
+ new ArrayList<BlobKey>(),
+ new ArrayList<URL>(),
+ ExecutionGraph.class.getClassLoader());
try {
eg.attachJobGraph(jobVertices);
}
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/state/StateBackendITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/state/StateBackendITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/state/StateBackendITCase.java
index 3aa60dc..222924c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/state/StateBackendITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/state/StateBackendITCase.java
@@ -20,6 +20,7 @@ package org.apache.flink.streaming.runtime.state;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.FoldingState;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.FoldingStateDescriptor;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
@@ -56,7 +57,7 @@ public class StateBackendITCase extends StreamingMultipleProgramsTestBase {
StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
see.setParallelism(1);
- see.setNumberOfExecutionRetries(0);
+ see.getConfig().setRestartStrategy(RestartStrategies.noRestart());
see.setStateBackend(new FailingStateBackend());
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
index 0e03b71..eed455c 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/timestamp/TimestampITCase.java
@@ -79,7 +79,6 @@ public class TimestampITCase {
Configuration config = new Configuration();
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TASK_MANAGERS);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_TASK_SLOTS);
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "0 ms");
config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 12);
cluster = new ForkableFlinkMiniCluster(config, false);
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
index 58b100e..39dd6cc 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -21,6 +21,7 @@ package org.apache.flink.streaming.api.scala
import com.esotericsoftware.kryo.Serializer
import org.apache.flink.annotation.{Internal, PublicEvolving, Public}
import org.apache.flink.api.common.io.{FileInputFormat, InputFormat}
+import org.apache.flink.api.common.restartstrategy.RestartStrategies.RestartStrategyConfiguration
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
import org.apache.flink.api.scala.ClosureCleaner
@@ -186,8 +187,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
* operator states. Time interval between state checkpoints is specified in in millis.
*
* Setting this option assumes that the job is used in production and thus if not stated
- * explicitly otherwise with calling with the
- * [[setNumberOfExecutionRetries(int)]] method in case of
+ * explicitly otherwise with calling the [[setRestartStrategy]] method in case of
* failure the job will be resubmitted to the cluster indefinitely.
*/
@deprecated
@@ -229,22 +229,49 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
*/
@PublicEvolving
def getStateBackend: AbstractStateBackend = javaEnv.getStateBackend()
-
+
/**
- * Sets the number of times that failed tasks are re-executed. A value of zero
- * effectively disables fault tolerance. A value of "-1" indicates that the system
- * default value (as defined in the configuration) should be used.
- */
+ * Sets the restart strategy configuration. The configuration specifies which restart strategy
+ * will be used for the execution graph in case of a restart.
+ *
+ * @param restartStrategyConfiguration Restart strategy configuration to be set
+ */
+ @PublicEvolving
+ def setRestartStrategy(restartStrategyConfiguration: RestartStrategyConfiguration): Unit = {
+ javaEnv.setRestartStrategy(restartStrategyConfiguration)
+ }
+
+ /**
+ * Returns the specified restart strategy configuration.
+ *
+ * @return The restart strategy configuration to be used
+ */
+ @PublicEvolving
+ def getRestartStrategy: RestartStrategyConfiguration = {
+ javaEnv.getRestartStrategy()
+ }
+
+ /**
+ * Sets the number of times that failed tasks are re-executed. A value of zero
+ * effectively disables fault tolerance. A value of "-1" indicates that the system
+ * default value (as defined in the configuration) should be used.
+ *
+ * @deprecated This method will be replaced by [[setRestartStrategy()]]. The
+ * FixedDelayRestartStrategyConfiguration contains the number of execution retries.
+ */
@PublicEvolving
def setNumberOfExecutionRetries(numRetries: Int): Unit = {
javaEnv.setNumberOfExecutionRetries(numRetries)
}
/**
- * Gets the number of times the system will try to re-execute failed tasks. A value
- * of "-1" indicates that the system default value (as defined in the configuration)
- * should be used.
- */
+ * Gets the number of times the system will try to re-execute failed tasks. A value
+ * of "-1" indicates that the system default value (as defined in the configuration)
+ * should be used.
+ *
+ * @deprecated This method will be replaced by [[getRestartStrategy]]. The
+ * FixedDelayRestartStrategyConfiguration contains the number of execution retries.
+ */
@PublicEvolving
def getNumberOfExecutionRetries = javaEnv.getNumberOfExecutionRetries
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
index 9eca074..9d2f062 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.test.checkpointing;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.ConfigConstants;
@@ -67,7 +68,6 @@ public class EventTimeAllWindowCheckpointingITCase extends TestLogger {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, PARALLELISM / 2);
config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 48);
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "0 ms");
cluster = new ForkableFlinkMiniCluster(config, false);
cluster.start();
@@ -96,7 +96,7 @@ public class EventTimeAllWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env
@@ -160,7 +160,7 @@ public class EventTimeAllWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env
@@ -221,7 +221,7 @@ public class EventTimeAllWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env
@@ -289,7 +289,7 @@ public class EventTimeAllWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
index d9ced4b..c6a6b28 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.test.checkpointing;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
@@ -70,7 +71,7 @@ import static org.apache.flink.test.util.TestUtils.tryExecute;
import static org.junit.Assert.*;
/**
- * This verfies that checkpointing works correctly with event time windows. This is more
+ * This verifies that checkpointing works correctly with event time windows. This is more
* strict than {@link WindowCheckpointingITCase} because for event-time the contents
* of the emitted windows are deterministic.
*/
@@ -98,7 +99,6 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, PARALLELISM / 2);
config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 48);
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "0 ms");
cluster = new ForkableFlinkMiniCluster(config, false);
cluster.start();
@@ -149,7 +149,7 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env.setStateBackend(this.stateBackend);
@@ -213,7 +213,7 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env.setStateBackend(this.stateBackend);
@@ -282,7 +282,7 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env.setStateBackend(this.stateBackend);
@@ -346,7 +346,7 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env.setStateBackend(this.stateBackend);
@@ -414,7 +414,7 @@ public class EventTimeWindowCheckpointingITCase extends TestLogger {
env.setParallelism(PARALLELISM);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env.setStateBackend(this.stateBackend);
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
index 58f8225..b89f35a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/SavepointITCase.java
@@ -27,6 +27,7 @@ import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.ApplicationID;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.functions.RichMapFunction;
+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;
@@ -170,7 +171,7 @@ public class SavepointITCase extends TestLogger {
LOG.info("JobManager: " + jobManager + ".");
// Submit the job
- final JobGraph jobGraph = createJobGraph(parallelism, 0, 1000);
+ final JobGraph jobGraph = createJobGraph(parallelism, 0, 1000, 1000);
final JobID jobId = jobGraph.getJobID();
// Wait for the source to be notified about the expected number
@@ -463,7 +464,7 @@ public class SavepointITCase extends TestLogger {
LOG.info("JobManager: " + jobManager + ".");
// Submit the job
- final JobGraph jobGraph = createJobGraph(parallelism, 0, 1000);
+ final JobGraph jobGraph = createJobGraph(parallelism, 0, 1000, 1000);
final JobID jobId = jobGraph.getJobID();
// Wait for the source to be notified about the expected number
@@ -599,7 +600,7 @@ public class SavepointITCase extends TestLogger {
LOG.info("JobManager: " + jobManager + ".");
// Submit the job
- final JobGraph jobGraph = createJobGraph(parallelism, 0, 1000);
+ final JobGraph jobGraph = createJobGraph(parallelism, 0, 1000, 1000);
final JobID jobId = jobGraph.getJobID();
// Wait for the source to be notified about the expected number
@@ -704,10 +705,6 @@ public class SavepointITCase extends TestLogger {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, numTaskManagers);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, numSlotsPerTaskManager);
- // Long delay to ensure that the test times out if the job
- // manager tries to restart the job.
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "1 hour");
-
LOG.info("Flink configuration: " + config + ".");
// Start Flink
@@ -725,7 +722,9 @@ public class SavepointITCase extends TestLogger {
// High value to ensure timeouts if restarted.
int numberOfRetries = 1000;
// Submit the job
- final JobGraph jobGraph = createJobGraph(parallelism, numberOfRetries, 1000);
+ // Long delay to ensure that the test times out if the job
+ // manager tries to restart the job.
+ final JobGraph jobGraph = createJobGraph(parallelism, numberOfRetries, 3600000, 1000);
// Set non-existing savepoint path
jobGraph.setSavepointPath("unknown path");
@@ -758,13 +757,14 @@ public class SavepointITCase extends TestLogger {
private JobGraph createJobGraph(
int parallelism,
int numberOfRetries,
+ long restartDelay,
int checkpointingInterval) {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(parallelism);
- env.setNumberOfExecutionRetries(numberOfRetries);
env.enableCheckpointing(checkpointingInterval);
env.disableOperatorChaining();
+ env.getConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(numberOfRetries, restartDelay));
env.getConfig().disableSysoutLogging();
DataStream<Integer> stream = env
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointNotifierITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointNotifierITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointNotifierITCase.java
index 5fa0666..8927f43 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointNotifierITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamCheckpointNotifierITCase.java
@@ -84,7 +84,7 @@ public class StreamCheckpointNotifierITCase {
Configuration config = new Configuration();
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TASK_MANAGERS);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_TASK_SLOTS);
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "0 ms");
+ config.setString(ConfigConstants.RESTART_STRATEGY_FIXED_DELAY_DELAY, "0 ms");
config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 12);
cluster = new ForkableFlinkMiniCluster(config, false);
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamFaultToleranceTestBase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamFaultToleranceTestBase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamFaultToleranceTestBase.java
index a1e1e40..0fffdb5 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamFaultToleranceTestBase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/StreamFaultToleranceTestBase.java
@@ -49,7 +49,6 @@ public abstract class StreamFaultToleranceTestBase extends TestLogger {
Configuration config = new Configuration();
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TASK_MANAGERS);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_TASK_SLOTS);
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "0 ms");
config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 12);
cluster = new ForkableFlinkMiniCluster(config, false);
http://git-wip-us.apache.org/repos/asf/flink/blob/5eae47f5/flink-tests/src/test/java/org/apache/flink/test/checkpointing/WindowCheckpointingITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/WindowCheckpointingITCase.java b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/WindowCheckpointingITCase.java
index c9286ce..eb5ef5a 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/WindowCheckpointingITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/WindowCheckpointingITCase.java
@@ -20,6 +20,7 @@ package org.apache.flink.test.checkpointing;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.client.program.ProgramInvocationException;
@@ -81,7 +82,6 @@ public class WindowCheckpointingITCase extends TestLogger {
config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, 2);
config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, PARALLELISM / 2);
config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 48);
- config.setString(ConfigConstants.EXECUTION_RETRY_DELAY_KEY, "0 ms");
cluster = new ForkableFlinkMiniCluster(config, false);
cluster.start();
@@ -109,7 +109,7 @@ public class WindowCheckpointingITCase extends TestLogger {
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env
@@ -167,7 +167,7 @@ public class WindowCheckpointingITCase extends TestLogger {
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env
@@ -225,7 +225,7 @@ public class WindowCheckpointingITCase extends TestLogger {
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env
@@ -273,7 +273,7 @@ public class WindowCheckpointingITCase extends TestLogger {
env.setStreamTimeCharacteristic(timeCharacteristic);
env.getConfig().setAutoWatermarkInterval(10);
env.enableCheckpointing(100);
- env.setNumberOfExecutionRetries(3);
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 0));
env.getConfig().disableSysoutLogging();
env