GJL commented on a change in pull request #8309: [FLINK-12229] [runtime] Implement LazyFromSourcesScheduling Strategy URL: https://github.com/apache/flink/pull/8309#discussion_r283453780
########## File path: flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/strategy/LazyFromSourcesSchedulingStrategyTest.java ########## @@ -0,0 +1,337 @@ +/* + * 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.scheduler.strategy; + +import org.apache.flink.api.common.InputDependencyConstraint; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.io.network.partition.ResultPartitionID; +import org.apache.flink.runtime.io.network.partition.ResultPartitionType; +import org.apache.flink.runtime.scheduler.ExecutionVertexDeploymentOption; +import org.apache.flink.util.TestLogger; + +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.flink.api.common.InputDependencyConstraint.ANY; +import static org.apache.flink.runtime.io.network.partition.ResultPartitionType.BLOCKING; +import static org.apache.flink.runtime.io.network.partition.ResultPartitionType.PIPELINED; +import static org.apache.flink.runtime.scheduler.strategy.StrategyTestUtil.connectConsumerVerticesToPartition; +import static org.apache.flink.runtime.scheduler.strategy.StrategyTestUtil.getExecutionVertexIdsFromDeployOptions; +import static org.apache.flink.runtime.scheduler.strategy.StrategyTestUtil.initVerticesAndPartitions; +import static org.hamcrest.Matchers.containsInAnyOrder; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; + +/** + * Unit tests for {@link LazyFromSourcesSchedulingStrategy}. + */ +public class LazyFromSourcesSchedulingStrategyTest extends TestLogger { + + /** + * Tests that when start scheduling lazy from sources scheduling strategy will start input vertices in scheduling topology. + */ + @Test + public void testStartScheduling() { + final int jobVertexCnt = 2; + final int taskCnt = 3; + TestingSchedulingTopology schedulingTopology = new TestingSchedulingTopology(); + TestingSchedulingExecutionVertex[][] vertices = new TestingSchedulingExecutionVertex[jobVertexCnt][taskCnt]; + TestingSchedulingResultPartition[][] partitions = new TestingSchedulingResultPartition[jobVertexCnt - 1][taskCnt]; + + ResultPartitionType[] partitionTypes = new ResultPartitionType[jobVertexCnt - 1]; + InputDependencyConstraint[] inputDependencyConstraints = new InputDependencyConstraint[jobVertexCnt]; + partitionTypes[0] = BLOCKING; + inputDependencyConstraints[0] = ANY; + inputDependencyConstraints[1] = ANY; + + initVerticesAndPartitions(schedulingTopology, partitionTypes, inputDependencyConstraints, vertices, partitions); + + for (int i = 0; i < taskCnt; i++) { + connectConsumerVerticesToPartition(vertices[1][i], partitions[0][i]); + } + + TestingSchedulerOperations testingSchedulerOperation = new TestingSchedulerOperations(); + LazyFromSourcesSchedulingStrategy schedulingStrategy = new LazyFromSourcesSchedulingStrategy( + testingSchedulerOperation, + schedulingTopology); + + schedulingStrategy.startScheduling(); + + Set<ExecutionVertexID> toBeScheduledVertices = Arrays.stream(vertices[0]) + .map(TestingSchedulingExecutionVertex::getId).collect(Collectors.toSet()); + Collection<ExecutionVertexDeploymentOption> scheduledVertices = testingSchedulerOperation.getScheduledVertices().get(0); + + assertThat(getExecutionVertexIdsFromDeployOptions(scheduledVertices), + containsInAnyOrder(toBeScheduledVertices.toArray())); + } + + /** + * Tests that when restart tasks will only schedule input ready vertices in given ones. + */ + @Test + public void testRestartTasks() { + final int jobVertexCnt = 2; + final int taskCnt = 3; + TestingSchedulingTopology schedulingTopology = new TestingSchedulingTopology(); + TestingSchedulingExecutionVertex[][] vertices = new TestingSchedulingExecutionVertex[jobVertexCnt][taskCnt]; + TestingSchedulingResultPartition[][] partitions = new TestingSchedulingResultPartition[jobVertexCnt - 1][taskCnt]; + + ResultPartitionType[] partitionTypes = new ResultPartitionType[jobVertexCnt - 1]; + InputDependencyConstraint[] inputDependencyConstraints = new InputDependencyConstraint[jobVertexCnt]; + partitionTypes[0] = BLOCKING; + inputDependencyConstraints[0] = ANY; + inputDependencyConstraints[1] = ANY; + + initVerticesAndPartitions(schedulingTopology, partitionTypes, inputDependencyConstraints, vertices, partitions); + + for (int i = 0; i < taskCnt; i++) { + connectConsumerVerticesToPartition(vertices[1][i], partitions[0][i]); + } + + TestingSchedulerOperations testingSchedulerOperation = new TestingSchedulerOperations(); + LazyFromSourcesSchedulingStrategy schedulingStrategy = new LazyFromSourcesSchedulingStrategy( + testingSchedulerOperation, + schedulingTopology); + + schedulingStrategy.startScheduling(); + + Set<ExecutionVertexID> toBeScheduledVertices = Arrays.stream(vertices[0]) + .map(TestingSchedulingExecutionVertex::getId).collect(Collectors.toSet()); + Collection<ExecutionVertexDeploymentOption> scheduledVertices = testingSchedulerOperation.getScheduledVertices().get(0); + assertThat(getExecutionVertexIdsFromDeployOptions(scheduledVertices), + containsInAnyOrder(toBeScheduledVertices.toArray())); + + Set<ExecutionVertexID> verticesToRestart = new HashSet<>(); + for (TestingSchedulingExecutionVertex[] vArray : vertices) { + for (TestingSchedulingExecutionVertex v : vArray) { + verticesToRestart.add(v.getId()); + } + } + schedulingStrategy.restartTasks(verticesToRestart); + scheduledVertices = testingSchedulerOperation.getScheduledVertices().get(1); + assertThat(getExecutionVertexIdsFromDeployOptions(scheduledVertices), + containsInAnyOrder(toBeScheduledVertices.toArray())); + } + + /** + * Tests that when partition consumable notified will start available {@link PIPELINED} downstream vertices. + */ + @Test + public void testPIPELINEDPartitionConsumable() { Review comment: This test and other tests suffer from a few issues: - Tests are still a bit too long - `initVerticesAndPartitions()` adds some automation but it is difficult to use and hard to understand - One of the issues of this function is that it uses output arguments (see Clean Code by Robert C Martin, Chapter 3, page 45) - Intuitively, I would say that if `SchedulingIntermediateDataSet` was unit tested separately, we do not have to make all test topologies parallel, i.e., only one execution vertex per operator - This test sets all producer vertices to `RUNNING`, which is not needed for this test. I have thought about how to improve the test, and you can find my prototype below. I am not sure if my approach can cover everything needed for all required test scenarios. Please have a look and let me know what you think. ``` // // actual test code // @Test public void testPipelinedPartitionConsumable() { final TestingSchedulingTopology testingSchedulingTopology = new TestingSchedulingTopology(); final List<TestingSchedulingExecutionVertex> producers = testingSchedulingTopology.addExecutionVertices().withParallelism(2).finish(); final List<TestingSchedulingExecutionVertex> consumers = testingSchedulingTopology.addExecutionVertices().withParallelism(2).finish(); testingSchedulingTopology.connectAllToAll(producers, consumers).withResultPartitionType(PIPELINED).finish(); final LazyFromSourcesSchedulingStrategy schedulingStrategy = startScheduling(testingSchedulingTopology); final TestingSchedulingExecutionVertex producer1 = producers.get(0); final SchedulingResultPartition partition1 = producer1.getProducedResultPartitions().iterator().next(); schedulingStrategy.onExecutionStateChange(producer1.getId(), ExecutionState.RUNNING); schedulingStrategy.onPartitionConsumable(producer1.getId(), new ResultPartitionID(partition1.getId(), new ExecutionAttemptID())); // an idea on how to make the assertions more readible // not insisting on it, could be added later assertThat(testingSchedulerOperation, hasScheduledVertices(consumers)); } // // hamcrest matcher // private static Matcher<TestingSchedulerOperations> hasScheduledVertices(final List<TestingSchedulingExecutionVertex> consumers) { final Matcher<Iterable<? extends ExecutionVertexID>> vertexIdMatcher = containsInAnyOrder(consumers.stream() .map(SchedulingExecutionVertex::getId) .toArray(ExecutionVertexID[]::new)); return new TypeSafeDiagnosingMatcher<TestingSchedulerOperations>() { @Override protected boolean matchesSafely(final TestingSchedulerOperations item, final Description mismatchDescription) { final boolean matches = vertexIdMatcher.matches(toExecutionVertexIDs(item.getLatestScheduledVertices())); if (!matches) { vertexIdMatcher.describeMismatch(item.getLatestScheduledVertices(), mismatchDescription); } return matches; } @Override public void describeTo(final Description description) { vertexIdMatcher.describeTo(description); } }; } // // TestingSchedulingTopology // public SchedulingExecutionVerticesBuilder addExecutionVertices() { return new SchedulingExecutionVerticesBuilder(); } public ProducerConsumerConnectionBuilder connectAllToAll( final List<TestingSchedulingExecutionVertex> producers, final List<TestingSchedulingExecutionVertex> consumers) { return new ProducerConsumerAllToAllConnectionBuilder(producers, consumers); } public abstract class ProducerConsumerConnectionBuilder { protected final List<TestingSchedulingExecutionVertex> producers; protected final List<TestingSchedulingExecutionVertex> consumers; protected ResultPartitionType resultPartitionType; protected ProducerConsumerConnectionBuilder( final List<TestingSchedulingExecutionVertex> producers, final List<TestingSchedulingExecutionVertex> consumers) { this.producers = producers; this.consumers = consumers; } public ProducerConsumerConnectionBuilder withResultPartitionType(final ResultPartitionType resultPartitionType) { this.resultPartitionType = resultPartitionType; return this; } public List<TestingSchedulingResultPartition> finish() { final List<TestingSchedulingResultPartition> resultPartitions = connect(); TestingSchedulingTopology.this.addSchedulingExecutionVertices(producers); TestingSchedulingTopology.this.addSchedulingExecutionVertices(consumers); return resultPartitions; } protected TestingSchedulingResultPartition.Builder initTestingSchedulingResultPartitionBuilder() { return new TestingSchedulingResultPartition.Builder() .withResultPartitionType(resultPartitionType); } protected abstract List<TestingSchedulingResultPartition> connect(); } private class ProducerConsumerAllToAllConnectionBuilder extends ProducerConsumerConnectionBuilder { private ProducerConsumerAllToAllConnectionBuilder( final List<TestingSchedulingExecutionVertex> producers, final List<TestingSchedulingExecutionVertex> consumers) { super(producers, consumers); } @Override protected List<TestingSchedulingResultPartition> connect() { final List<TestingSchedulingResultPartition> resultPartitions = new ArrayList<>(); final IntermediateDataSetID intermediateDataSetId = new IntermediateDataSetID(); for (TestingSchedulingExecutionVertex producer : producers) { final TestingSchedulingResultPartition resultPartition = initTestingSchedulingResultPartitionBuilder() .withIntermediateDataSetID(intermediateDataSetId) .build(); resultPartition.setProducer(producer); producer.addProducedPartition(resultPartition); for (TestingSchedulingExecutionVertex consumer : consumers) { consumer.addConsumedPartition(resultPartition); resultPartition.addConsumer(consumer); resultPartitions.add(resultPartition); } } return resultPartitions; } } public class SchedulingExecutionVerticesBuilder { private final JobVertexID jobVertexId = new JobVertexID(); private int parallelism = 1; private InputDependencyConstraint inputDependencyConstraint; public SchedulingExecutionVerticesBuilder withParallelism(final int parallelism) { this.parallelism = parallelism; return this; } public SchedulingExecutionVerticesBuilder withInputDependencyConstraint(final InputDependencyConstraint inputDependencyConstraint) { this.inputDependencyConstraint = inputDependencyConstraint; return this; } public List<TestingSchedulingExecutionVertex> finish() { final List<TestingSchedulingExecutionVertex> vertices = new ArrayList<>(); for (int subtaskIndex = 0; subtaskIndex < parallelism; subtaskIndex++) { vertices.add(new TestingSchedulingExecutionVertex(jobVertexId, subtaskIndex, inputDependencyConstraint)); } TestingSchedulingTopology.this.addSchedulingExecutionVertices(vertices); return vertices; } } ``` ---------------------------------------------------------------- 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 With regards, Apache Git Services
