[FLINK-8649] Scala StreamExecutionEnvironment.createInput should pass on the TypeInfo

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 1.4.0
• Fix Version/s: 1.5.0
• Component/s: API / Scala
• Assignee: Gábor Gévay
• Reporter: Gábor Gévay
• Created: 13/Feb/18 17:30
• Updated: 13/Apr/21 20:40
• Resolved: 21/Mar/18 22:41

Description

This is StreamExecutionEnvironment.createInput in the Scala API:

def createInput[T: TypeInformation](inputFormat: InputFormat[T, _]): DataStream[T] =
  asScalaStream(javaEnv.createInput(inputFormat))

It should pass on the implicitly got TypeInformation to Java like this:

def createInput[T: TypeInformation](inputFormat: InputFormat[T, _]): DataStream[T] =
  asScalaStream(javaEnv.createInput(inputFormat, implicitly[TypeInformation[T]]))

The current situation creates a problem, for example, when we have generics in the type like in the following code, where the Java API can't deduce the TypeInformation on its own:

 StreamExecutionEnvironment.getExecutionEnvironment.createInput[Tuple2[Integer, Integer]](new ParallelIteratorInputFormat[Tuple2[Integer, Integer]](null))

Issue Links

• links to: GitHub Pull Request #5479