Github user uce commented on a diff in the pull request:
https://github.com/apache/flink/pull/754#discussion_r31716687
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/jobmanager/scheduler/BatchSchedulingTest.java
---
@@ -0,0 +1,521 @@
+/*
+ * 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.jobmanager.scheduler;
+
+import org.apache.flink.runtime.client.JobExecutionException;
+import org.apache.flink.runtime.io.network.api.reader.RecordReader;
+import org.apache.flink.runtime.io.network.api.writer.RecordWriter;
+import
org.apache.flink.runtime.io.network.partition.consumer.UnionInputGate;
+import org.apache.flink.runtime.jobgraph.AbstractJobVertex;
+import org.apache.flink.runtime.jobgraph.JobGraph;
+import org.apache.flink.runtime.jobgraph.ScheduleMode;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.jobmanager.Tasks;
+import org.apache.flink.runtime.testingUtils.TestingCluster;
+import org.apache.flink.runtime.testingUtils.TestingUtils;
+import org.apache.flink.types.IntValue;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.BitSet;
+
+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.jobgraph.DistributionPattern.POINTWISE;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+public class BatchSchedulingTest {
+
+ // Test cluster config
+ private final static int NUMBER_OF_TMS = 1;
+ private final static int NUMBER_OF_SLOTS_PER_TM = 2;
+ private final static int PARALLELISM = NUMBER_OF_TMS *
NUMBER_OF_SLOTS_PER_TM;
+
+ private static TestingCluster flink;
+
+ @BeforeClass
+ public static void setUpTestCluster() throws Exception {
+ flink = TestingUtils.startTestingCluster(
+ NUMBER_OF_SLOTS_PER_TM,
+ NUMBER_OF_TMS,
+ TestingUtils.DEFAULT_AKKA_ASK_TIMEOUT());
+ }
+
+ @AfterClass
+ public static void tearDownTestCluster() throws Exception {
+ if (flink != null) {
+ flink.stop();
+ }
+ }
+
+ @Test
+ public void testBatchSchedulingScheduleModeNotSet() throws Exception {
+ // Create the JobGraph
+ JobGraph jobGraph = new
JobGraph("testBatchSchedulingScheduleModeNotSet");
+
+ AbstractJobVertex v1 = new AbstractJobVertex("v1");
+ v1.setInvokableClass(SourceTask.class);
+ v1.setParallelism(PARALLELISM);
+
+ AbstractJobVertex v2 = new AbstractJobVertex("v2");
+ v2.setInvokableClass(UnionForwarder.class);
+ v2.setParallelism(PARALLELISM);
+
+ v2.connectNewDataSetAsInput(v1, POINTWISE);
+
+ jobGraph.addVertex(v1);
+ jobGraph.addVertex(v2);
+
+ v1.setAsBatchSource();
+ v1.addBatchSuccessors(v2);
+
+ try {
+ // The execution should fail, because we configured a
successor without
+ // setting the correct schedule mode.
+ flink.submitJobAndWait(jobGraph, false,
TestingUtils.TESTING_DURATION());
+ fail("Did not throw expected Exception.");
+ }
+ catch (JobExecutionException expected) {
+ assertEquals(IllegalStateException.class,
expected.getCause().getClass());
+ }
+ }
+
+ @Test(expected = JobExecutionException.class)
+ public void testBatchSchedulingNoSourceSet() throws Exception {
+ // Create the JobGraph
+ JobGraph jobGraph = new
JobGraph("testBatchSchedulingNoSourceSet");
+ jobGraph.setScheduleMode(ScheduleMode.BATCH_FROM_SOURCES);
+
+ AbstractJobVertex src = new AbstractJobVertex("src");
+
+ jobGraph.addVertex(src);
+
+ // This should throw an Exception, because schedule mode is
BATCH,
+ // but no source has been configured.
+ flink.submitJobAndWait(jobGraph, false,
TestingUtils.TESTING_DURATION());
+ }
+
+ @Test(expected = JobExecutionException.class)
+ public void testBatchSchedulingSourceWithInput() throws Exception {
+ // Create the JobGraph
+ JobGraph jobGraph = new
JobGraph("testBatchSchedulingSourceWithInput");
+ jobGraph.setScheduleMode(ScheduleMode.BATCH_FROM_SOURCES);
+
+ AbstractJobVertex v1 = new AbstractJobVertex("v1");
+
+ AbstractJobVertex v2 = new AbstractJobVertex("v2");
+ v2.connectNewDataSetAsInput(v1, POINTWISE);
+
+ jobGraph.addVertex(v1);
+ jobGraph.addVertex(v2);
+
+ v2.setAsBatchSource();
+
+ // This should throw an Exception, because the configured
source has an input vertex.
+ flink.submitJobAndWait(jobGraph, false,
TestingUtils.TESTING_DURATION());
+ }
+
+ /**
+ * <pre>
+ * O verify
+ * |
+ * . <------------- denotes a pipelined result
+ * O union
+ * +----´|`----+
+ * | | |
+ * â â â <------- denotes a blocking result
+ * O O O
+ * src0 src1 src2
+ * </pre>
+ */
+ @Test
+ public void testBatchSchedulingLegWise() throws Exception {
+ // Create the JobGraph
+ JobGraph jobGraph = new JobGraph("testBatchSchedulingLegWise");
+ jobGraph.setScheduleMode(ScheduleMode.BATCH_FROM_SOURCES);
+
+ // Union
+ AbstractJobVertex union = new AbstractJobVertex("union");
+ union.setInvokableClass(UnionForwarder.class);
+ union.setParallelism(PARALLELISM);
+ jobGraph.addVertex(union);
+
+ // Create source vertices
+ AbstractJobVertex[] src = new AbstractJobVertex[3];
+ for (int i = 0; i < src.length; i++) {
+ src[i] = new AbstractJobVertex("src " + i);
+ src[i].setInvokableClass(SourceTask.class);
+ src[i].setParallelism(PARALLELISM);
+ src[i].getConfiguration().setInteger("index", i);
+ jobGraph.addVertex(src[i]);
+
+ // Connect sources to union node
+ union.connectNewDataSetAsInput(src[i], POINTWISE,
BLOCKING);
+ }
+
+ // Create verify
+ AbstractJobVertex verify = new AbstractJobVertex("verify");
+ verify.setInvokableClass(VerifyIndexes.class);
+ verify.setParallelism(PARALLELISM);
+ verify.getConfiguration().setInteger("maxIndex", src.length);
+
+ verify.connectNewDataSetAsInput(union, POINTWISE, PIPELINED);
+ jobGraph.addVertex(verify);
+
+ // Slot sharing group for the last pipeline
+ SlotSharingGroup ssg = new SlotSharingGroup(union.getID(),
verify.getID());
+ union.setSlotSharingGroup(ssg);
+ verify.setSlotSharingGroup(ssg);
+
+ // - Configure batch scheduling
------------------------------------------------------------
+
+ // src0 is the first to go...
+ src[0].setAsBatchSource();
+
+ src[0].addBatchSuccessors(src[1]); // src0 => src1
+
+ src[1].addBatchSuccessors(src[2]); // src1 => src2
+
+ src[2].addBatchSuccessors(union); // src2 => [union => verify]
+
+ flink.submitJobAndWait(jobGraph, false,
TestingUtils.TESTING_DURATION());
--- End diff --
This is basically up to the job graph generation. This only works with
blocking results.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
