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: [email protected]
For additional commands, e-mail: [email protected]
