[FLINK-12929] Pass TypeInformation in addSource

Co-authored-by: Georg Rollinger <[email protected]>
spotify · Jul 10, 2019 · 7ea00ad · 7ea00ad
1 parent 8acc1d3
commit 7ea00ad
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 11 deletions.
diff --git a/.../src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/.../src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -1453,17 +1453,16 @@ public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, TypeI
  @SuppressWarnings("unchecked")
  public <OUT> DataStreamSource<OUT> addSource(SourceFunction<OUT> function, String sourceName, TypeInformation<OUT> typeInfo) {
 
+ if (function instanceof ResultTypeQueryable) {
+ typeInfo = ((ResultTypeQueryable<OUT>) function).getProducedType();
+ }
  if (typeInfo == null) {
- if (function instanceof ResultTypeQueryable) {
- typeInfo = ((ResultTypeQueryable<OUT>) function).getProducedType();
- } else {
- try {
- typeInfo = TypeExtractor.createTypeInfo(
- SourceFunction.class,
- function.getClass(), 0, null, null);
- } catch (final InvalidTypesException e) {
- typeInfo = (TypeInformation<OUT>) new MissingTypeInfo(sourceName, e);
- }
+ try {
+ typeInfo = TypeExtractor.createTypeInfo(
+ SourceFunction.class,
+ function.getClass(), 0, null, null);
+ } catch (final InvalidTypesException e) {
+ typeInfo = (TypeInformation<OUT>) new MissingTypeInfo(sourceName, e);
  }
  }
 

diff --git a/...cala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/...cala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -615,7 +615,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
 
  val cleanFun = scalaClean(function)
  val typeInfo = implicitly[TypeInformation[T]]
- asScalaStream(javaEnv.addSource(cleanFun).returns(typeInfo))
+ asScalaStream(javaEnv.addSource(cleanFun, typeInfo))
  }
 
  /**