[SPARK-26360] Avoid extra validateQuery call in createStreamingWriteSupport

• Type: Improvement
• Status: Resolved
• Priority: Trivial
• Resolution: Fixed
• Affects Version/s: 2.3.2, 2.4.0
• Fix Version/s: 3.0.0
• Component/s: Structured Streaming
• Assignee: Wu Wenjie
• Reporter: Wu Wenjie
• Shepherd: Sean R. Owen
• Created: 13/Dec/18 13:53
• Updated: 12/Dec/22 18:10
• Resolved: 14/Dec/18 02:51

Description

When I'm reading structured streaming source code, I find there is a redundant KafkaWriter.validateQuery() function call in createStreamingWriteSupport func in class `KafkaSourceProvider`.

// KafkaSourceProvider.scala
  override def createStreamingWriteSupport(
      queryId: String,
      schema: StructType,
      mode: OutputMode,
      options: DataSourceOptions): StreamingWriteSupport = {
   .....
    // validate once here
    KafkaWriter.validateQuery(schema.toAttributes, producerParams, topic)

    // validate twice here
    new KafkaStreamingWriteSupport(topic, producerParams, schema)
  }

// KafkaStreamingWriteSupport.scala
class KafkaStreamingWriteSupport(
    topic: Option[String],
    producerParams: ju.Map[String, Object],
    schema: StructType)
  extends StreamingWriteSupport {

  validateQuery(schema.toAttributes, producerParams, topic)
  ....
}

 

I think we just need to remove one of these two.

Issue Links

• links to: GitHub Pull Request #23309