Unable to connect to Confluent from Databricks

[Sascha]

I'm facing the same issue as this post: https://community.databricks.com/s/question/0D58Y00009DE82zSAD/databricks-kafka-read-not-connecting

In my case I'm connecting to Confluent Cloud. I'm able to ping the bootstrap server, I'm able to netstat succesfully on the port 9092. But when I try to consume the data using a batch consumer (or stream, that doesn't matter), the log4j server logs get flooded with the message:

22/10/04 07:39:18 WARN NetworkClient: [Consumer clientId=Databricks, groupId=new_group2] Bootstrap broker pkc-75m1o.europe-west3.gcp.confluent.cloud:9092 (id: -1 rack: null) disconnected

userid = "<API Key>"

password = "<API Secret>"

host = "pkc-75m1o.europe-west3.gcp.confluent.cloud:9092"

topic = "topic_0"

sasl_mech = "PLAIN"

inputDF = spark \

 .read \

 .format("kafka") \

 .option("kafka.bootstrap.servers", host) \

 .option("ssl.endpoint.identification.algorithm", "https") \

 .option("sasl.mechanism", sasl_mech) \

 .option("security.protocol", "SASL_SSL") \

 .option("sasl.jaas.config", "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") \

 .option("retry.backoff.ms", "1000") \

 .option("value.deserializer", "ByteArrayDeserializer") \

 .load()

display(inputDF)

[Debayan]

Hi @Sascha Zevenhuizen​ , have you enabled kerberos security on the settings by any chance?

[Sascha · Accepted solution]

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() )

[Sascha]

Hi @Kaniz Fatma​ , that was the thread I started my trouble-shooting with, and it's surely helpful!

[Sascha]

Yeah ok!