>> chainedOperatorHashes = new HashMap<>();
-
- setChaining(hashes, legacyHashes, chainedOperatorHashes);
+ setChaining(hashes, legacyHashes);
setPhysicalEdges();
@@ -251,20 +250,17 @@ public class StreamingJobGraphGenerator {
*
* This will recursively create all {@link JobVertex} instances.
*/
- private void setChaining(Map hashes, List> legacyHashes, Map>> chainedOperatorHashes) {
+ private void setChaining(Map hashes, List> legacyHashes) {
for (Integer sourceNodeId : streamGraph.getSourceIDs()) {
- createChain(sourceNodeId, sourceNodeId, hashes, legacyHashes, 0, chainedOperatorHashes);
+ createChain(
+ sourceNodeId,
+ 0,
+ new OperatorChainInfo(sourceNodeId, hashes, legacyHashes, streamGraph));
}
}
- private List createChain(
- Integer startNodeId,
- Integer currentNodeId,
- Map hashes,
- List> legacyHashes,
- int chainIndex,
- Map>> chainedOperatorHashes) {
-
+ private List createChain(Integer currentNodeId, int chainIndex, OperatorChainInfo chainInfo) {
+ Integer startNodeId = chainInfo.getStartNodeId();
if (!builtVertices.contains(startNodeId)) {
List transitiveOutEdges = new ArrayList();
@@ -284,28 +280,20 @@ public class StreamingJobGraphGenerator {
for (StreamEdge chainable : chainableOutputs) {
transitiveOutEdges.addAll(
- createChain(startNodeId, chainable.getTargetId(), hashes, legacyHashes, chainIndex + 1, chainedOperatorHashes));
+ createChain(chainable.getTargetId(), chainIndex + 1, chainInfo));
}
for (StreamEdge nonChainable : nonChainableOutputs) {
transitiveOutEdges.add(nonChainable);
- createChain(nonChainable.getTargetId(), nonChainable.getTargetId(), hashes, legacyHashes, 0, chainedOperatorHashes);
- }
-
- List> operatorHashes =
- chainedOperatorHashes.computeIfAbsent(startNodeId, k -> new ArrayList<>());
-
- byte[] primaryHashBytes = hashes.get(currentNodeId);
- OperatorID currentOperatorId = new OperatorID(primaryHashBytes);
-
- for (Map legacyHash : legacyHashes) {
- operatorHashes.add(new Tuple2<>(primaryHashBytes, legacyHash.get(currentNodeId)));
+ createChain(nonChainable.getTargetId(), 0, chainInfo.newChain(nonChainable.getTargetId()));
}
chainedNames.put(currentNodeId, createChainedName(currentNodeId, chainableOutputs));
chainedMinResources.put(currentNodeId, createChainedMinResources(currentNodeId, chainableOutputs));
chainedPreferredResources.put(currentNodeId, createChainedPreferredResources(currentNodeId, chainableOutputs));
+ OperatorID currentOperatorId = chainInfo.addNodeToChain(currentNodeId, chainedNames.get(currentNodeId));
+
if (currentNode.getInputFormat() != null) {
getOrCreateFormatContainer(startNodeId).addInputFormat(currentOperatorId, currentNode.getInputFormat());
}
@@ -315,7 +303,7 @@ public class StreamingJobGraphGenerator {
}
StreamConfig config = currentNodeId.equals(startNodeId)
- ? createJobVertex(startNodeId, hashes, legacyHashes, chainedOperatorHashes)
+ ? createJobVertex(startNodeId, chainInfo)
: new StreamConfig(new Configuration());
setVertexConfig(currentNodeId, config, chainableOutputs, nonChainableOutputs);
@@ -393,14 +381,12 @@ public class StreamingJobGraphGenerator {
private StreamConfig createJobVertex(
Integer streamNodeId,
- Map hashes,
- List> legacyHashes,
- Map>> chainedOperatorHashes) {
+ OperatorChainInfo chainInfo) {
JobVertex jobVertex;
StreamNode streamNode = streamGraph.getStreamNode(streamNodeId);
- byte[] hash = hashes.get(streamNodeId);
+ byte[] hash = chainInfo.getHash(streamNodeId);
if (hash == null) {
throw new IllegalStateException("Cannot find node hash. " +
@@ -409,7 +395,7 @@ public class StreamingJobGraphGenerator {
JobVertexID jobVertexId = new JobVertexID(hash);
- List> chainedOperators = chainedOperatorHashes.get(streamNodeId);
+ List> chainedOperators = chainInfo.getChainedOperatorHashes(streamNodeId);
List operatorIDPairs = new ArrayList<>();
if (chainedOperators != null) {
for (Tuple2 chainedOperator : chainedOperators) {
@@ -434,6 +420,15 @@ public class StreamingJobGraphGenerator {
operatorIDPairs);
}
+ for (OperatorCoordinator.Provider coordinatorProvider : chainInfo.getCoordinatorProviders()) {
+ try {
+ jobVertex.addOperatorCoordinator(new SerializedValue<>(coordinatorProvider));
+ } catch (IOException e) {
+ throw new FlinkRuntimeException(String.format(
+ "Coordinator Provider for node %s is not serializable.", chainedNames.get(streamNodeId)));
+ }
+ }
+
jobVertex.setResources(chainedMinResources.get(streamNodeId), chainedPreferredResources.get(streamNodeId));
jobVertex.setInvokableClass(streamNode.getJobVertexClass());
@@ -973,4 +968,67 @@ public class StreamingJobGraphGenerator {
jobGraph.setSnapshotSettings(settings);
}
+
+ /**
+ * A private class to help maintain the information of an operator chain during the recursive call in
+ * {@link #createChain(Integer, int, OperatorChainInfo)}.
+ */
+ private static class OperatorChainInfo {
+ private final Integer startNodeId;
+ private final Map hashes;
+ private final List> legacyHashes;
+ private final Map>> chainedOperatorHashes;
+ private final List coordinatorProviders;
+ private final StreamGraph streamGraph;
+
+ private OperatorChainInfo(
+ int startNodeId,
+ Map hashes,
+ List> legacyHashes,
+ StreamGraph streamGraph) {
+ this.startNodeId = startNodeId;
+ this.hashes = hashes;
+ this.legacyHashes = legacyHashes;
+ this.chainedOperatorHashes = new HashMap<>();
+ this.coordinatorProviders = new ArrayList<>();
+ this.streamGraph = streamGraph;
+ }
+
+ byte[] getHash(Integer streamNodeId) {
+ return hashes.get(streamNodeId);
+ }
+
+ private Integer getStartNodeId() {
+ return startNodeId;
+ }
+
+ private List> getChainedOperatorHashes(int startNodeId) {
+ return chainedOperatorHashes.get(startNodeId);
+ }
+
+ private List getCoordinatorProviders() {
+ return coordinatorProviders;
+ }
+
+ private OperatorID addNodeToChain(int currentNodeId, String operatorName) {
+ List> operatorHashes =
+ chainedOperatorHashes.computeIfAbsent(startNodeId, k -> new ArrayList<>());
+
+ byte[] primaryHashBytes = hashes.get(currentNodeId);
+
+ for (Map legacyHash : legacyHashes) {
+ operatorHashes.add(new Tuple2<>(primaryHashBytes, legacyHash.get(currentNodeId)));
+ }
+
+ streamGraph
+ .getStreamNode(currentNodeId)
+ .getCoordinatorProvider(operatorName, new OperatorID(getHash(currentNodeId)))
+ .map(coordinatorProviders::add);
+ return new OperatorID(primaryHashBytes);
+ }
+
+ private OperatorChainInfo newChain(Integer startNodeId) {
+ return new OperatorChainInfo(startNodeId, hashes, legacyHashes, streamGraph);
+ }
+ }
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
index e4971d2cc9996b59b76d0d725da2b5d7ccaea5b0..d8907ac0d662a28aabb919f476389ff1f81f46b6 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -64,6 +64,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
public class SourceOperator
extends AbstractStreamOperator
implements OperatorEventHandler, PushingAsyncDataInput {
+ private static final long serialVersionUID = 1405537676017904695L;
// Package private for unit test.
static final ListStateDescriptor SPLITS_STATE_DESC =
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java
new file mode 100644
index 0000000000000000000000000000000000000000..f43919af92cdfa622d3df446441c18952e8ef0b8
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/LegacySourceTransformation.java
@@ -0,0 +1,90 @@
+/*
+ * 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.streaming.api.transformations;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.streaming.api.operators.ChainingStrategy;
+import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamSource;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * This represents a Source. This does not actually transform anything since it has no inputs but
+ * it is the root {@code Transformation} of any topology.
+ *
+ * @param The type of the elements that this source produces
+ */
+@Internal
+public class LegacySourceTransformation extends PhysicalTransformation {
+
+ private final StreamOperatorFactory operatorFactory;
+
+ /**
+ * Creates a new {@code LegacySourceTransformation} from the given operator.
+ *
+ * @param name The name of the {@code LegacySourceTransformation}, this will be shown in Visualizations and the Log
+ * @param operator The {@code StreamSource} that is the operator of this Transformation
+ * @param outputType The type of the elements produced by this {@code LegacySourceTransformation}
+ * @param parallelism The parallelism of this {@code LegacySourceTransformation}
+ */
+ public LegacySourceTransformation(
+ String name,
+ StreamSource operator,
+ TypeInformation outputType,
+ int parallelism) {
+ this(name, SimpleOperatorFactory.of(operator), outputType, parallelism);
+ }
+
+ public LegacySourceTransformation(
+ String name,
+ StreamOperatorFactory operatorFactory,
+ TypeInformation outputType,
+ int parallelism) {
+ super(name, outputType, parallelism);
+ this.operatorFactory = operatorFactory;
+ }
+
+ @VisibleForTesting
+ public StreamSource getOperator() {
+ return (StreamSource) ((SimpleOperatorFactory) operatorFactory).getOperator();
+ }
+
+ /**
+ * Returns the {@code StreamOperatorFactory} of this {@code LegacySourceTransformation}.
+ */
+ public StreamOperatorFactory getOperatorFactory() {
+ return operatorFactory;
+ }
+
+ @Override
+ public Collection> getTransitivePredecessors() {
+ return Collections.singleton(this);
+ }
+
+ @Override
+ public final void setChainingStrategy(ChainingStrategy strategy) {
+ operatorFactory.setChainingStrategy(strategy);
+ }
+}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java
index 94cb93229978c10c630d1ce6a15926be179a5fca..2b82e2755c14c4533203783c894a124927487ae0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/transformations/SourceTransformation.java
@@ -1,81 +1,62 @@
/*
- * 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.
+ 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.streaming.api.transformations;
import org.apache.flink.annotation.Internal;
-import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.dag.Transformation;
import org.apache.flink.streaming.api.operators.ChainingStrategy;
-import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
-import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
-import org.apache.flink.streaming.api.operators.StreamSource;
+import org.apache.flink.streaming.api.operators.SourceOperator;
+import org.apache.flink.streaming.api.operators.SourceOperatorFactory;
import java.util.Collection;
import java.util.Collections;
/**
- * This represents a Source. This does not actually transform anything since it has no inputs but
- * it is the root {@code Transformation} of any topology.
- *
- * @param The type of the elements that this source produces
+ * A {@link PhysicalTransformation} for {@link Source}.
*/
@Internal
-public class SourceTransformation extends PhysicalTransformation {
-
- private final StreamOperatorFactory operatorFactory;
-
+public class SourceTransformation extends PhysicalTransformation {
+ private final SourceOperatorFactory sourceFactory;
/**
- * Creates a new {@code SourceTransformation} from the given operator.
+ * Creates a new {@code Transformation} with the given name, output type and parallelism.
*
- * @param name The name of the {@code SourceTransformation}, this will be shown in Visualizations and the Log
- * @param operator The {@code StreamSource} that is the operator of this Transformation
- * @param outputType The type of the elements produced by this {@code SourceTransformation}
- * @param parallelism The parallelism of this {@code SourceTransformation}
+ * @param name The name of the {@code Transformation}, this will be shown in Visualizations and the Log
+ * @param sourceFactory The operator factory for {@link SourceOperator}.
+ * @param outputType The output type of this {@code Transformation}
+ * @param parallelism The parallelism of this {@code Transformation}
*/
public SourceTransformation(
String name,
- StreamSource operator,
- TypeInformation outputType,
- int parallelism) {
- this(name, SimpleOperatorFactory.of(operator), outputType, parallelism);
- }
-
- public SourceTransformation(
- String name,
- StreamOperatorFactory operatorFactory,
- TypeInformation outputType,
+ SourceOperatorFactory sourceFactory,
+ TypeInformation outputType,
int parallelism) {
super(name, outputType, parallelism);
- this.operatorFactory = operatorFactory;
- }
-
- @VisibleForTesting
- public StreamSource getOperator() {
- return (StreamSource) ((SimpleOperatorFactory) operatorFactory).getOperator();
+ this.sourceFactory = sourceFactory;
}
/**
- * Returns the {@code StreamOperatorFactory} of this {@code SourceTransformation}.
+ * Returns the {@code StreamOperatorFactory} of this {@code LegacySourceTransformation}.
*/
- public StreamOperatorFactory getOperatorFactory() {
- return operatorFactory;
+ public SourceOperatorFactory getOperatorFactory() {
+ return sourceFactory;
}
@Override
@@ -85,6 +66,6 @@ public class SourceTransformation extends PhysicalTransformation {
@Override
public final void setChainingStrategy(ChainingStrategy strategy) {
- operatorFactory.setChainingStrategy(strategy);
+ sourceFactory.setChainingStrategy(strategy);
}
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
index 3ee0002b0c24eecb097610a8a8711c3e73aa93f0..1fb977a9a46168dc81f4855f69061986a449cc6e 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java
@@ -28,6 +28,9 @@ import org.apache.flink.api.common.operators.ResourceSpec;
import org.apache.flink.api.common.operators.util.UserCodeWrapper;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.mocks.MockSource;
+import org.apache.flink.api.dag.Transformation;
import org.apache.flink.api.java.io.DiscardingOutputFormat;
import org.apache.flink.api.java.io.TypeSerializerInputFormat;
import org.apache.flink.api.java.tuple.Tuple2;
@@ -43,6 +46,7 @@ import org.apache.flink.runtime.jobgraph.ScheduleMode;
import org.apache.flink.runtime.jobgraph.tasks.JobCheckpointingSettings;
import org.apache.flink.runtime.jobmanager.scheduler.CoLocationGroup;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
import org.apache.flink.runtime.operators.util.TaskConfig;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
@@ -56,18 +60,27 @@ import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.InputFormatSourceFunction;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
+import org.apache.flink.streaming.api.operators.AbstractStreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.CoordinatedOperatorFactory;
import org.apache.flink.streaming.api.operators.MailboxExecutor;
import org.apache.flink.streaming.api.operators.OneInputStreamOperatorFactory;
import org.apache.flink.streaming.api.operators.SimpleOperatorFactory;
+import org.apache.flink.streaming.api.operators.SourceOperatorFactory;
import org.apache.flink.streaming.api.operators.StreamMap;
+import org.apache.flink.streaming.api.operators.StreamOperator;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
+import org.apache.flink.streaming.api.operators.StreamOperatorParameters;
import org.apache.flink.streaming.api.operators.YieldingOperatorFactory;
+import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.api.transformations.ShuffleMode;
import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
import org.apache.flink.streaming.runtime.partitioner.RebalancePartitioner;
import org.apache.flink.streaming.runtime.partitioner.RescalePartitioner;
+import org.apache.flink.streaming.runtime.tasks.SourceOperatorStreamTask;
import org.apache.flink.streaming.util.TestAnyModeReadingStreamOperator;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.TestLogger;
import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
@@ -234,6 +247,26 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
assertTrue(printConfig.isChainEnd());
}
+ @Test
+ public void testOperatorCoordinatorAddedToJobVertex() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream stream =
+ env.continuousSource(new MockSource(Boundedness.BOUNDED, 1), "TestingSource");
+
+ OneInputTransformation resultTransform = new OneInputTransformation(
+ stream.getTransformation(),
+ "AnyName",
+ new CoordinatedTransformOperatorFactory(),
+ BasicTypeInfo.INT_TYPE_INFO,
+ env.getParallelism());
+
+ new TestingSingleOutputStreamOperator<>(env, resultTransform).print();
+
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(env.getStreamGraph());
+
+ assertEquals(2, jobGraph.getVerticesAsArray()[0].getOperatorCoordinators().size());
+ }
+
/**
* Verifies that the resources are merged correctly for chained operators (covers source and sink cases)
* when generating job graph.
@@ -422,6 +455,30 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
assertTrue(sinkFormat2 instanceof DiscardingOutputFormat);
}
+ @Test
+ public void testCoordinatedOperator() {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream source =
+ env.continuousSource(new MockSource(Boundedness.BOUNDED, 1), "TestSource");
+ source.addSink(new DiscardingSink<>());
+
+ StreamGraph streamGraph = env.getStreamGraph();
+ JobGraph jobGraph = StreamingJobGraphGenerator.createJobGraph(streamGraph);
+ // There should be only one job vertex.
+ assertEquals(1, jobGraph.getNumberOfVertices());
+
+ JobVertex jobVertex = jobGraph.getVerticesAsArray()[0];
+ List> coordinatorProviders = jobVertex.getOperatorCoordinators();
+ // There should be only one coordinator provider.
+ assertEquals(1, coordinatorProviders.size());
+ // The invokable class should be SourceOperatorStreamTask.
+ final ClassLoader classLoader = getClass().getClassLoader();
+ assertEquals(SourceOperatorStreamTask.class, jobVertex.getInvokableClass(classLoader));
+ StreamOperatorFactory operatorFactory =
+ new StreamConfig(jobVertex.getConfiguration()).getStreamOperatorFactory(classLoader);
+ assertTrue(operatorFactory instanceof SourceOperatorFactory);
+ }
+
/**
* Test setting shuffle mode to {@link ShuffleMode#PIPELINED}.
*/
@@ -890,4 +947,42 @@ public class StreamingJobGraphGeneratorTest extends TestLogger {
}
}
+ // ------------ private classes -------------
+ private static class CoordinatedTransformOperatorFactory
+ extends AbstractStreamOperatorFactory
+ implements CoordinatedOperatorFactory, OneInputStreamOperatorFactory {
+
+ @Override
+ public OperatorCoordinator.Provider getCoordinatorProvider(String operatorName, OperatorID operatorID) {
+ return new OperatorCoordinator.Provider() {
+ @Override
+ public OperatorID getOperatorId() {
+ return null;
+ }
+
+ @Override
+ public OperatorCoordinator create(OperatorCoordinator.Context context) {
+ return null;
+ }
+ };
+ }
+
+ @Override
+ public > T createStreamOperator(StreamOperatorParameters parameters) {
+ return null;
+ }
+
+ @Override
+ public Class getStreamOperatorClass(ClassLoader classLoader) {
+ return null;
+ }
+ }
+
+ private static class TestingSingleOutputStreamOperator extends SingleOutputStreamOperator {
+
+ public TestingSingleOutputStreamOperator(StreamExecutionEnvironment environment,
+ Transformation transformation) {
+ super(environment, transformation);
+ }
+ }
}
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 0c7dd6c5d3f87300b068654429f58695e52348a7..37bb4f99c8f4849275f3d8123334a4d5437bc9b6 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
@@ -23,6 +23,7 @@ import org.apache.flink.annotation.{Internal, Public, PublicEvolving}
import org.apache.flink.api.common.io.{FileInputFormat, FilePathFilter, InputFormat}
import org.apache.flink.api.common.restartstrategy.RestartStrategies.RestartStrategyConfiguration
import org.apache.flink.api.common.typeinfo.TypeInformation
+import org.apache.flink.api.connector.source.{Source, SourceSplit}
import org.apache.flink.api.java.typeutils.ResultTypeQueryable
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
import org.apache.flink.api.scala.ClosureCleaner
@@ -660,6 +661,15 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
addSource(sourceFunction)
}
+ /**
+ * Create a DataStream using a [[Source]].
+ */
+ def continuousSource[T: TypeInformation](
+ source: Source[T, _ <: SourceSplit, _],
+ sourceName: String): Unit = {
+ asScalaStream(javaEnv.continuousSource(source, sourceName))
+ }
+
/**
* Triggers the program execution. The environment will execute all parts of
* the program that have resulted in a "sink" operation. Sink operations are
diff --git a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecTableSourceScan.scala b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecTableSourceScan.scala
index 095add95cef24c14bfdf485da47e1b5dd7c1466e..e5020cc8266b8be7205ee3275a31a6864550764f 100644
--- a/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecTableSourceScan.scala
+++ b/flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/plan/nodes/physical/stream/StreamExecTableSourceScan.scala
@@ -79,7 +79,7 @@ class StreamExecTableSourceScan(
name: String,
outTypeInfo: RowDataTypeInfo): Transformation[RowData] = {
// It's better to use StreamExecutionEnvironment.createInput()
- // rather than addSource() for streaming, because it take care of checkpoint.
+ // rather than addLegacySource() for streaming, because it take care of checkpoint.
env
.createInput(inputFormat, outTypeInfo)
.name(name)
diff --git a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/BatchExecutorTest.java b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/BatchExecutorTest.java
index fe6b5c62df8b439f7a0ca3bcf9a417ed6157690a..7a1feb787673a835a07543ed42fc8cb31ebfadde 100644
--- a/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/BatchExecutorTest.java
+++ b/flink-table/flink-table-planner-blink/src/test/java/org/apache/flink/table/planner/delegation/BatchExecutorTest.java
@@ -25,7 +25,7 @@ import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.graph.StreamGraph;
import org.apache.flink.streaming.api.operators.StreamSource;
-import org.apache.flink.streaming.api.transformations.SourceTransformation;
+import org.apache.flink.streaming.api.transformations.LegacySourceTransformation;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.util.TestLogger;
@@ -47,7 +47,7 @@ public class BatchExecutorTest extends TestLogger {
public BatchExecutorTest() {
batchExecutor = new BatchExecutor(LocalStreamEnvironment.getExecutionEnvironment());
- final Transformation testTransform = new SourceTransformation<>(
+ final Transformation testTransform = new LegacySourceTransformation<>(
"MockTransform",
new StreamSource<>(new SourceFunction() {
@Override
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
index fa0add7a689ddcad0e4c67ff3d8c749f663ec3b1..cc1bae2cfdc4e54bbd29eb6a40ac62186b952808 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/runtime/harness/HarnessTestBase.scala
@@ -366,7 +366,7 @@ class HarnessTestBase extends StreamingWithStateTestBase {
}
case union: UnionTransformation[_] => extractFromInputs(union.getInputs.toSeq: _*)
case p: PartitionTransformation[_] => extractFromInputs(p.getInput)
- case _: SourceTransformation[_] => null
+ case _: LegacySourceTransformation[_] => null
case _ => throw new UnsupportedOperationException("This should not happen.")
}
}