Spark Structured Streaming用Kafka消费EventHub（SP密钥认证）遇类找不到错误

核心原因 #

org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerLoginCallbackHandler类属于Kafka OAuthBearer认证的扩展组件，Synapse Spark环境默认的Kafka依赖包中可能未包含该类，或版本不匹配导致无法找到。

解决方案 #

1. 补充缺失的Kafka依赖包 #

在Synapse Notebook开头添加maven依赖声明，引入包含该类的kafka-clients包，注意版本需与Synapse Spark集成的Kafka版本匹配（例如Spark 3.2对应Kafka 2.8.x）：

%maven org.apache.kafka:kafka-clients:2.8.1

注：若使用的是Spark 3.3+，可对应升级Kafka版本至3.2.x，需保证版本兼容性。

2. 简化配置（优先尝试） #

在大多数Service Principal密钥认证场景下，无需手动指定回调处理器，OAuthBearerLoginModule会自动处理认证流程。可尝试移除kafka.sasl.login.callback.handler.class配置项，修改后的kafka_options如下：

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 = {
    "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"
}

3. 检查Spark池全局依赖配置 #

若使用Synapse专用Spark池，可通过以下方式全局添加依赖：

• 进入Spark池的库配置页面
• 上传对应版本的kafka-clients.jar包，或从Maven中央仓库搜索添加
• 重启Spark池后再运行Notebook

额外验证点 #

• 确认event_hubs_server为EventHub命名空间的完整FQDN（如xxx.servicebus.windows.net）
• 检查Service Principal的权限：需拥有EventHub的Azure Event Hubs Data Receiver角色权限
• 验证JAAS配置中的clientId、clientSecret、tenant_id参数是否正确

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