使用AWS_MSK_IAM认证在Databricks Spark消费AWS MSK时触发ClassCastException错误排查求助

这个错误的核心原因很明确：Databricks的Spark Kafka连接器使用了被shaded（包路径重命名）的Kafka客户端，也就是kafkashaded.org.apache.kafka开头的类，而你使用的AWS MSK IAM认证库是基于原生org.apache.kafka包编译的，两者的类加载环境和包路径不兼容，导致无法完成类型转换。

下面是两种解决方案，优先推荐第一种：

方案1：使用Databricks内置的MSK IAM支持（最简单可靠） #

Databricks已经预打包了适配自身shaded Kafka客户端的MSK IAM认证库，你只需要修改配置参数，使用Databricks提供的shaded版本的回调处理器即可：

修改后的Spark代码 #

raw = ( spark
  .readStream
  .format('kafka')
  .option('kafka.bootstrap.servers', 'b-1.kafka.*********.***********.amazonaws.com:9098')
  .option('subscribe', 'TopicTest')
  .option('startingOffsets', 'earliest')
  .option('kafka.security.protocol', 'SASL_SSL')
  .option('kafka.sasl.mechanism', 'AWS_MSK_IAM')
  # 若使用集群的Instance Profile认证，无需指定awsRoleArn；若要指定特定IAM角色，添加该参数
  .option('kafka.sasl.jaas.config', 'software.amazon.msk.auth.iam.IAMLoginModule required;')
  # 关键：使用Databricks shaded版本的回调处理器
  .option('kafka.sasl.client.callback.handler.class', 'com.databricks.kafka.shaded.software.amazon.msk.auth.iam.IAMClientCallbackHandler')
  .load()
)

额外权限配置 #

确保你的Databricks集群使用的身份（Instance Profile或Service Principal）拥有MSK的访问权限，需要在IAM中配置类似以下的策略：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kafka-cluster:Connect",
        "kafka-topic:Describe",
        "kafka-topic:Read"
      ],
      "Resource": [
        "arn:aws:kafka:REGION:ACCOUNT_ID:cluster/YOUR_MSK_CLUSTER/*",
        "arn:aws:kafka:REGION:ACCOUNT_ID:topic/YOUR_MSK_CLUSTER/TopicTest"
      ]
    }
  ]
}

同时要将这个IAM身份添加到MSK集群的IAM认证策略中，允许它访问目标主题。

方案2：自定义适配shaded Kafka的MSK IAM库（不推荐） #

如果你必须使用自己的MSK IAM库，需要重新编译该库，将所有org.apache.kafka的引用替换为kafkashaded.org.apache.kafka，然后将编译后的JAR包上传到Databricks集群。这种方法维护成本高，容易因为版本不匹配出现新问题，所以仅在特殊场景下使用。

验证要点 #

1. 确认Databricks集群的Spark版本与MSK集群的Kafka版本兼容（可参考Databricks官方文档的版本对应表）
2. 检查集群的Instance Profile/Service Principal是否正确关联了MSK权限策略
3. 确保代码中没有同时加载原生Kafka和shaded Kafka的依赖，避免类冲突

内容的提问来源于stack exchange，提问作者Danilo Cairolli