title: Consume Secure Kafka from databricks spark cluster
body: <p>I am trying to consume from a secure Kafka Topic ( using SASL_PLAINTEXT, ScramLogin Method). </p>
  
  <p>Spark Version 2.3.1
  Scala 2.11
  Kafka latest</p>
  
  <p>I am using the Spark Structured stream to construct the stream. For this purpose I imported the library : spark-sql-kafka-0-10_2.11-2.3.1</p>
  
  <p>This imports the older version (0.10.0.1) of the kafka-clients.jar.</p>
  
  <p>Below is my client.jaas : </p>
  
  <pre><code>KafkaClient {
   org.apache.kafka.common.security.scram.ScramLoginModule required
   mechanism=SCRAM-SHA-512
   security.protocol=SASL_PLAINTEXT
   client.id="*****"
   username="****"
   password="*****";
  };
  </code></pre>
  
  <p>I am using the ScramLoginModule and the kafka-clients jar specified above, does not even have this jar. So, I added the later version of the kafka Clients jar file kafka-clients-1.1.1-cp1.jar</p>
  
  <p>In my notebook, I have the following code : </p>
  
  <pre><code>import org.apache.spark.sql.functions._
  import org.apache.spark.sql.SparkSession
  import spark.implicits._
  
  val kafkaSource = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers",   "host:port")   
      .option("subscribe", "topic")    
    .option("startingOffsets", "earliest") 
    .option("maxOffsetsPerTrigger", 1)
  
    .option("kafka.sasl.mechanism","SCRAM-SHA-512")
    .option("kafka.security.protocol","SASL_PLAINTEXT")
    .load()
  
  
  val kafkaStream = kafkaSource.
    select(
      $"key" cast "string",   // deserialize keys
      $"value" cast "string", // deserialize values
      $"topic",
      $"partition",
      $"offset",
      $"timestamp",
      $"timestampType")
  
  import org.apache.spark.sql.ForeachWriter
  
  kafkaStream.writeStream.foreach(new ForeachWriter[Row] {
  
    override def process(row: Row): Unit = {
      println("Processing : " + row.mkString(","))
    }
  
    override def close(errorOrNull: Throwable): Unit = {}
  
    override def open(partitionId: Long, version: Long): Boolean = {
      true
    }
  }).start().awaitTermination()
  </code></pre>
  
  <p>From the driver logs, I can see that the I am able to connect fine . </p>
  
  <pre><code>18/09/05 13:09:46 INFO AbstractLogin: Successfully logged in.
  18/09/05 13:09:46 INFO ConsumerConfig: ConsumerConfig values: 
  </code></pre>
  
  <p>But after this, nothing happens !! It just gets stuck after printing hte following lines : </p>
  
  <pre><code>18/09/05 13:09:47 INFO AppInfoParser: Kafka version : 0.10.0.1
  18/09/05 13:09:47 INFO AppInfoParser: Kafka commitId : a7a17cdec9eaa6c5
  </code></pre>
  
  <p>The above version of Kafka is also very confusing as I am nowhere using this version (Although this was the version of the kafka client that gets imported automatically when I add the spark-sql-kafka-0-10_2.11-2.3.1 library. But I had chosen to exclude the kafka-clients jar when I did that import).  The kafka clients that I am using as I said is of a higher version and so is the kafka cluster to which i am connecting. </p>
  
  <p>To replicate this, I ran the same code on a spark cluster running on my own laptop, and everything works fine, where I am able to consume and print the kafka messages. </p>
  
  <p>Anyone who has experienced similar issue, please advise !!</p>