Support for managed identity based authentication in python kafka client

[Kruthika]

We followed this document https://docs.databricks.com/aws/en/connect/streaming/kafka?language=Python#msk-aad to use Kafka clie...

As part of the SFI, the guidance is to move away from client secret and use managed identity instead. As per our investigation so far, we have not find a way to do the same. Can you please guide us how we can do the same in this case?

What we have tried so far?

1. To create token and pass directly - did not work
   kafka_options = {
       "kafka.bootstrap.servers": brokers,
       "subscribe": topic,
       "kafka.security.protocol": "SASL_SSL",
       "kafka.sasl.mechanism": "OAUTHBEARER",
       "kafka.sasl.jaas.config": f'org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required oauth.token="{oauth_token}";',
       "kafka.sasl.login.callback.handler.class": "org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler"
   }

   What are trying currently?

   1. To use custom handler instead of the default handler. We are stuck here because right libraries are not found

   Our current code with secret:

   sasl_config = f'kafkashaded.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 = {

     "kafka.bootstrap.servers"  : f"{event_hubs_server}:9093",

     "subscribe"                : event_hubs_topic,

     "kafka.sasl.mechanism"     : "OAUTHBEARER",

     "kafka.security.protocol"  : "SASL_SSL",

     "kafka.sasl.jaas.config"   : sasl_config,

     "kafka.sasl.oauthbearer.token.endpoint.url": f"<url>",

     "kafka.sasl.login.callback.handler.class": "kafkashaded.org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler",

     "kafka.request.timeout.ms" : "60000",

     "kafka.session.timeout.ms" : "60000",

     "maxOffsetsPerTrigger"     : "1000",

     "failOnDataLoss"           : "false",

     "startingOffsets"          : "earliest"

   }