[FLINK-16919] Parameter 'jobName' does not take effect in BatchTableEnvUtil

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Invalid
• Component/s: Table SQL / Planner
• Labels: pull-request-available
• Assignee: Unassigned
• Reporter: Zhanchun Zhang
• Created: 01/Apr/20 10:17
• Updated: 10/Dec/20 14:35
• Resolved: 10/Dec/20 14:35

Description

  def collect[T](
      tEnv: TableEnvironment,
      table: Table,
      sink: CollectTableSink[T],
      jobName: Option[String]): Seq[T] = {   // jobName was not used
    val typeSerializer = fromDataTypeToLegacyInfo(sink.getConsumedDataType)
      .asInstanceOf[TypeInformation[T]]
      .createSerializer(tEnv.asInstanceOf[TableEnvironmentImpl]
        .getPlanner.asInstanceOf[PlannerBase].getExecEnv.getConfig)
    val id = new AbstractID().toString
    sink.init(typeSerializer.asInstanceOf[TypeSerializer[T]], id)
    val sinkName = UUID.randomUUID().toString
    tEnv.registerTableSink(sinkName, sink)
    tEnv.insertInto(s"`$sinkName`", table)

    val res = tEnv.execute(jobName.getOrElse("test"))
    val accResult: JArrayList[Array[Byte]] = res.getAccumulatorResult(id)
    SerializedListAccumulator.deserializeList(accResult, typeSerializer)
  }

Just shown as the code, parameter 'jobName' not used.

Issue Links

• links to: GitHub Pull Request #11599