zhuzhurk commented on a change in pull request #13641:
URL: https://github.com/apache/flink/pull/13641#discussion_r532462500
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphVariousFailuesTest.java
##########
@@ -20,86 +20,31 @@
import org.apache.flink.api.common.JobStatus;
import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
-import org.apache.flink.runtime.execution.SuppressRestartsException;
-import
org.apache.flink.runtime.executiongraph.restart.InfiniteDelayRestartStrategy;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.scheduler.SchedulerBase;
+import org.apache.flink.runtime.scheduler.SchedulerTestingUtils;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
-
public class ExecutionGraphVariousFailuesTest extends TestLogger {
- /**
- * Test that failing in state restarting will retrigger the restarting
logic. This means that
- * it only goes into the state FAILED after the restart strategy says
the job is no longer
- * restartable.
- */
- @Test
- public void testFailureWhileRestarting() throws Exception {
- final ExecutionGraph eg =
ExecutionGraphTestUtils.createSimpleTestGraph(new
InfiniteDelayRestartStrategy(2));
-
eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread());
- eg.scheduleForExecution();
-
- assertEquals(JobStatus.RUNNING, eg.getState());
- ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
-
- eg.failGlobal(new Exception("Test 1"));
- assertEquals(JobStatus.FAILING, eg.getState());
- ExecutionGraphTestUtils.completeCancellingForAllVertices(eg);
-
- // we should restart since we have two restart attempts left
- assertEquals(JobStatus.RESTARTING, eg.getState());
-
- eg.failGlobal(new Exception("Test 2"));
-
- // we should restart since we have one restart attempts left
- assertEquals(JobStatus.RESTARTING, eg.getState());
-
- eg.failGlobal(new Exception("Test 3"));
-
- // after depleting all our restart attempts we should go into
Failed
- assertEquals(JobStatus.FAILED, eg.getState());
- }
-
- /**
- * Tests that a {@link SuppressRestartsException} in state RESTARTING
stops the restarting
- * immediately and sets the execution graph's state to FAILED.
- */
- @Test
- public void testSuppressRestartFailureWhileRestarting() throws
Exception {
- final ExecutionGraph eg =
ExecutionGraphTestUtils.createSimpleTestGraph(new
InfiniteDelayRestartStrategy(10));
-
eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread());
- eg.scheduleForExecution();
-
- assertEquals(JobStatus.RUNNING, eg.getState());
- ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
-
- eg.failGlobal(new Exception("test"));
- assertEquals(JobStatus.FAILING, eg.getState());
-
- ExecutionGraphTestUtils.completeCancellingForAllVertices(eg);
- assertEquals(JobStatus.RESTARTING, eg.getState());
-
- // suppress a possible restart
- eg.failGlobal(new SuppressRestartsException(new
Exception("Test")));
-
- assertEquals(JobStatus.FAILED, eg.getState());
- }
-
/**
* Tests that a failing scheduleOrUpdateConsumers call with a
non-existing execution attempt
* id, will not fail the execution graph.
*/
@Test
public void testFailingScheduleOrUpdateConsumers() throws Exception {
- final ExecutionGraph eg =
ExecutionGraphTestUtils.createSimpleTestGraph(new
InfiniteDelayRestartStrategy(10));
-
eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread());
- eg.scheduleForExecution();
+ final SchedulerBase scheduler =
SchedulerTestingUtils.newSchedulerBuilder(new JobGraph()).build();
+
scheduler.start(ComponentMainThreadExecutorServiceAdapter.forMainThread());
+ scheduler.startScheduling();
+
+ final ExecutionGraph eg = scheduler.getExecutionGraph();
Review comment:
`scheduleOrUpdateConsumers` is still used by lazy-from-sources
scheduling with schedulerNG. So we still need this test.
It is not needed by pipelined region scheduling though. Actually the
partition cache mechanism is not needed anymore for pipelined region scheduling.
But maybe we still need keep this mechanism for other possible scheduling
strategy which may schedule a task before all its upstream task finishes.
WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
