title: Spark Kafka connection failed : No resolvable bootstrap urls given in bootstrap.servers
body: <p>I try to read a kafka topic with <strong>Spark 3.0.2</strong>, I do a <strong>spark shell</strong> with the following libs:</p>
  <ul>
  <li>spark-sql-kafka-0-10_2.12-3.0.2.jar</li>
  <li>kafka-avro-serializer-6.2.0.jar</li>
  <li>kafka-clients-2.4.1.jar</li>
  <li>spark-streaming-kafka-0-10-assembly_2.12-3.0.2.jar</li>
  <li>spark-tags_2.12-3.0.2.jar</li>
  <li>spark-token-provider-kafka-0-10_2.12-3.0.2.jar</li>
  <li>commons-pool2-2.6.2.jar</li>
  </ul>
  <p>I am getting the following output with the error stacktrace:</p>
  <pre><code>-------------------------------------------
  Batch: 0
  -------------------------------------------
  +-----+----------+----------+-------+--------+--------+
  |COL1|CUSTOMSREF|MPSIDCCKEY|MPSCOMP|MPSCREF1|MPSCREF2|
  +-----+----------+----------+-------+--------+--------+
  +-----+----------+----------+-------+--------+--------+
  
      21/07/21 10:14:30 WARN TaskSetManager: Lost task 4.0 in stage 4.0 (TID 20, 172.20.0.4, executor 1): org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
              at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:820)
              at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:631)
              at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:612)
              at org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.createConsumer(KafkaDataConsumer.scala:122)
              at org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumer.&lt;init&gt;(KafkaDataConsumer.scala:59)
              at org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumerPool$ObjectFactory.create(InternalKafkaConsumerPool.scala:206)
              at org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumerPool$ObjectFactory.create(InternalKafkaConsumerPool.scala:201)
              at org.apache.commons.pool2.BaseKeyedPooledObjectFactory.makeObject(BaseKeyedPooledObjectFactory.java:60)
              at org.apache.commons.pool2.impl.GenericKeyedObjectPool.create(GenericKeyedObjectPool.java:1041)
              at org.apache.commons.pool2.impl.GenericKeyedObjectPool.borrowObject(GenericKeyedObjectPool.java:342)
              at org.apache.commons.pool2.impl.GenericKeyedObjectPool.borrowObject(GenericKeyedObjectPool.java:265)
              at org.apache.spark.sql.kafka010.consumer.InternalKafkaConsumerPool.borrowObject(InternalKafkaConsumerPool.scala:84)
              at org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.retrieveConsumer(KafkaDataConsumer.scala:554)
              at org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.getOrRetrieveConsumer(KafkaDataConsumer.scala:539)
              at org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.$anonfun$get$1(KafkaDataConsumer.scala:285)
              at org.apache.spark.util.UninterruptibleThread.runUninterruptibly(UninterruptibleThread.scala:77)
              at org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.runUninterruptiblyIfPossible(KafkaDataConsumer.scala:598)
              at org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.get(KafkaDataConsumer.scala:281)
              at org.apache.spark.sql.kafka010.KafkaBatchPartitionReader.next(KafkaBatchPartitionReader.scala:63)
              at org.apache.spark.sql.execution.datasources.v2.PartitionIterator.hasNext(DataSourceRDD.scala:79)
              at org.apache.spark.sql.execution.datasources.v2.MetricsIterator.hasNext(DataSourceRDD.scala:112)
              at org.apache.spark.InterruptibleIterator.hasNext(InterruptibleIterator.scala:37)
              at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
              at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)
              at org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43)
              at org.apache.spark.sql.execution.WholeStageCodegenExec$$anon$1.hasNext(WholeStageCodegenExec.scala:729)
              at scala.collection.Iterator$$anon$10.hasNext(Iterator.scala:458)
              at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.$anonfun$run$7(WriteToDataSourceV2Exec.scala:438)
              at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1411)
              at org.apache.spark.sql.execution.datasources.v2.DataWritingSparkTask$.run(WriteToDataSourceV2Exec.scala:477)
              at org.apache.spark.sql.execution.datasources.v2.V2TableWriteExec.$anonfun$writeWithV2$2(WriteToDataSourceV2Exec.scala:385)
              at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
              at org.apache.spark.scheduler.Task.run(Task.scala:127)
              at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:462)
              at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1377)
              at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:465)
              at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
              at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
              at java.lang.Thread.run(Thread.java:748)
      Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers
              at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:88)
              at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:47)
              at org.apache.kafka.clients.consumer.KafkaConsumer.&lt;init&gt;(KafkaConsumer.java:735)
              ... 38 more
  </code></pre>
  <p>Can anyone knows how to resolve it?</p>
  <p>Here is spark code in scala :</p>
  <pre><code>import io.confluent.kafka.serializers.KafkaAvroDeserializer
  import org.apache.avro.Schema
  import org.apache.avro.generic.GenericRecord
  import org.apache.kafka.common.serialization.Deserializer
  import java.nio.file.Files
  import java.nio.file.Paths
  import org.apache.spark.sql.avro.functions._
  import org.apache.spark.sql.streaming.Trigger
  
  val schema = new String(Files.readAllBytes(Paths.get(&quot;/path/to/avro/schema.avsc&quot;)))
  val topic = &quot;topic_A&quot;
  
  val streamDf = spark
    .readStream.format(&quot;kafka&quot;)
    .option(&quot;kafka.bootstrap.servers&quot;, &quot;kafka1:9192&quot;)
    .option(&quot;subscribe&quot;, topic)
    .option(&quot;kafka.group.id&quot;, &quot;id_A&quot;)
    .option(&quot;kafka.security.protocol&quot;, &quot;SASL_SSL&quot;)
    .option(&quot;kafka.sasl.mechanism&quot;, &quot;PLAIN&quot;)
    .option(&quot;kafka.ssl.truststore.location&quot;, &quot;/path/to/truststore/certificate.jks&quot;)
    .option(&quot;kafka.ssl.truststore.password&quot;, &quot;password&quot;)
    .option(&quot;kafka.sasl.jaas.config&quot;, &quot;&quot;&quot;org.apache.kafka.common.security.plain.PlainLoginModule required username=&quot;user_A&quot; password=&quot;password&quot;;&quot;&quot;&quot;)
    .option(&quot;startingOffsets&quot;, &quot;latest&quot;)
    .load()
  
  val dataDf = streamDf.selectExpr(&quot;CAST(key as STRING)&quot;, &quot;value&quot;)
    .select(from_avro(col(&quot;value&quot;),schema) as &quot;data&quot;)
    .select(&quot;data.*&quot;)
  
  dataDf
      .writeStream
      .format(&quot;console&quot;)
      .outputMode(&quot;append&quot;)
      .trigger(Trigger.ProcessingTime(&quot;3 seconds&quot;))
      .start
  </code></pre>
  <p>Thanks a lot.</p>