gaborgsomogyi opened a new pull request #23956: [SPARK-27042][SS] Close cached Kafka consumer in case of task retry URL: https://github.com/apache/spark/pull/23956 ## What changes were proposed in this pull request? If a task is failing due to a cached corrupted `KafkaProducer` and the task is retried in the same executor then the task is getting the same `KafkaProducer` over and over again unless it's invalidated with the timeout configured by `spark.kafka.producer.cache.timeout` which is not really probable. After several retries the query stops. In this PR I'm closing the old cached `KafkaProducer` and reopen a new one. The functionality is similar to the `KafkaConsumer` side [here](https://github.com/apache/spark/blob/382d5a82b0e8ef1dd01209e828eae67fe7993a56/external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaDataConsumer.scala#L576). ## How was this patch tested? Additional unit tests + on cluster.
---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: [email protected] With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
