[FLINK-14794][runtime] Fixing KeyedStream#transform to properly relay

type information if invoked with factories.
spotify · Nov 15, 2019 · 749086d · 749086d
1 parent 1f28a24
commit 749086d
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 6 deletions.
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -1244,10 +1244,11 @@ public <R> SingleOutputStreamOperator<R> transform(
  String operatorName,
  TypeInformation<R> outTypeInfo,
  OneInputStreamOperatorFactory<T, R> operatorFactory) {
+
  return doTransform(operatorName, outTypeInfo, operatorFactory);
  }
 
- private <R> SingleOutputStreamOperator<R> doTransform(
+ protected <R> SingleOutputStreamOperator<R> doTransform(
  String operatorName,
  TypeInformation<R> outTypeInfo,
  StreamOperatorFactory<R> operatorFactory) {

diff --git a/...k-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java b/...k-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
@@ -49,9 +49,9 @@
 import org.apache.flink.streaming.api.graph.StreamGraphGenerator;
 import org.apache.flink.streaming.api.operators.KeyedProcessOperator;
 import org.apache.flink.streaming.api.operators.LegacyKeyedProcessOperator;
-import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
 import org.apache.flink.streaming.api.operators.StreamGroupedFold;
 import org.apache.flink.streaming.api.operators.StreamGroupedReduce;
+import org.apache.flink.streaming.api.operators.StreamOperatorFactory;
 import org.apache.flink.streaming.api.operators.co.IntervalJoinOperator;
 import org.apache.flink.streaming.api.transformations.OneInputTransformation;
 import org.apache.flink.streaming.api.transformations.PartitionTransformation;
@@ -256,11 +256,12 @@ protected DataStream<T> setConnectionType(StreamPartitioner<T> partitioner) {
  // ------------------------------------------------------------------------
 
  @Override
- @PublicEvolving
- public <R> SingleOutputStreamOperator<R> transform(String operatorName,
- TypeInformation<R> outTypeInfo, OneInputStreamOperator<T, R> operator) {
+ protected <R> SingleOutputStreamOperator<R> doTransform(
+ final String operatorName,
+ final TypeInformation<R> outTypeInfo,
+ final StreamOperatorFactory<R> operatorFactory) {
 
- SingleOutputStreamOperator<R> returnStream = super.transform(operatorName, outTypeInfo, operator);
+ SingleOutputStreamOperator<R> returnStream = super.doTransform(operatorName, outTypeInfo, operatorFactory);
 
  // inject the key selector and key type
  OneInputTransformation<T, R> transform = (OneInputTransformation<T, R>) returnStream.getTransformation();