Using spark conf instead of System.getProperties.

Author: ScrapCodes

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

@@ -28,15 +28,16 @@ import scala.collection.immutable.SortedMap
 import scala.collection.mutable
 import scala.util.control.NonFatal
 
+import org.apache.spark.SparkEnv
 import org.apache.spark.internal.Logging
 import org.apache.spark.util.ShutdownHookManager
 
 private[kafka010] object CachedKafkaProducer extends Logging {
 
   private type Producer = KafkaProducer[Array[Byte], Array[Byte]]
 
-  private val cacheExpireTimeout: Long =
-    System.getProperty("spark.kafka.guava.cache.timeout.minutes", "10").toLong
+  private lazy val cacheExpireTimeout: Long =
+    SparkEnv.get.conf.getTimeAsMs("spark.kafka.producer.cache.timeout", "10m")
 
   private val removalListener = new RemovalListener[String, Producer]() {
     override def onRemoval(notification: RemovalNotification[String, Producer]): Unit = {
@@ -47,9 +48,8 @@ private[kafka010] object CachedKafkaProducer extends Logging {
     }
   }
 
-  private val guavaCache: Cache[String, Producer] = CacheBuilder.newBuilder()
-    .recordStats()
-    .expireAfterAccess(cacheExpireTimeout, TimeUnit.MINUTES)
+  private lazy val guavaCache: Cache[String, Producer] = CacheBuilder.newBuilder()
+    .expireAfterAccess(cacheExpireTimeout, TimeUnit.MILLISECONDS)
     .removalListener(removalListener)
     .build[String, Producer]()
 

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

@@ -20,11 +20,10 @@ package org.apache.spark.sql.kafka010
 import java.{util => ju}
 
 import org.apache.kafka.common.serialization.ByteArraySerializer
-import org.scalatest.PrivateMethodTester
 
 import org.apache.spark.sql.test.SharedSQLContext
 
-class CanonicalizeKafkaParamsSuite extends SharedSQLContext with PrivateMethodTester {
+class CanonicalizeKafkaParamsSuite extends SharedSQLContext {
 
   test("Same unique id is returned for same set of kafka parameters") {
     CanonicalizeKafkaParams.clear()