[FLINK-8649] [scala api] Pass on TypeInfo in StreamExecutionEnvironme…

…nt.createInput This closes apache#5478.
static-max · Mar 21, 2018 · 7e43f81 · 7e43f81
1 parent 129e215
commit 7e43f81
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 1 deletion.
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
@@ -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.java.typeutils.ResultTypeQueryable
 import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
 import org.apache.flink.api.scala.ClosureCleaner
 import org.apache.flink.configuration.Configuration
@@ -594,7 +595,11 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
  */
  @PublicEvolving
  def createInput[T: TypeInformation](inputFormat: InputFormat[T, _]): DataStream[T] =
- asScalaStream(javaEnv.createInput(inputFormat))
+ if (inputFormat.isInstanceOf[ResultTypeQueryable[_]]) {
+ asScalaStream(javaEnv.createInput(inputFormat))
+ } else {
+ asScalaStream(javaEnv.createInput(inputFormat, implicitly[TypeInformation[T]]))
+ }
 
  /**
  * Create a DataStream using a user defined source function for arbitrary

diff --git a/...-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala b/...-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/DataStreamTest.scala
@@ -21,6 +21,7 @@ package org.apache.flink.streaming.api.scala
 import java.lang
 
 import org.apache.flink.api.common.functions._
+import org.apache.flink.api.java.io.ParallelIteratorInputFormat
 import org.apache.flink.api.java.typeutils.TypeExtractor
 import org.apache.flink.streaming.api.collector.selector.OutputSelector
 import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
@@ -673,6 +674,12 @@ class DataStreamTest extends AbstractTestBase {
  assert(sg.getIterationSourceSinkPairs.size() == 2)
  }
 
+ @Test
+ def testCreateInputPassesOnTypeInfo(): Unit = {
+ StreamExecutionEnvironment.getExecutionEnvironment.createInput[Tuple1[Integer]](
+ new ParallelIteratorInputFormat[Tuple1[Integer]](null))
+ }
+
  /////////////////////////////////////////////////////////////
  // Utilities
  /////////////////////////////////////////////////////////////