Hi @Debayan Mukherjee​ , no I haven't.

But with the help of Confluent I changed the statement to the below, and somehow this solved it.

inputDF = (spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", host)
  .option("kafka.ssl.endpoint.identification.algorithm", "https")
  .option("kafka.sasl.mechanism", "PLAIN")
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.sasl.jaas.config", "kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username='{}' password='{}';".format(userid, password))
  .option("subscribe", topic)
  .option("kafka.client.id", "Databricks")
  .option("kafka.group.id", "new_group2")
  .option("spark.streaming.kafka.maxRatePerPartition", "5")
  .option("startingOffsets", "earliest")
  .option("kafka.session.timeout.ms", "10000")
  .load() )