[FLINK-12929] scala.StreamExecutionEnvironment.addSource does not propagate TypeInformation

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Fix Version/s: 1.9.0
• Component/s: API / Scala, API / Type Serialization System
• Labels: pull-request-available
• Assignee: Unassigned
• Reporter: Fabio Lombardelli
• Created: 21/Jun/19 15:15
• Updated: 13/Nov/19 19:20
• Resolved: 10/Jul/19 11:19

Description

In scala.StreamExecutionEnvironment.addSource I would expect that typeInfo is also passed to the javaEnv.addSource as second parameter and not only passed to the returns method:

  def addSource[T: TypeInformation](function: SourceFunction[T]): DataStream[T] = {
    require(function != null, "Function must not be null.")
    
    val cleanFun = scalaClean(function)
    val typeInfo = implicitly[TypeInformation[T]]
    asScalaStream(javaEnv.addSource(cleanFun, <missing typeInfo>).returns(typeInfo))
  }

Issue Links

• causes: FLINK-14757 scala.StreamExecutionEnvironment Doesn't set return type upon adding source
• links to: GitHub Pull Request #8914