tillrohrmann commented on a change in pull request #13641: URL: https://github.com/apache/flink/pull/13641#discussion_r540934025
########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/metrics/RestartTimeGaugeTest.java ########## @@ -0,0 +1,83 @@ +/* + * 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.metrics; + +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.runtime.executiongraph.TestingJobStatusProvider; +import org.apache.flink.util.TestLogger; + +import org.junit.Test; + +import java.util.HashMap; +import java.util.Map; + +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.is; +import static org.junit.Assert.assertThat; + +/** + * Tests for {@link RestartTimeGauge}. + */ +public class RestartTimeGaugeTest extends TestLogger { + + @Test + public void testNotRestarted() { + final RestartTimeGauge gauge = new RestartTimeGauge(new TestingJobStatusProvider(JobStatus.RUNNING, -1)); + assertThat(gauge.getValue(), is(0L)); + } + + @Test + public void testInRestarting() { + final Map<JobStatus, Long> statusTimestampMap = new HashMap<>(); + statusTimestampMap.put(JobStatus.RESTARTING, 1L); + + final RestartTimeGauge gauge = new RestartTimeGauge( + new TestingJobStatusProvider( + JobStatus.RESTARTING, + status -> statusTimestampMap.getOrDefault(status, -1L))); + // System.currentTimeMillis() is surely to be larger than 123L Review comment: The comment seems outdated. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java ########## @@ -82,202 +82,12 @@ private final TestingComponentMainThreadExecutor testMainThreadUtil = EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor(); - /** - * Tests that slots are released if we cannot assign the allocated resource to the - * Execution. - */ - @Test - public void testSlotReleaseOnFailedResourceAssignment() throws Exception { - final JobVertex jobVertex = createNoOpJobVertex(); - final JobVertexID jobVertexId = jobVertex.getID(); - - final CompletableFuture<LogicalSlot> slotFuture = new CompletableFuture<>(); - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1); - slotProvider.addSlot(jobVertexId, 0, slotFuture); - - ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph( - slotProvider, - new NoRestartStrategy(), - jobVertex); - - executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId); - - final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt(); - - final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner(); - - final LogicalSlot slot = createTestingLogicalSlot(slotOwner); - - final LogicalSlot otherSlot = new TestingLogicalSlotBuilder().createTestingLogicalSlot(); - - CompletableFuture<Execution> allocationFuture = execution.allocateResourcesForExecution( - executionGraph.getSlotProviderStrategy(), - LocationPreferenceConstraint.ALL, - Collections.emptySet()); - - assertFalse(allocationFuture.isDone()); - - assertEquals(ExecutionState.SCHEDULED, execution.getState()); - - // assign a different resource to the execution - assertTrue(execution.tryAssignResource(otherSlot)); - - // completing now the future should cause the slot to be released - slotFuture.complete(slot); - - assertEquals(slot, slotOwner.getReturnedSlotFuture().get()); - } - private TestingLogicalSlot createTestingLogicalSlot(SlotOwner slotOwner) { return new TestingLogicalSlotBuilder() .setSlotOwner(slotOwner) .createTestingLogicalSlot(); } - /** - * Tests that the slot is released in case of a execution cancellation when having - * a slot assigned and being in state SCHEDULED. - */ - @Test - public void testSlotReleaseOnExecutionCancellationInScheduled() throws Exception { Review comment: I think this test is still valid because we want to return the assigned slots if the `Execution` gets cancelled. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java ########## @@ -82,202 +82,12 @@ private final TestingComponentMainThreadExecutor testMainThreadUtil = EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor(); - /** - * Tests that slots are released if we cannot assign the allocated resource to the - * Execution. - */ - @Test - public void testSlotReleaseOnFailedResourceAssignment() throws Exception { - final JobVertex jobVertex = createNoOpJobVertex(); - final JobVertexID jobVertexId = jobVertex.getID(); - - final CompletableFuture<LogicalSlot> slotFuture = new CompletableFuture<>(); - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1); - slotProvider.addSlot(jobVertexId, 0, slotFuture); - - ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph( - slotProvider, - new NoRestartStrategy(), - jobVertex); - - executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId); - - final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt(); - - final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner(); - - final LogicalSlot slot = createTestingLogicalSlot(slotOwner); - - final LogicalSlot otherSlot = new TestingLogicalSlotBuilder().createTestingLogicalSlot(); - - CompletableFuture<Execution> allocationFuture = execution.allocateResourcesForExecution( - executionGraph.getSlotProviderStrategy(), - LocationPreferenceConstraint.ALL, - Collections.emptySet()); - - assertFalse(allocationFuture.isDone()); - - assertEquals(ExecutionState.SCHEDULED, execution.getState()); - - // assign a different resource to the execution - assertTrue(execution.tryAssignResource(otherSlot)); - - // completing now the future should cause the slot to be released - slotFuture.complete(slot); - - assertEquals(slot, slotOwner.getReturnedSlotFuture().get()); - } - private TestingLogicalSlot createTestingLogicalSlot(SlotOwner slotOwner) { return new TestingLogicalSlotBuilder() .setSlotOwner(slotOwner) .createTestingLogicalSlot(); } - /** - * Tests that the slot is released in case of a execution cancellation when having - * a slot assigned and being in state SCHEDULED. - */ - @Test - public void testSlotReleaseOnExecutionCancellationInScheduled() throws Exception { - final JobVertex jobVertex = createNoOpJobVertex(); - final JobVertexID jobVertexId = jobVertex.getID(); - - final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner(); - - final LogicalSlot slot = createTestingLogicalSlot(slotOwner); - - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1); - slotProvider.addSlot(jobVertexId, 0, CompletableFuture.completedFuture(slot)); - - ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph( - slotProvider, - new NoRestartStrategy(), - jobVertex); - - executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId); - - final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt(); - - CompletableFuture<Execution> allocationFuture = execution.allocateResourcesForExecution( - executionGraph.getSlotProviderStrategy(), - LocationPreferenceConstraint.ALL, - Collections.emptySet()); - - assertTrue(allocationFuture.isDone()); - - assertEquals(ExecutionState.SCHEDULED, execution.getState()); - - assertEquals(slot, execution.getAssignedResource()); - - // cancelling the execution should move it into state CANCELED - execution.cancel(); - assertEquals(ExecutionState.CANCELED, execution.getState()); - - assertEquals(slot, slotOwner.getReturnedSlotFuture().get()); - } - - /** - * Tests that the slot is released in case of a execution cancellation when being in state - * RUNNING. - */ - @Test - public void testSlotReleaseOnExecutionCancellationInRunning() throws Exception { - final JobVertex jobVertex = createNoOpJobVertex(); - final JobVertexID jobVertexId = jobVertex.getID(); - - final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner(); - - final LogicalSlot slot = createTestingLogicalSlot(slotOwner); - - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1); - slotProvider.addSlot(jobVertexId, 0, CompletableFuture.completedFuture(slot)); - - ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph( - slotProvider, - new NoRestartStrategy(), - jobVertex); - - ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId); - - final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt(); - - CompletableFuture<Execution> allocationFuture = execution.allocateResourcesForExecution( - executionGraph.getSlotProviderStrategy(), - LocationPreferenceConstraint.ALL, - Collections.emptySet()); - - assertTrue(allocationFuture.isDone()); - - assertEquals(ExecutionState.SCHEDULED, execution.getState()); - - assertEquals(slot, execution.getAssignedResource()); - - execution.deploy(); - - execution.switchToRunning(); - - // cancelling the execution should move it into state CANCELING - execution.cancel(); - assertEquals(ExecutionState.CANCELING, execution.getState()); - - execution.completeCancelling(); - - assertEquals(slot, slotOwner.getReturnedSlotFuture().get()); - } - - /** - * Tests that a slot allocation from a {@link SlotProvider} is cancelled if the - * {@link Execution} is cancelled. - */ - @Test - public void testSlotAllocationCancellationWhenExecutionCancelled() throws Exception { Review comment: I think this test is also still valid. At least I couldn't find a test which ensures that we cancel our slot allocations when we cancel the execution. I think currently, this only happens if a task failed. But maybe the test need to make sure that the `DefaultScheduler` cancels all pending slot requests if the `SchedulerNG.cancel` is called. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java ########## @@ -1,637 +0,0 @@ -/* - * 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; - -import org.apache.flink.api.common.JobID; -import org.apache.flink.api.common.JobStatus; -import org.apache.flink.api.common.time.Time; -import org.apache.flink.configuration.Configuration; -import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; -import org.apache.flink.runtime.blob.VoidBlobWriter; -import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory; -import org.apache.flink.runtime.clusterframework.types.AllocationID; -import org.apache.flink.runtime.clusterframework.types.ResourceID; -import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter; -import org.apache.flink.runtime.execution.ExecutionState; -import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy; -import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway; -import org.apache.flink.runtime.instance.SimpleSlotContext; -import org.apache.flink.runtime.io.network.partition.NoOpJobMasterPartitionTracker; -import org.apache.flink.runtime.io.network.partition.ResultPartitionType; -import org.apache.flink.runtime.jobgraph.DistributionPattern; -import org.apache.flink.runtime.jobgraph.JobGraph; -import org.apache.flink.runtime.jobgraph.JobVertex; -import org.apache.flink.runtime.jobgraph.ScheduleMode; -import org.apache.flink.runtime.jobmanager.scheduler.Locality; -import org.apache.flink.runtime.jobmanager.slots.DummySlotOwner; -import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway; -import org.apache.flink.runtime.jobmanager.slots.TestingSlotOwner; -import org.apache.flink.runtime.jobmaster.LogicalSlot; -import org.apache.flink.runtime.jobmaster.SlotOwner; -import org.apache.flink.runtime.jobmaster.SlotRequestId; -import org.apache.flink.runtime.jobmaster.TestingLogicalSlot; -import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder; -import org.apache.flink.runtime.jobmaster.slotpool.SingleLogicalSlot; -import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider; -import org.apache.flink.runtime.shuffle.NettyShuffleMaster; -import org.apache.flink.runtime.taskmanager.TaskManagerLocation; -import org.apache.flink.runtime.testtasks.NoOpInvokable; -import org.apache.flink.runtime.testutils.DirectScheduledExecutorService; -import org.apache.flink.util.FlinkException; -import org.apache.flink.util.TestLogger; - -import org.junit.After; -import org.junit.Test; - -import javax.annotation.Nonnull; - -import java.net.InetAddress; -import java.util.Set; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -import static org.hamcrest.Matchers.empty; -import static org.hamcrest.Matchers.is; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertThat; - -/** - * Tests for the scheduling of the execution graph. This tests that - * for example the order of deployments is correct and that bulk slot allocation - * works properly. - */ -public class ExecutionGraphSchedulingTest extends TestLogger { - - private final ScheduledExecutorService executor = new DirectScheduledExecutorService(); - - @After - public void shutdown() { - executor.shutdownNow(); - } - - // ------------------------------------------------------------------------ - // Tests - // ------------------------------------------------------------------------ - - /** - * Tests that with scheduling futures and pipelined deployment, the target vertex will - * not deploy its task before the source vertex does. - */ - @Test - public void testScheduleSourceBeforeTarget() throws Exception { - - // [pipelined] - // we construct a simple graph (source) ----------------> (target) - - final int parallelism = 1; - - final JobVertex sourceVertex = new JobVertex("source"); - sourceVertex.setParallelism(parallelism); - sourceVertex.setInvokableClass(NoOpInvokable.class); - - final JobVertex targetVertex = new JobVertex("target"); - targetVertex.setParallelism(parallelism); - targetVertex.setInvokableClass(NoOpInvokable.class); - - targetVertex.connectNewDataSetAsInput(sourceVertex, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - final CompletableFuture<LogicalSlot> sourceFuture = new CompletableFuture<>(); - final CompletableFuture<LogicalSlot> targetFuture = new CompletableFuture<>(); - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlot(sourceVertex.getID(), 0, sourceFuture); - slotProvider.addSlot(targetVertex.getID(), 0, targetFuture); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - // set up two TaskManager gateways and slots - - final InteractionsCountingTaskManagerGateway gatewaySource = createTaskManager(); - final InteractionsCountingTaskManagerGateway gatewayTarget = createTaskManager(); - - final LogicalSlot sourceSlot = createTestingLogicalSlot(gatewaySource); - final LogicalSlot targetSlot = createTestingLogicalSlot(gatewayTarget); - - eg.scheduleForExecution(); - - // job should be running - assertEquals(JobStatus.RUNNING, eg.getState()); - - // we fulfill the target slot before the source slot - // that should not cause a deployment or deployment related failure - targetFuture.complete(targetSlot); - - assertThat(gatewayTarget.getSubmitTaskCount(), is(0)); - assertEquals(JobStatus.RUNNING, eg.getState()); - - // now supply the source slot - sourceFuture.complete(sourceSlot); - - // by now, all deployments should have happened - assertThat(gatewaySource.getSubmitTaskCount(), is(1)); - assertThat(gatewayTarget.getSubmitTaskCount(), is(1)); - - assertEquals(JobStatus.RUNNING, eg.getState()); - } - - private TestingLogicalSlot createTestingLogicalSlot(InteractionsCountingTaskManagerGateway gatewaySource) { - return new TestingLogicalSlotBuilder() - .setTaskManagerGateway(gatewaySource) - .createTestingLogicalSlot(); - } - - /** - * This test verifies that before deploying a pipelined connected component, the - * full set of slots is available, and that not some tasks are deployed, and later the - * system realizes that not enough resources are available. - */ - @Test - public void testDeployPipelinedConnectedComponentsTogether() throws Exception { - - // [pipelined] - // we construct a simple graph (source) ----------------> (target) - - final int parallelism = 8; - - final JobVertex sourceVertex = new JobVertex("source"); - sourceVertex.setParallelism(parallelism); - sourceVertex.setInvokableClass(NoOpInvokable.class); - - final JobVertex targetVertex = new JobVertex("target"); - targetVertex.setParallelism(parallelism); - targetVertex.setInvokableClass(NoOpInvokable.class); - - targetVertex.connectNewDataSetAsInput(sourceVertex, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] sourceFutures = new CompletableFuture[parallelism]; - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] targetFutures = new CompletableFuture[parallelism]; - - // - // Create the slots, futures, and the slot provider - - final InteractionsCountingTaskManagerGateway[] sourceTaskManagers = new InteractionsCountingTaskManagerGateway[parallelism]; - final InteractionsCountingTaskManagerGateway[] targetTaskManagers = new InteractionsCountingTaskManagerGateway[parallelism]; - - final LogicalSlot[] sourceSlots = new LogicalSlot[parallelism]; - final LogicalSlot[] targetSlots = new LogicalSlot[parallelism]; - - for (int i = 0; i < parallelism; i++) { - sourceTaskManagers[i] = createTaskManager(); - targetTaskManagers[i] = createTaskManager(); - - sourceSlots[i] = createTestingLogicalSlot(sourceTaskManagers[i]); - targetSlots[i] = createTestingLogicalSlot(targetTaskManagers[i]); - - sourceFutures[i] = new CompletableFuture<>(); - targetFutures[i] = new CompletableFuture<>(); - } - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlots(sourceVertex.getID(), sourceFutures); - slotProvider.addSlots(targetVertex.getID(), targetFutures); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - - // - // we complete some of the futures - - for (int i = 0; i < parallelism; i += 2) { - sourceFutures[i].complete(sourceSlots[i]); - } - - // - // kick off the scheduling - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - eg.scheduleForExecution(); - - verifyNothingDeployed(eg, sourceTaskManagers); - - // complete the remaining sources - for (int i = 1; i < parallelism; i += 2) { - sourceFutures[i].complete(sourceSlots[i]); - } - verifyNothingDeployed(eg, sourceTaskManagers); - - // complete the targets except for one - for (int i = 1; i < parallelism; i++) { - targetFutures[i].complete(targetSlots[i]); - } - verifyNothingDeployed(eg, targetTaskManagers); - - // complete the last target slot future - targetFutures[0].complete(targetSlots[0]); - - // - // verify that all deployments have happened - - for (InteractionsCountingTaskManagerGateway gateway : sourceTaskManagers) { - assertThat(gateway.getSubmitTaskCount(), is(1)); - } - for (InteractionsCountingTaskManagerGateway gateway : targetTaskManagers) { - assertThat(gateway.getSubmitTaskCount(), is(1)); - } - } - - /** - * This test verifies that if one slot future fails, the deployment will be aborted. - */ - @Test - public void testOneSlotFailureAbortsDeploy() throws Exception { - - // [pipelined] - // we construct a simple graph (source) ----------------> (target) - - final int parallelism = 6; - - final JobVertex sourceVertex = new JobVertex("source"); - sourceVertex.setParallelism(parallelism); - sourceVertex.setInvokableClass(NoOpInvokable.class); - - final JobVertex targetVertex = new JobVertex("target"); - targetVertex.setParallelism(parallelism); - targetVertex.setInvokableClass(NoOpInvokable.class); - - targetVertex.connectNewDataSetAsInput(sourceVertex, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - // - // Create the slots, futures, and the slot provider - - final InteractionsCountingTaskManagerGateway taskManager = createTaskManager(); - final BlockingQueue<AllocationID> returnedSlots = new ArrayBlockingQueue<>(parallelism); - final TestingSlotOwner slotOwner = new TestingSlotOwner(); - slotOwner.setReturnAllocatedSlotConsumer( - (LogicalSlot logicalSlot) -> returnedSlots.offer(logicalSlot.getAllocationId())); - - final LogicalSlot[] sourceSlots = new LogicalSlot[parallelism]; - final LogicalSlot[] targetSlots = new LogicalSlot[parallelism]; - - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] sourceFutures = new CompletableFuture[parallelism]; - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] targetFutures = new CompletableFuture[parallelism]; - - for (int i = 0; i < parallelism; i++) { - sourceSlots[i] = createSingleLogicalSlot(slotOwner, taskManager, new SlotRequestId()); - targetSlots[i] = createSingleLogicalSlot(slotOwner, taskManager, new SlotRequestId()); - - sourceFutures[i] = new CompletableFuture<>(); - targetFutures[i] = new CompletableFuture<>(); - } - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlots(sourceVertex.getID(), sourceFutures); - slotProvider.addSlots(targetVertex.getID(), targetFutures); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - // - // we complete some of the futures - - for (int i = 0; i < parallelism; i += 2) { - sourceFutures[i].complete(sourceSlots[i]); - targetFutures[i].complete(targetSlots[i]); - } - - // kick off the scheduling - eg.scheduleForExecution(); - - // fail one slot - sourceFutures[1].completeExceptionally(new TestRuntimeException()); - - // wait until the job failed as a whole - eg.getTerminationFuture().get(2000, TimeUnit.MILLISECONDS); - - // wait until all slots are back - for (int i = 0; i < parallelism; i++) { - returnedSlots.poll(2000L, TimeUnit.MILLISECONDS); - } - - // no deployment calls must have happened - assertThat(taskManager.getSubmitTaskCount(), is(0)); - - // all completed futures must have been returns - for (int i = 0; i < parallelism; i += 2) { - assertFalse(sourceSlots[i].isAlive()); - assertFalse(targetSlots[i].isAlive()); - } - } - - /** - * This tests makes sure that with eager scheduling no task is deployed if a single - * slot allocation fails. Moreover we check that allocated slots will be returned. - */ - @Test - public void testEagerSchedulingWithSlotTimeout() throws Exception { - - // we construct a simple graph: (task) - - final int parallelism = 3; - - final JobVertex vertex = new JobVertex("task"); - vertex.setParallelism(parallelism); - vertex.setInvokableClass(NoOpInvokable.class); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", vertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - final BlockingQueue<AllocationID> returnedSlots = new ArrayBlockingQueue<>(2); - final TestingSlotOwner slotOwner = new TestingSlotOwner(); - slotOwner.setReturnAllocatedSlotConsumer( - (LogicalSlot logicalSlot) -> returnedSlots.offer(logicalSlot.getAllocationId())); - - final InteractionsCountingTaskManagerGateway taskManager = createTaskManager(); - final LogicalSlot[] slots = new LogicalSlot[parallelism]; - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] slotFutures = new CompletableFuture[parallelism]; - - for (int i = 0; i < parallelism; i++) { - slots[i] = createSingleLogicalSlot(slotOwner, taskManager, new SlotRequestId()); - slotFutures[i] = new CompletableFuture<>(); - } - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlots(vertex.getID(), slotFutures); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - - // we complete one future - slotFutures[1].complete(slots[1]); - - // kick off the scheduling - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - eg.scheduleForExecution(); - - // we complete another future - slotFutures[2].complete(slots[2]); - - // check that the ExecutionGraph is not terminated yet - assertThat(eg.getTerminationFuture().isDone(), is(false)); - - // time out one of the slot futures - slotFutures[0].completeExceptionally(new TimeoutException("Test time out")); - - assertThat(eg.getTerminationFuture().get(), is(JobStatus.FAILED)); - - // wait until all slots are back - for (int i = 0; i < parallelism - 1; i++) { - returnedSlots.poll(2000, TimeUnit.MILLISECONDS); - } - - // verify that no deployments have happened - assertThat(taskManager.getSubmitTaskCount(), is(0)); - } - - /** - * Tests that an ongoing scheduling operation does not fail the {@link ExecutionGraph} - * if it gets concurrently cancelled. - */ - @Test - public void testSchedulingOperationCancellationWhenCancel() throws Exception { Review comment: I guess this test is similar to `ExecutionGraphRestartTest.testFailWhileCanceling` which is missing. If the state is `CANCELLING`, then we should not transition into `FAILING`. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java ########## @@ -82,202 +82,12 @@ private final TestingComponentMainThreadExecutor testMainThreadUtil = EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor(); - /** - * Tests that slots are released if we cannot assign the allocated resource to the - * Execution. - */ - @Test - public void testSlotReleaseOnFailedResourceAssignment() throws Exception { - final JobVertex jobVertex = createNoOpJobVertex(); - final JobVertexID jobVertexId = jobVertex.getID(); - - final CompletableFuture<LogicalSlot> slotFuture = new CompletableFuture<>(); - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1); - slotProvider.addSlot(jobVertexId, 0, slotFuture); - - ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph( - slotProvider, - new NoRestartStrategy(), - jobVertex); - - executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId); - - final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt(); - - final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner(); - - final LogicalSlot slot = createTestingLogicalSlot(slotOwner); - - final LogicalSlot otherSlot = new TestingLogicalSlotBuilder().createTestingLogicalSlot(); - - CompletableFuture<Execution> allocationFuture = execution.allocateResourcesForExecution( - executionGraph.getSlotProviderStrategy(), - LocationPreferenceConstraint.ALL, - Collections.emptySet()); - - assertFalse(allocationFuture.isDone()); - - assertEquals(ExecutionState.SCHEDULED, execution.getState()); - - // assign a different resource to the execution - assertTrue(execution.tryAssignResource(otherSlot)); - - // completing now the future should cause the slot to be released - slotFuture.complete(slot); - - assertEquals(slot, slotOwner.getReturnedSlotFuture().get()); - } - private TestingLogicalSlot createTestingLogicalSlot(SlotOwner slotOwner) { return new TestingLogicalSlotBuilder() .setSlotOwner(slotOwner) .createTestingLogicalSlot(); } - /** - * Tests that the slot is released in case of a execution cancellation when having - * a slot assigned and being in state SCHEDULED. - */ - @Test - public void testSlotReleaseOnExecutionCancellationInScheduled() throws Exception { - final JobVertex jobVertex = createNoOpJobVertex(); - final JobVertexID jobVertexId = jobVertex.getID(); - - final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner(); - - final LogicalSlot slot = createTestingLogicalSlot(slotOwner); - - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(1); - slotProvider.addSlot(jobVertexId, 0, CompletableFuture.completedFuture(slot)); - - ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph( - slotProvider, - new NoRestartStrategy(), - jobVertex); - - executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId); - - final Execution execution = executionJobVertex.getTaskVertices()[0].getCurrentExecutionAttempt(); - - CompletableFuture<Execution> allocationFuture = execution.allocateResourcesForExecution( - executionGraph.getSlotProviderStrategy(), - LocationPreferenceConstraint.ALL, - Collections.emptySet()); - - assertTrue(allocationFuture.isDone()); - - assertEquals(ExecutionState.SCHEDULED, execution.getState()); - - assertEquals(slot, execution.getAssignedResource()); - - // cancelling the execution should move it into state CANCELED - execution.cancel(); - assertEquals(ExecutionState.CANCELED, execution.getState()); - - assertEquals(slot, slotOwner.getReturnedSlotFuture().get()); - } - - /** - * Tests that the slot is released in case of a execution cancellation when being in state - * RUNNING. - */ - @Test - public void testSlotReleaseOnExecutionCancellationInRunning() throws Exception { Review comment: I think this test is also still valid because we want to return the logical slot to its owner if the `Execution` is cancelled. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/checkpoint/ExecutionGraphCheckpointCoordinatorTest.java ########## @@ -159,7 +164,7 @@ private ExecutionGraph createExecutionGraphAndEnableCheckpointing( false, 0); - executionGraph.enableCheckpointing( + scheduler.getExecutionGraph().enableCheckpointing( Review comment: I guess now we would have to specify a specialized `CheckpointRecoveryFactory` which we pass to the `SchedulerBuilder`. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphRestartTest.java ########## @@ -1,866 +0,0 @@ -/* - * 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; - -import org.apache.flink.api.common.ExecutionConfig; -import org.apache.flink.api.common.JobID; -import org.apache.flink.api.common.JobStatus; -import org.apache.flink.api.common.restartstrategy.RestartStrategies; -import org.apache.flink.api.common.time.Time; -import org.apache.flink.runtime.clusterframework.types.AllocationID; -import org.apache.flink.runtime.clusterframework.types.ResourceProfile; -import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor; -import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter; -import org.apache.flink.runtime.execution.ExecutionState; -import org.apache.flink.runtime.execution.SuppressRestartsException; -import org.apache.flink.runtime.executiongraph.failover.FailoverStrategy; -import org.apache.flink.runtime.executiongraph.restart.FixedDelayRestartStrategy; -import org.apache.flink.runtime.executiongraph.restart.InfiniteDelayRestartStrategy; -import org.apache.flink.runtime.executiongraph.restart.RestartStrategy; -import org.apache.flink.runtime.executiongraph.utils.NotCancelAckingTaskGateway; -import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway; -import org.apache.flink.runtime.executiongraph.utils.SimpleSlotProvider; -import org.apache.flink.runtime.io.network.partition.ResultPartitionType; -import org.apache.flink.runtime.jobgraph.DistributionPattern; -import org.apache.flink.runtime.jobgraph.JobGraph; -import org.apache.flink.runtime.jobgraph.JobVertex; -import org.apache.flink.runtime.jobgraph.ScheduleMode; -import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup; -import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway; -import org.apache.flink.runtime.jobmaster.JobMasterId; -import org.apache.flink.runtime.jobmaster.slotpool.LocationPreferenceSlotSelectionStrategy; -import org.apache.flink.runtime.jobmaster.slotpool.Scheduler; -import org.apache.flink.runtime.jobmaster.slotpool.SchedulerImpl; -import org.apache.flink.runtime.jobmaster.slotpool.SlotPool; -import org.apache.flink.runtime.jobmaster.slotpool.SlotPoolImpl; -import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider; -import org.apache.flink.runtime.jobmaster.slotpool.TestingSlotPoolImpl; -import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway; -import org.apache.flink.runtime.resourcemanager.utils.TestingResourceManagerGateway; -import org.apache.flink.runtime.taskexecutor.slot.SlotOffer; -import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation; -import org.apache.flink.runtime.taskmanager.TaskManagerLocation; -import org.apache.flink.runtime.testtasks.NoOpInvokable; -import org.apache.flink.util.FlinkException; -import org.apache.flink.util.TestLogger; - -import org.junit.Test; - -import javax.annotation.Nonnull; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.Iterator; -import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.function.Consumer; - -import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.completeCancellingForAllVertices; -import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createNoOpVertex; -import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.finishAllVertices; -import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.switchToRunning; -import static org.hamcrest.Matchers.is; -import static org.hamcrest.Matchers.lessThanOrEqualTo; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertThat; -import static org.junit.Assert.assertTrue; - -/** - * Tests the restart behaviour of the {@link ExecutionGraph}. - */ -public class ExecutionGraphRestartTest extends TestLogger { Review comment: I think `testTaskFailingWhileGlobalFailing` is irrelevant now `testFailWhileRestarting` should be covered by `DefaultSchedulerTest.failJobIfCannotRestart` For the following tests I couldn't find test coverage: ``` testCancelWhileRestarting testCancelWhileFailing testFailWhileCanceling testFailingExecutionAfterRestart testFailExecutionAfterCancel testSuspendWhileRestarting ``` ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java ########## @@ -244,104 +250,6 @@ public void testSendCancelAndReceiveFail() throws Exception { assertEquals(vertices.length - 1, exec.getVertex().getExecutionGraph().getRegisteredExecutions().size()); } - // -------------------------------------------------------------------------------------------- - // Actions after a vertex has been canceled or while canceling - // -------------------------------------------------------------------------------------------- - - @Test - public void testScheduleOrDeployAfterCancel() { - try { - final ExecutionVertex vertex = getExecutionVertex(); - setVertexState(vertex, ExecutionState.CANCELED); - - assertEquals(ExecutionState.CANCELED, vertex.getExecutionState()); - - // 1) - // scheduling after being canceled should be tolerated (no exception) because - // it can occur as the result of races - { - vertex.scheduleForExecution( - TestingSlotProviderStrategy.from(new ProgrammedSlotProvider(1)), - LocationPreferenceConstraint.ALL, - Collections.emptySet()); - - assertEquals(ExecutionState.CANCELED, vertex.getExecutionState()); - } - - // 2) - // deploying after canceling from CREATED needs to raise an exception, because - // the scheduler (or any caller) needs to know that the slot should be released - try { - - final LogicalSlot slot = new TestingLogicalSlotBuilder().createTestingLogicalSlot(); - - vertex.deployToSlot(slot); - fail("Method should throw an exception"); - } - catch (IllegalStateException e) { - assertEquals(ExecutionState.CANCELED, vertex.getExecutionState()); - } - } - catch (Exception e) { - e.printStackTrace(); - fail(e.getMessage()); - } - } - - @Test - public void testActionsWhileCancelling() { Review comment: I think `DefaultSchedulerTest#skipDeploymentIfVertexVersionOutdated()` is fine. ########## 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: I think `DefaultSchedulerTest#deployTasksOnlyWhenAllSlotRequestsAreFulfilled` replaces `ExecutionGraphDeploymentTest#testEagerSchedulingWaitsOnAllInputPreferredLocations()` because what we are interested is that the scheduler deploys the whole pipelined region. There is no longer the need that pipelined consumers will be scheduled after the producers because they are scheduled together. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java ########## @@ -82,202 +82,12 @@ private final TestingComponentMainThreadExecutor testMainThreadUtil = EXECUTOR_RESOURCE.getComponentMainThreadTestExecutor(); - /** - * Tests that slots are released if we cannot assign the allocated resource to the - * Execution. - */ - @Test - public void testSlotReleaseOnFailedResourceAssignment() throws Exception { Review comment: `testSlotReleaseOnFailedResourceAssignment` should indeed be obsolete. ########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphSchedulingTest.java ########## @@ -1,637 +0,0 @@ -/* - * 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; - -import org.apache.flink.api.common.JobID; -import org.apache.flink.api.common.JobStatus; -import org.apache.flink.api.common.time.Time; -import org.apache.flink.configuration.Configuration; -import org.apache.flink.metrics.groups.UnregisteredMetricsGroup; -import org.apache.flink.runtime.blob.VoidBlobWriter; -import org.apache.flink.runtime.checkpoint.StandaloneCheckpointRecoveryFactory; -import org.apache.flink.runtime.clusterframework.types.AllocationID; -import org.apache.flink.runtime.clusterframework.types.ResourceID; -import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter; -import org.apache.flink.runtime.execution.ExecutionState; -import org.apache.flink.runtime.executiongraph.restart.NoRestartStrategy; -import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway; -import org.apache.flink.runtime.instance.SimpleSlotContext; -import org.apache.flink.runtime.io.network.partition.NoOpJobMasterPartitionTracker; -import org.apache.flink.runtime.io.network.partition.ResultPartitionType; -import org.apache.flink.runtime.jobgraph.DistributionPattern; -import org.apache.flink.runtime.jobgraph.JobGraph; -import org.apache.flink.runtime.jobgraph.JobVertex; -import org.apache.flink.runtime.jobgraph.ScheduleMode; -import org.apache.flink.runtime.jobmanager.scheduler.Locality; -import org.apache.flink.runtime.jobmanager.slots.DummySlotOwner; -import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway; -import org.apache.flink.runtime.jobmanager.slots.TestingSlotOwner; -import org.apache.flink.runtime.jobmaster.LogicalSlot; -import org.apache.flink.runtime.jobmaster.SlotOwner; -import org.apache.flink.runtime.jobmaster.SlotRequestId; -import org.apache.flink.runtime.jobmaster.TestingLogicalSlot; -import org.apache.flink.runtime.jobmaster.TestingLogicalSlotBuilder; -import org.apache.flink.runtime.jobmaster.slotpool.SingleLogicalSlot; -import org.apache.flink.runtime.jobmaster.slotpool.SlotProvider; -import org.apache.flink.runtime.shuffle.NettyShuffleMaster; -import org.apache.flink.runtime.taskmanager.TaskManagerLocation; -import org.apache.flink.runtime.testtasks.NoOpInvokable; -import org.apache.flink.runtime.testutils.DirectScheduledExecutorService; -import org.apache.flink.util.FlinkException; -import org.apache.flink.util.TestLogger; - -import org.junit.After; -import org.junit.Test; - -import javax.annotation.Nonnull; - -import java.net.InetAddress; -import java.util.Set; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -import static org.hamcrest.Matchers.empty; -import static org.hamcrest.Matchers.is; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertThat; - -/** - * Tests for the scheduling of the execution graph. This tests that - * for example the order of deployments is correct and that bulk slot allocation - * works properly. - */ -public class ExecutionGraphSchedulingTest extends TestLogger { - - private final ScheduledExecutorService executor = new DirectScheduledExecutorService(); - - @After - public void shutdown() { - executor.shutdownNow(); - } - - // ------------------------------------------------------------------------ - // Tests - // ------------------------------------------------------------------------ - - /** - * Tests that with scheduling futures and pipelined deployment, the target vertex will - * not deploy its task before the source vertex does. - */ - @Test - public void testScheduleSourceBeforeTarget() throws Exception { - - // [pipelined] - // we construct a simple graph (source) ----------------> (target) - - final int parallelism = 1; - - final JobVertex sourceVertex = new JobVertex("source"); - sourceVertex.setParallelism(parallelism); - sourceVertex.setInvokableClass(NoOpInvokable.class); - - final JobVertex targetVertex = new JobVertex("target"); - targetVertex.setParallelism(parallelism); - targetVertex.setInvokableClass(NoOpInvokable.class); - - targetVertex.connectNewDataSetAsInput(sourceVertex, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - final CompletableFuture<LogicalSlot> sourceFuture = new CompletableFuture<>(); - final CompletableFuture<LogicalSlot> targetFuture = new CompletableFuture<>(); - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlot(sourceVertex.getID(), 0, sourceFuture); - slotProvider.addSlot(targetVertex.getID(), 0, targetFuture); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - // set up two TaskManager gateways and slots - - final InteractionsCountingTaskManagerGateway gatewaySource = createTaskManager(); - final InteractionsCountingTaskManagerGateway gatewayTarget = createTaskManager(); - - final LogicalSlot sourceSlot = createTestingLogicalSlot(gatewaySource); - final LogicalSlot targetSlot = createTestingLogicalSlot(gatewayTarget); - - eg.scheduleForExecution(); - - // job should be running - assertEquals(JobStatus.RUNNING, eg.getState()); - - // we fulfill the target slot before the source slot - // that should not cause a deployment or deployment related failure - targetFuture.complete(targetSlot); - - assertThat(gatewayTarget.getSubmitTaskCount(), is(0)); - assertEquals(JobStatus.RUNNING, eg.getState()); - - // now supply the source slot - sourceFuture.complete(sourceSlot); - - // by now, all deployments should have happened - assertThat(gatewaySource.getSubmitTaskCount(), is(1)); - assertThat(gatewayTarget.getSubmitTaskCount(), is(1)); - - assertEquals(JobStatus.RUNNING, eg.getState()); - } - - private TestingLogicalSlot createTestingLogicalSlot(InteractionsCountingTaskManagerGateway gatewaySource) { - return new TestingLogicalSlotBuilder() - .setTaskManagerGateway(gatewaySource) - .createTestingLogicalSlot(); - } - - /** - * This test verifies that before deploying a pipelined connected component, the - * full set of slots is available, and that not some tasks are deployed, and later the - * system realizes that not enough resources are available. - */ - @Test - public void testDeployPipelinedConnectedComponentsTogether() throws Exception { - - // [pipelined] - // we construct a simple graph (source) ----------------> (target) - - final int parallelism = 8; - - final JobVertex sourceVertex = new JobVertex("source"); - sourceVertex.setParallelism(parallelism); - sourceVertex.setInvokableClass(NoOpInvokable.class); - - final JobVertex targetVertex = new JobVertex("target"); - targetVertex.setParallelism(parallelism); - targetVertex.setInvokableClass(NoOpInvokable.class); - - targetVertex.connectNewDataSetAsInput(sourceVertex, DistributionPattern.ALL_TO_ALL, ResultPartitionType.PIPELINED); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] sourceFutures = new CompletableFuture[parallelism]; - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] targetFutures = new CompletableFuture[parallelism]; - - // - // Create the slots, futures, and the slot provider - - final InteractionsCountingTaskManagerGateway[] sourceTaskManagers = new InteractionsCountingTaskManagerGateway[parallelism]; - final InteractionsCountingTaskManagerGateway[] targetTaskManagers = new InteractionsCountingTaskManagerGateway[parallelism]; - - final LogicalSlot[] sourceSlots = new LogicalSlot[parallelism]; - final LogicalSlot[] targetSlots = new LogicalSlot[parallelism]; - - for (int i = 0; i < parallelism; i++) { - sourceTaskManagers[i] = createTaskManager(); - targetTaskManagers[i] = createTaskManager(); - - sourceSlots[i] = createTestingLogicalSlot(sourceTaskManagers[i]); - targetSlots[i] = createTestingLogicalSlot(targetTaskManagers[i]); - - sourceFutures[i] = new CompletableFuture<>(); - targetFutures[i] = new CompletableFuture<>(); - } - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlots(sourceVertex.getID(), sourceFutures); - slotProvider.addSlots(targetVertex.getID(), targetFutures); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - - // - // we complete some of the futures - - for (int i = 0; i < parallelism; i += 2) { - sourceFutures[i].complete(sourceSlots[i]); - } - - // - // kick off the scheduling - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - eg.scheduleForExecution(); - - verifyNothingDeployed(eg, sourceTaskManagers); - - // complete the remaining sources - for (int i = 1; i < parallelism; i += 2) { - sourceFutures[i].complete(sourceSlots[i]); - } - verifyNothingDeployed(eg, sourceTaskManagers); - - // complete the targets except for one - for (int i = 1; i < parallelism; i++) { - targetFutures[i].complete(targetSlots[i]); - } - verifyNothingDeployed(eg, targetTaskManagers); - - // complete the last target slot future - targetFutures[0].complete(targetSlots[0]); - - // - // verify that all deployments have happened - - for (InteractionsCountingTaskManagerGateway gateway : sourceTaskManagers) { - assertThat(gateway.getSubmitTaskCount(), is(1)); - } - for (InteractionsCountingTaskManagerGateway gateway : targetTaskManagers) { - assertThat(gateway.getSubmitTaskCount(), is(1)); - } - } - - /** - * This test verifies that if one slot future fails, the deployment will be aborted. - */ - @Test - public void testOneSlotFailureAbortsDeploy() throws Exception { - - // [pipelined] - // we construct a simple graph (source) ----------------> (target) - - final int parallelism = 6; - - final JobVertex sourceVertex = new JobVertex("source"); - sourceVertex.setParallelism(parallelism); - sourceVertex.setInvokableClass(NoOpInvokable.class); - - final JobVertex targetVertex = new JobVertex("target"); - targetVertex.setParallelism(parallelism); - targetVertex.setInvokableClass(NoOpInvokable.class); - - targetVertex.connectNewDataSetAsInput(sourceVertex, DistributionPattern.POINTWISE, ResultPartitionType.PIPELINED); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", sourceVertex, targetVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - // - // Create the slots, futures, and the slot provider - - final InteractionsCountingTaskManagerGateway taskManager = createTaskManager(); - final BlockingQueue<AllocationID> returnedSlots = new ArrayBlockingQueue<>(parallelism); - final TestingSlotOwner slotOwner = new TestingSlotOwner(); - slotOwner.setReturnAllocatedSlotConsumer( - (LogicalSlot logicalSlot) -> returnedSlots.offer(logicalSlot.getAllocationId())); - - final LogicalSlot[] sourceSlots = new LogicalSlot[parallelism]; - final LogicalSlot[] targetSlots = new LogicalSlot[parallelism]; - - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] sourceFutures = new CompletableFuture[parallelism]; - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] targetFutures = new CompletableFuture[parallelism]; - - for (int i = 0; i < parallelism; i++) { - sourceSlots[i] = createSingleLogicalSlot(slotOwner, taskManager, new SlotRequestId()); - targetSlots[i] = createSingleLogicalSlot(slotOwner, taskManager, new SlotRequestId()); - - sourceFutures[i] = new CompletableFuture<>(); - targetFutures[i] = new CompletableFuture<>(); - } - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlots(sourceVertex.getID(), sourceFutures); - slotProvider.addSlots(targetVertex.getID(), targetFutures); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - - // - // we complete some of the futures - - for (int i = 0; i < parallelism; i += 2) { - sourceFutures[i].complete(sourceSlots[i]); - targetFutures[i].complete(targetSlots[i]); - } - - // kick off the scheduling - eg.scheduleForExecution(); - - // fail one slot - sourceFutures[1].completeExceptionally(new TestRuntimeException()); - - // wait until the job failed as a whole - eg.getTerminationFuture().get(2000, TimeUnit.MILLISECONDS); - - // wait until all slots are back - for (int i = 0; i < parallelism; i++) { - returnedSlots.poll(2000L, TimeUnit.MILLISECONDS); - } - - // no deployment calls must have happened - assertThat(taskManager.getSubmitTaskCount(), is(0)); - - // all completed futures must have been returns - for (int i = 0; i < parallelism; i += 2) { - assertFalse(sourceSlots[i].isAlive()); - assertFalse(targetSlots[i].isAlive()); - } - } - - /** - * This tests makes sure that with eager scheduling no task is deployed if a single - * slot allocation fails. Moreover we check that allocated slots will be returned. - */ - @Test - public void testEagerSchedulingWithSlotTimeout() throws Exception { - - // we construct a simple graph: (task) - - final int parallelism = 3; - - final JobVertex vertex = new JobVertex("task"); - vertex.setParallelism(parallelism); - vertex.setInvokableClass(NoOpInvokable.class); - - final JobID jobId = new JobID(); - final JobGraph jobGraph = new JobGraph(jobId, "test", vertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - final BlockingQueue<AllocationID> returnedSlots = new ArrayBlockingQueue<>(2); - final TestingSlotOwner slotOwner = new TestingSlotOwner(); - slotOwner.setReturnAllocatedSlotConsumer( - (LogicalSlot logicalSlot) -> returnedSlots.offer(logicalSlot.getAllocationId())); - - final InteractionsCountingTaskManagerGateway taskManager = createTaskManager(); - final LogicalSlot[] slots = new LogicalSlot[parallelism]; - @SuppressWarnings({"unchecked", "rawtypes"}) - final CompletableFuture<LogicalSlot>[] slotFutures = new CompletableFuture[parallelism]; - - for (int i = 0; i < parallelism; i++) { - slots[i] = createSingleLogicalSlot(slotOwner, taskManager, new SlotRequestId()); - slotFutures[i] = new CompletableFuture<>(); - } - - ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - slotProvider.addSlots(vertex.getID(), slotFutures); - - final ExecutionGraph eg = createExecutionGraph(jobGraph, slotProvider); - - // we complete one future - slotFutures[1].complete(slots[1]); - - // kick off the scheduling - eg.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - eg.scheduleForExecution(); - - // we complete another future - slotFutures[2].complete(slots[2]); - - // check that the ExecutionGraph is not terminated yet - assertThat(eg.getTerminationFuture().isDone(), is(false)); - - // time out one of the slot futures - slotFutures[0].completeExceptionally(new TimeoutException("Test time out")); - - assertThat(eg.getTerminationFuture().get(), is(JobStatus.FAILED)); - - // wait until all slots are back - for (int i = 0; i < parallelism - 1; i++) { - returnedSlots.poll(2000, TimeUnit.MILLISECONDS); - } - - // verify that no deployments have happened - assertThat(taskManager.getSubmitTaskCount(), is(0)); - } - - /** - * Tests that an ongoing scheduling operation does not fail the {@link ExecutionGraph} - * if it gets concurrently cancelled. - */ - @Test - public void testSchedulingOperationCancellationWhenCancel() throws Exception { - final JobVertex jobVertex = new JobVertex("NoOp JobVertex"); - jobVertex.setInvokableClass(NoOpInvokable.class); - jobVertex.setParallelism(2); - final JobGraph jobGraph = new JobGraph(jobVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - final CompletableFuture<LogicalSlot> slotFuture1 = new CompletableFuture<>(); - final CompletableFuture<LogicalSlot> slotFuture2 = new CompletableFuture<>(); - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(2); - slotProvider.addSlots(jobVertex.getID(), new CompletableFuture[]{slotFuture1, slotFuture2}); - final ExecutionGraph executionGraph = createExecutionGraph(jobGraph, slotProvider); - - executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - executionGraph.scheduleForExecution(); - - final TestingLogicalSlot slot = createTestingSlot(); - final CompletableFuture<?> releaseFuture = slot.getReleaseFuture(); - slotFuture1.complete(slot); - - // cancel should change the state of all executions to CANCELLED - executionGraph.cancel(); - - // complete the now CANCELLED execution --> this should cause a failure - slotFuture2.complete(new TestingLogicalSlotBuilder().createTestingLogicalSlot()); - - Thread.sleep(1L); - // release the first slot to finish the cancellation - releaseFuture.complete(null); - - // NOTE: This test will only occasionally fail without the fix since there is - // a race between the releaseFuture and the slotFuture2 - assertThat(executionGraph.getTerminationFuture().get(), is(JobStatus.CANCELED)); - } - - /** - * Tests that a partially completed eager scheduling operation fails if a - * completed slot is released. See FLINK-9099. - */ - @Test - public void testSlotReleasingFailsSchedulingOperation() throws Exception { - final int parallelism = 2; - - final JobVertex jobVertex = new JobVertex("Testing job vertex"); - jobVertex.setInvokableClass(NoOpInvokable.class); - jobVertex.setParallelism(parallelism); - final JobGraph jobGraph = new JobGraph(jobVertex); - jobGraph.setScheduleMode(ScheduleMode.EAGER); - - final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); - - final LogicalSlot slot = createSingleLogicalSlot(new DummySlotOwner(), new SimpleAckingTaskManagerGateway(), new SlotRequestId()); - slotProvider.addSlot(jobVertex.getID(), 0, CompletableFuture.completedFuture(slot)); - - final CompletableFuture<LogicalSlot> slotFuture = new CompletableFuture<>(); - slotProvider.addSlot(jobVertex.getID(), 1, slotFuture); - - final ExecutionGraph executionGraph = createExecutionGraph(jobGraph, slotProvider); - - executionGraph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread()); - executionGraph.scheduleForExecution(); - - assertThat(executionGraph.getState(), is(JobStatus.RUNNING)); - - final ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertex.getID()); - final ExecutionVertex[] taskVertices = executionJobVertex.getTaskVertices(); - assertThat(taskVertices[0].getExecutionState(), is(ExecutionState.SCHEDULED)); - assertThat(taskVertices[1].getExecutionState(), is(ExecutionState.SCHEDULED)); - - // fail the single allocated slot --> this should fail the scheduling operation - slot.releaseSlot(new FlinkException("Test failure")); - - assertThat(executionGraph.getTerminationFuture().get(), is(JobStatus.FAILED)); - } - - /** - * Tests that all slots are being returned to the {@link SlotOwner} if the - * {@link ExecutionGraph} is being cancelled. See FLINK-9908 - */ - @Test - public void testCancellationOfIncompleteScheduling() throws Exception { Review comment: How does the `DefaultScheduler` cancels pending slot requests if the user calls `DefaultScheduler.cancel()`? ---------------------------------------------------------------- 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
