Addressed few code refactoring nits.

Author: ScrapCodes

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

@@ -39,13 +39,8 @@ private[kafka010] case class CachedKafkaProducer(
 
   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))
+    val producer = new KafkaProducer[Array[Byte], Array[Byte]](configMap)
     logDebug(s"Created a new instance of KafkaProducer for " +
       s"$kafkaParams with Id: $id")
     closed = false
@@ -59,7 +54,7 @@ private[kafka010] case class CachedKafkaProducer(
         if (!closed) {
           closed = true
           kafkaProducer.close()
-          logInfo(s"Closed kafka producer: $this")
+          logDebug(s"Closed kafka producer: $this")
         }
       }
     } catch {
@@ -94,6 +89,11 @@ private[kafka010] object CachedKafkaProducer extends Logging {
     }
   }
 
+  private def updatedAuthConfigIfNeeded(kafkaParamsMap: ju.Map[String, Object]) =
+    KafkaConfigUpdater("executor", kafkaParamsMap.asScala.toMap)
+      .setAuthenticationConfigIfNeeded()
+      .build()
+
   private val closeQueue = new ConcurrentLinkedQueue[CachedKafkaProducer]()
 
   private val removalListener = new RemovalListener[Seq[(String, Object)], CachedKafkaProducer]() {
@@ -122,7 +122,7 @@ private[kafka010] object CachedKafkaProducer extends Logging {
    * one instance per specified kafkaParams.
    */
   private[kafka010] def acquire(kafkaParamsMap: ju.Map[String, Object]): CachedKafkaProducer = {
-    val paramsSeq: Seq[(String, Object)] = paramsToSeq(kafkaParamsMap)
+    val paramsSeq: Seq[(String, Object)] = paramsToSeq(updatedAuthConfigIfNeeded(kafkaParamsMap))
     try {
       val producer = this.synchronized {
         val cachedKafkaProducer: CachedKafkaProducer = guavaCache.get(paramsSeq)
@@ -148,7 +148,6 @@ private[kafka010] object CachedKafkaProducer extends Logging {
     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 {
@@ -165,11 +164,11 @@ private[kafka010] object CachedKafkaProducer extends Logging {
       }
     }
     // We need a close queue, so that we can close the producer(s) outside of a synchronized block.
-    processPendingClose(producer)
+    processPendingClose()
   }
 
   /** Process pending closes. */
-  private def processPendingClose(producer: CachedKafkaProducer): Unit = {
+  private def processPendingClose(): Unit = {
     // Check and close any other producers previously evicted, but pending to be closed.
     for (p <- closeQueue.iterator().asScala) {
       if (!p.inUse()) {

external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/CachedKafkaProducerSuite.scala

@@ -40,7 +40,7 @@ class CachedKafkaProducerSuite extends SharedSQLContext with KafkaTest {
   }
 
   test("Should return the cached instance on calling acquire with same params.") {
-    val kafkaParams: ju.HashMap[String, Object] = generateKafkaParams
+    val kafkaParams = generateKafkaParams
     val producer = CachedKafkaProducer.acquire(kafkaParams)
     val producer2 = CachedKafkaProducer.acquire(kafkaParams)
     assert(producer.kafkaProducer == producer2.kafkaProducer)
@@ -50,7 +50,7 @@ class CachedKafkaProducerSuite extends SharedSQLContext with KafkaTest {
   }
 
   test("Should return the new instance on calling acquire with different params.") {
-    val kafkaParams: ju.HashMap[String, Object] = generateKafkaParams
+    val kafkaParams = generateKafkaParams
     val producer = CachedKafkaProducer.acquire(kafkaParams)
     kafkaParams.remove("acks") // mutate the kafka params.
     val producer2 = CachedKafkaProducer.acquire(kafkaParams)
@@ -89,7 +89,7 @@ class CachedKafkaProducerSuite extends SharedSQLContext with KafkaTest {
   }
 
   test("Should not close a producer in-use.") {
-    val kafkaParams: ju.HashMap[String, Object] = generateKafkaParams
+    val kafkaParams = generateKafkaParams
     val producer: CachedKafkaProducer = CachedKafkaProducer.acquire(kafkaParams)
     producer.kafkaProducer // initializing the producer.
     assert(producer.getInUseCount == 1)
@@ -113,14 +113,6 @@ class CachedKafkaProducerStressSuite extends KafkaContinuousTest with KafkaTest
 
   override val brokerProps = Map("auto.create.topics.enable" -> "false")
 
-  override def afterAll(): Unit = {
-    if (testUtils != null) {
-      testUtils.teardown()
-      testUtils = null
-    }
-    super.afterAll()
-  }
-
   override def sparkConf: SparkConf = {
     val conf = super.sparkConf
     conf.set("spark.kafka.producer.cache.timeout", "2ms")