Synapse Spark Pool(spark 3.1): unable to find class: org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler

Estephania Palomino Virgen 5 Microsoft Employee

I am trying to test Kafka Consumer to read events from Azure EventHub and load it as DataStream in python using Azure Synapse Workspace. I need to do the auth using service principal and secret, please advise me on what I might be doing wrong or if I am missing something here.

sasl_config = f'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId="{client_id}" clientSecret="{client_secret}" scope="https://{event_hubs_server}/.default" ssl.protocol="SSL";'

kafka_options = {
    # Port 9093 is the EventHubs Kafka port
    "kafka.bootstrap.servers": f"{event_hubs_server}:9093",
    "kafka.sasl.jaas.config": sasl_config,
    "kafka.sasl.oauthbearer.token.endpoint.url": f"https://login.microsoft.com/{tenant_id}/oauth2/v2.0/token",
    "subscribe": event_hubs_topic,
    "kafka.security.protocol": "SASL_SSL",
    "kafka.sasl.mechanism": "OAUTHBEARER",
    "startingOffsets":"earliest",
    "kafka.sasl.login.callback.handler.class": "org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler" }  

df = spark.readStream.format("kafka").options(**kafka_options).load()

Error message: Class org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler could not be found.

BhargavaGunnam-MSFT 26,306 Reputation points Microsoft Employee

2024-04-05T17:16:08.71+00:00
Hello Estephania Palomino Virgen,

Since you are using service principal, can you please check your environment variables are correctly configured?

AZURE_TENANT_ID

AZURE_CLIENT_ID

AZURE_CLIENT_SECRET

https://github.com/Azure/azure-event-hubs-for-kafka/tree/master/tutorials/oauth/python

and per the error message, Class org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler could not be found.

Can you please verify that your class is available in your runtime environment?