title: Kafka consumer using AWS_MSK_IAM ClassCastException error
body: <p>I have MSK running on AWS and I'd like to consume information using AWS_MSK_IAM authentication.</p>
  <p>My MSK is properly configured and I can consume the information using Kafka CLI with the following command:</p>
  <pre><code>../bin/kafka-console-consumer.sh --bootstrap-server b-1.kafka.*********.***********.amazonaws.com:9098 --consumer.config client_auth.properties --topic TopicTest --from-beginning
  </code></pre>
  <p>My client_auth.properties has the following information:</p>
  <pre><code># Sets up TLS for encryption and SASL for authN.
  security.protocol = SASL_SSL
  
  # Identifies the SASL mechanism to use.
  sasl.mechanism = AWS_MSK_IAM
  
  # Binds SASL client implementation.
  sasl.jaas.config = software.amazon.msk.auth.iam.IAMLoginModule required;
  
  # Encapsulates constructing a SigV4 signature based on extracted credentials.
  # The SASL client bound by &quot;sasl.jaas.config&quot; invokes this class.
  sasl.client.callback.handler.class = software.amazon.msk.auth.iam.IAMClientCallbackHandler
  </code></pre>
  <p><strong>When I try to consume from my Databricks cluster using spark, I receive the following error:</strong></p>
  <pre><code>Caused by: kafkashaded.org.apache.kafka.common.KafkaException: java.lang.ClassCastException: software.amazon.msk.auth.iam.IAMClientCallbackHandler cannot be cast to kafkashaded.org.apache.kafka.common.security.auth.AuthenticateCallbackHandler
  </code></pre>
  <p>Here is my cluster config:
  <a href="https://i.stack.imgur.com/KKz0N.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/KKz0N.png" alt="enter image description here" /></a></p>
  <p>The libraries I'm using in the cluster:</p>
  <p><a href="https://i.stack.imgur.com/zJLFS.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/zJLFS.png" alt="enter image description here" /></a></p>
  <p>And the code I'm running on Databricks:</p>
  <pre><code>raw = (
      spark
          .readStream
          .format('kafka')
          .option('kafka.bootstrap.servers', 'b-.kafka.*********.***********.amazonaws.com:9098')
          .option('subscribe', 'TopicTest') 
          .option('startingOffsets', 'earliest')
          .option('kafka.sasl.mechanism', 'AWS_MSK_IAM')
          .option('kafka.security.protocol', 'SASL_SSL')
          .option('kafka.sasl.jaas.config', 'software.amazon.msk.auth.iam.IAMLoginModule required;')
          .option('kafka.sasl.client.callback.handler.class', 'software.amazon.msk.auth.iam.IAMClientCallbackHandler')
          .load()
  )
  </code></pre>