Fix

Author: gaborgsomogyi

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

@@ -66,7 +66,7 @@ private[kafka010] object CachedKafkaProducer extends Logging {
       .build[Seq[(String, Object)], Producer](cacheLoader)
 
   private def createKafkaProducer(paramsSeq: Seq[(String, Object)]): Producer = {
-    val kafkaProducer: Producer = new Producer(paramsSeq.map(x => x._1 -> x._2).toMap.asJava)
+    val kafkaProducer: Producer = new Producer(paramsSeq.toMap.asJava)
     if (log.isDebugEnabled()) {
       val redactedParamsSeq = KafkaRedactionUtil.redactParams(paramsSeq)
       logDebug(s"Created a new instance of KafkaProducer for $redactedParamsSeq.")

external/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaRedactionUtil.scala

@@ -26,14 +26,19 @@ import org.apache.spark.util.Utils.{redact, REDACTION_REPLACEMENT_TEXT}
 
 private[spark] object KafkaRedactionUtil extends Logging {
   private[spark] def redactParams(params: Seq[(String, Object)]): Seq[(String, Object)] = {
-    val redactionPattern = SparkEnv.get.conf.get(SECRET_REDACTION_PATTERN)
+    val redactionPattern = Some(SparkEnv.get.conf.get(SECRET_REDACTION_PATTERN))
     params.map { case (key, value) =>
       if (key.equalsIgnoreCase(SaslConfigs.SASL_JAAS_CONFIG)) {
         (key, redactJaasParam(value.asInstanceOf[String]))
       } else {
-        val (_, newValue) = redact(Some(redactionPattern), Seq((key, value.asInstanceOf[String])))
-          .head
-        (key, newValue)
+        value match {
+          case s: String =>
+            val (_, newValue) = redact(redactionPattern, Seq((key, s))).head
+            (key, newValue)
+
+          case _ =>
+            (key, value)
+        }
       }
     }
   }

external/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaRedactionUtilSuite.scala

@@ -22,6 +22,7 @@ import java.{util => ju}
 import org.apache.kafka.clients.consumer.ConsumerConfig
 import org.apache.kafka.common.config.{SaslConfigs, SslConfigs}
 import org.apache.kafka.common.security.auth.SecurityProtocol.SASL_SSL
+import org.apache.kafka.common.serialization.StringDeserializer
 
 import org.apache.spark.SparkFunSuite
 import org.apache.spark.util.Utils.REDACTION_REPLACEMENT_TEXT
@@ -32,6 +33,15 @@ class KafkaRedactionUtilSuite extends SparkFunSuite with KafkaDelegationTokenTes
     assert(KafkaRedactionUtil.redactParams(Seq()) === Seq())
   }
 
+  test("redactParams should give back non String parameters") {
+    setSparkEnv(Map())
+    val kafkaParams = Seq(
+      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer]
+    )
+
+    assert(KafkaRedactionUtil.redactParams(kafkaParams) === kafkaParams)
+  }
+
   test("redactParams should redact token password from parameters") {
     setSparkEnv(Map())
     val groupId = "id-" + ju.UUID.randomUUID().toString