[GitHub] spark pull request #19096: [SPARK-21869][SS] A cached Kafka producer should ...

[zsxwing <gi...@git.apache.org>]

Github user zsxwing commented on a diff in the pull request:

    https://github.com/apache/spark/pull/19096#discussion_r137409030
  
    --- Diff: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaWriteTask.scala ---
    @@ -43,8 +43,10 @@ private[kafka010] class KafkaWriteTask(
        * Writes key value data out to topics.
        */
       def execute(iterator: Iterator[InternalRow]): Unit = {
    -    producer = CachedKafkaProducer.getOrCreate(producerConfiguration)
    +    val paramsSeq = CachedKafkaProducer.paramsToSeq(producerConfiguration)
         while (iterator.hasNext && failedWrite == null) {
    +      // Prevent producer to get expired/evicted from guava cache.(SPARK-21869)
    +      producer = CachedKafkaProducer.getOrCreate(paramsSeq)
    --- End diff --
    
    This is really hacky. I'm wondering if we can track if a producer is using or not.


---

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org