[SPARK-21869][SS] A cached Kafka producer should not be closed if any task is using it - adds inuse tracking.

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/CachedKafkaProducer.scala

@@ -18,20 +18,70 @@
 package org.apache.spark.sql.kafka010
 
 import java.{util => ju}
-import java.util.concurrent.{ConcurrentMap, ExecutionException, TimeUnit}
+import java.util.concurrent.{ConcurrentLinkedQueue, ConcurrentMap, ExecutionException, TimeUnit}
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.collection.JavaConverters._
+import scala.util.control.NonFatal
 
 import com.google.common.cache._
 import com.google.common.util.concurrent.{ExecutionError, UncheckedExecutionException}
 import org.apache.kafka.clients.producer.KafkaProducer
-import scala.collection.JavaConverters._
-import scala.util.control.NonFatal
 
 import org.apache.spark.SparkEnv
 import org.apache.spark.internal.Logging
 
-private[kafka010] object CachedKafkaProducer extends Logging {
+private[kafka010] case class CachedKafkaProducer(
+    private val id: String = ju.UUID.randomUUID().toString,
+    private val inUseCount: AtomicInteger = new AtomicInteger(0),
+    private val kafkaParams: Seq[(String, Object)]) extends Logging {
+
+  private val configMap = kafkaParams.map(x => x._1 -> x._2).toMap.asJava
+
+  private def updatedAuthConfigIfNeeded(kafkaParamsMap: ju.Map[String, Object]) =
+    KafkaConfigUpdater("executor", kafkaParamsMap.asScala.toMap)
+      .setAuthenticationConfigIfNeeded()
+      .build()
+
+  lazy val kafkaProducer: KafkaProducer[Array[Byte], Array[Byte]] = {
+    val producer = new KafkaProducer[Array[Byte], Array[Byte]](updatedAuthConfigIfNeeded(configMap))
+    logDebug(s"Created a new instance of KafkaProducer for " +
+      s"$kafkaParams with Id: $id")
+    closed = false
+    producer
+  }
+  @volatile
+  private var isCached: Boolean = true
+  private var closed: Boolean = true
+  private def close(): Unit = {
+    try {
+      this.synchronized {
+        if (!closed) {
+          closed = true
+          kafkaProducer.close()
+          logInfo(s"Closed kafka producer: $this")
+        }
+      }
+    } catch {
+      case NonFatal(e) =>
+        logWarning(s"Error while closing kafka producer with params: $kafkaParams", e)
+    }
+  }
+
+  private def inUse(): Boolean = inUseCount.get() > 0
+
+  private def unCache(): Unit = isCached = false
+
+  private[kafka010] def getInUseCount: Int = inUseCount.get()
 
-  private type Producer = KafkaProducer[Array[Byte], Array[Byte]]
+  private[kafka010] def getKafkaParams: Seq[(String, Object)] = kafkaParams
+
+  private[kafka010] def flush(): Unit = kafkaProducer.flush()
+
+  private[kafka010] def isClosed: Boolean = closed
+}
+
+private[kafka010] object CachedKafkaProducer extends Logging {
 
   private val defaultCacheExpireTimeout = TimeUnit.MINUTES.toMillis(10)
 
@@ -40,81 +90,114 @@ private[kafka010] object CachedKafkaProducer extends Logging {
       "spark.kafka.producer.cache.timeout",
       s"${defaultCacheExpireTimeout}ms")).getOrElse(defaultCacheExpireTimeout)
 
-  private val cacheLoader = new CacheLoader[Seq[(String, Object)], Producer] {
-    override def load(config: Seq[(String, Object)]): Producer = {
-      val configMap = config.map(x => x._1 -> x._2).toMap.asJava
-      createKafkaProducer(configMap)
+  private val cacheLoader = new CacheLoader[Seq[(String, Object)], CachedKafkaProducer] {
+    override def load(params: Seq[(String, Object)]): CachedKafkaProducer = {
+      CachedKafkaProducer(kafkaParams = params)
     }
   }
 
-  private val removalListener = new RemovalListener[Seq[(String, Object)], Producer]() {
+  private val closeQueue = new ConcurrentLinkedQueue[CachedKafkaProducer]()
+
+  private val removalListener = new RemovalListener[Seq[(String, Object)], CachedKafkaProducer]() {
     override def onRemoval(
-        notification: RemovalNotification[Seq[(String, Object)], Producer]): Unit = {
-      val paramsSeq: Seq[(String, Object)] = notification.getKey
-      val producer: Producer = notification.getValue
-      logDebug(
-        s"Evicting kafka producer $producer params: $paramsSeq, due to ${notification.getCause}")
-      close(paramsSeq, producer)
+        notification: RemovalNotification[Seq[(String, Object)], CachedKafkaProducer]): Unit = {
+      val producer: CachedKafkaProducer = notification.getValue
+      if (producer.inUse()) {
+        logDebug(s"Evicting kafka producer $producer, due to ${notification.getCause}.")
+        // When `inuse` producer is evicted we wait for it to be released before finally closing it.
+        closeQueue.add(producer)
+        producer.unCache()
+      } else {
+        close(producer)
+      }
     }
   }
 
-  private lazy val guavaCache: LoadingCache[Seq[(String, Object)], Producer] =
+  private lazy val guavaCache: LoadingCache[Seq[(String, Object)], CachedKafkaProducer] =
     CacheBuilder.newBuilder().expireAfterAccess(cacheExpireTimeout, TimeUnit.MILLISECONDS)
       .removalListener(removalListener)
-      .build[Seq[(String, Object)], Producer](cacheLoader)
-
-  private def createKafkaProducer(producerConfiguration: ju.Map[String, Object]): Producer = {
-    val updatedKafkaProducerConfiguration =
-      KafkaConfigUpdater("executor", producerConfiguration.asScala.toMap)
-        .setAuthenticationConfigIfNeeded()
-        .build()
-    val kafkaProducer: Producer = new Producer(updatedKafkaProducerConfiguration)
-    logDebug(s"Created a new instance of KafkaProducer for $updatedKafkaProducerConfiguration.")
-    kafkaProducer
-  }
+      .build[Seq[(String, Object)], CachedKafkaProducer](cacheLoader)
 
   /**
    * Get a cached KafkaProducer for a given configuration. If matching KafkaProducer doesn't
    * exist, a new KafkaProducer will be created. KafkaProducer is thread safe, it is best to keep
    * one instance per specified kafkaParams.
    */
-  private[kafka010] def getOrCreate(kafkaParams: ju.Map[String, Object]): Producer = {
-    val paramsSeq: Seq[(String, Object)] = paramsToSeq(kafkaParams)
+  private[kafka010] def acquire(kafkaParamsMap: ju.Map[String, Object]): CachedKafkaProducer = {
+    val paramsSeq: Seq[(String, Object)] = paramsToSeq(kafkaParamsMap)
     try {
-      guavaCache.get(paramsSeq)
+      val producer = this.synchronized {
+        val cachedKafkaProducer: CachedKafkaProducer = guavaCache.get(paramsSeq)
+        cachedKafkaProducer.inUseCount.incrementAndGet()
+        logDebug(s"Granted producer $cachedKafkaProducer")
+        cachedKafkaProducer
+      }
+      producer
     } catch {
-      case e @ (_: ExecutionException | _: UncheckedExecutionException | _: ExecutionError)
+      case e@(_: ExecutionException | _: UncheckedExecutionException | _: ExecutionError)
         if e.getCause != null =>
         throw e.getCause
     }
   }
 
-  private def paramsToSeq(kafkaParams: ju.Map[String, Object]): Seq[(String, Object)] = {
-    val paramsSeq: Seq[(String, Object)] = kafkaParams.asScala.toSeq.sortBy(x => x._1)
+  private def paramsToSeq(kafkaParamsMap: ju.Map[String, Object]): Seq[(String, Object)] = {
+    val paramsSeq: Seq[(String, Object)] = kafkaParamsMap.asScala.toSeq.sortBy(x => x._1)
     paramsSeq
   }
 
-  /** For explicitly closing kafka producer */
-  private[kafka010] def close(kafkaParams: ju.Map[String, Object]): Unit = {
-    val paramsSeq = paramsToSeq(kafkaParams)
-    guavaCache.invalidate(paramsSeq)
+  /* Release a kafka producer back to the kafka cache. We simply decrement it's inuse count. */
+  private[kafka010] def release(producer: CachedKafkaProducer, failing: Boolean): Unit = {
+    this.synchronized {
+      // It should be ok to call release multiple times on the same producer object.
+      if (producer.inUse()) {
+        // So that we do not end up with -ve in-use counts.
+        producer.inUseCount.decrementAndGet()
+        logDebug(s"Released producer $producer.")
+      } else {
+        logWarning(s"Tried to release a not in use producer, $producer.")
+      }
+      if (failing) {
+        // If this producer is failing to write, we remove it from cache.
+        // So that it is re-created, eventually.
+        val cachedProducer = guavaCache.getIfPresent(producer.kafkaParams)
+        if (cachedProducer != null && cachedProducer.id == producer.id) {
+          logDebug(s"Invalidating a failing producer: $producer.")
+          guavaCache.invalidate(producer.kafkaParams)
+        }
+      }
+    }
+    if (!producer.inUse() && !producer.isCached) {
+      // it will take care of removing it from close queue as well.
+      close(producer)
+    }
   }
 
-  /** Auto close on cache evict */
-  private def close(paramsSeq: Seq[(String, Object)], producer: Producer): Unit = {
-    try {
-      logInfo(s"Closing the KafkaProducer with params: ${paramsSeq.mkString("\n")}.")
-      producer.close()
-    } catch {
-      case NonFatal(e) => logWarning("Error while closing kafka producer.", e)
+  /** Close this producer and process pending closes. */
+  private def close(producer: CachedKafkaProducer): Unit = {
+    producer.close()
+    // Check and close any other producers previously evicted, but pending to be closed.
+    for (p <- closeQueue.iterator().asScala) {
+      if (!p.inUse()) {
+        closeQueue.remove(p)
+        p.close()
+      }
     }
   }
 
+  // Intended for testing purpose only.
   private[kafka010] def clear(): Unit = {
-    logInfo("Cleaning up guava cache.")
+    logInfo("Cleaning up guava cache and force closing all kafka producers.")
     guavaCache.invalidateAll()
+    for (p <- closeQueue.iterator().asScala) {
+      p.close()
+    }
+    closeQueue.clear()
   }
 
-  // Intended for testing purpose only.
-  private def getAsMap: ConcurrentMap[Seq[(String, Object)], Producer] = guavaCache.asMap()
+  private[kafka010] def evict(params: Seq[(String, Object)]): Unit = {
+    guavaCache.invalidate(params)
+  }
+
+  private[kafka010] def getAsMap: ConcurrentMap[Seq[(String, Object)], CachedKafkaProducer] =
+    guavaCache.asMap()
 }

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaStreamingWrite.scala

@@ -92,7 +92,7 @@ class KafkaStreamDataWriter(
     inputSchema: Seq[Attribute])
   extends KafkaRowWriter(inputSchema, targetTopic) with DataWriter[InternalRow] {
 
-  private lazy val producer = CachedKafkaProducer.getOrCreate(producerParams)
+  protected val producer: CachedKafkaProducer = CachedKafkaProducer.acquire(producerParams)
 
   def write(row: InternalRow): Unit = {
     checkForErrors()
@@ -103,20 +103,21 @@ class KafkaStreamDataWriter(
     // Send is asynchronous, but we can't commit until all rows are actually in Kafka.
     // This requires flushing and then checking that no callbacks produced errors.
     // We also check for errors before to fail as soon as possible - the check is cheap.
-    checkForErrors()
-    producer.flush()
-    checkForErrors()
+    close()
     KafkaWriterCommitMessage
   }
 
-  def abort(): Unit = {}
+  def abort(): Unit = {
+    close()
+  }
 
-  def close(): Unit = {
-    checkForErrors()
-    if (producer != null) {
+  private def close(): Unit = {
+    try {
+      checkForErrors()
       producer.flush()
       checkForErrors()
-      CachedKafkaProducer.close(producerParams)
+    } finally {
+      CachedKafkaProducer.release(producer, failedWrite != null)
     }
   }
 }

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaWriteTask.scala

@@ -19,7 +19,7 @@ package org.apache.spark.sql.kafka010
 
 import java.{util => ju}
 
-import org.apache.kafka.clients.producer.{Callback, KafkaProducer, ProducerRecord, RecordMetadata}
+import org.apache.kafka.clients.producer.{Callback, ProducerRecord, RecordMetadata}
 
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Cast, Literal, UnsafeProjection}
@@ -35,32 +35,35 @@ private[kafka010] class KafkaWriteTask(
     inputSchema: Seq[Attribute],
     topic: Option[String]) extends KafkaRowWriter(inputSchema, topic) {
   // used to synchronize with Kafka callbacks
-  private var producer: KafkaProducer[Array[Byte], Array[Byte]] = _
+  protected val producer: CachedKafkaProducer =
+    CachedKafkaProducer.acquire(producerConfiguration)
 
   /**
    * Writes key value data out to topics.
    */
   def execute(iterator: Iterator[InternalRow]): Unit = {
-    producer = CachedKafkaProducer.getOrCreate(producerConfiguration)
     while (iterator.hasNext && failedWrite == null) {
       val currentRow = iterator.next()
       sendRow(currentRow, producer)
     }
   }
 
   def close(): Unit = {
-    checkForErrors()
-    if (producer != null) {
+    try {
+      checkForErrors()
       producer.flush()
       checkForErrors()
-      producer = null
+    } finally {
+      CachedKafkaProducer.release(producer, failedWrite != null)
     }
   }
+
 }
 
 private[kafka010] abstract class KafkaRowWriter(
     inputSchema: Seq[Attribute], topic: Option[String]) {
 
+  protected val producer: CachedKafkaProducer
   // used to synchronize with Kafka callbacks
   @volatile protected var failedWrite: Exception = _
   protected val projection = createProjection
@@ -79,7 +82,7 @@ private[kafka010] abstract class KafkaRowWriter(
    * assuming the row is in Kafka.
    */
   protected def sendRow(
-      row: InternalRow, producer: KafkaProducer[Array[Byte], Array[Byte]]): Unit = {
+      row: InternalRow, producer: CachedKafkaProducer): Unit = {
     val projectedRow = projection(row)
     val topic = projectedRow.getUTF8String(0)
     val key = projectedRow.getBinary(1)
@@ -89,7 +92,7 @@ private[kafka010] abstract class KafkaRowWriter(
         s"${KafkaSourceProvider.TOPIC_OPTION_KEY} option for setting a default topic.")
     }
     val record = new ProducerRecord[Array[Byte], Array[Byte]](topic.toString, key, value)
-    producer.send(record, callback)
+    producer.kafkaProducer.send(record, callback)
   }
 
   protected def checkForErrors(): Unit = {