Spark集群模式下Kafka JKS证书拉取失败及S3适配咨询

问题背景 #

在Spark集群模式下运行流处理任务连接Kafka时，出现以下SSL相关错误：

ERROR SslEngineBuilder: Modification time of key store could not be obtained: hdfs://ip:port/user/hadoop/jks/kafka.client.truststore.jks
java.nio.file.NoSuchFileException: hdfs:/ip:port/user/hadoop/jks/kafka.client.truststore.jks

已通过HDFS命令验证JKS文件存在于对应路径：

hadoop fs -ls hdfs://ip:port/user/hadoop/<folder1>

Spark项目中连接Kafka的核心配置代码如下：

sparkSession.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "")
.option("subscribe", "")
...
.option("kafka.ssl.keystore.password", "pswd")
.option("kafka.ssl.key.password", "pswrd")      
.option("kafka.ssl.truststore.location","hdfs:///node:port/user/hadoop/<folder1>/kafka.client.truststore.jks")
.option("kafka.ssl.keystore.location", "hdfs:///node:port/user/hadoop/<folder1>/kafka.client.keystore.jks")

咨询问题 #

1. 当前配置缺失什么？
2. 如何使用S3中的JKS文件实现相同的Kafka连接？

问题解答 #

1. 当前配置缺失项分析 #

• Kafka客户端不支持HDFS路径：Kafka原生客户端无法直接读取HDFS上的密钥文件，Spark仅传递配置给Kafka客户端，因此即使文件在HDFS存在，Kafka也无法访问。
• 缺少文件集群分发配置：需要将HDFS上的JKS文件分发到Spark集群所有节点的本地路径，或通过Spark的--files参数将文件上传到节点临时目录，再配置本地路径。
• 路径一致性问题：错误日志中的路径是hdfs://ip:port/user/hadoop/jks/...，但代码配置的是hdfs:///node:port/user/hadoop/<folder1>/...，需确认节点地址、文件夹名称完全匹配，避免拼写错误。

具体修复步骤：

• 提交任务时用--files参数上传HDFS上的JKS文件：

  spark-submit \
    --files hdfs://ip:port/user/hadoop/<folder1>/kafka.client.truststore.jks,hdfs://ip:port/user/hadoop/<folder1>/kafka.client.keystore.jks \
    --class your.main.Class \
    your-jar-file.jar
  

• 修改代码中的路径为本地文件名（--files会将文件放到节点当前工作目录/临时目录，直接用文件名即可）：

  .option("kafka.ssl.truststore.location", "kafka.client.truststore.jks")
  .option("kafka.ssl.keystore.location", "kafka.client.keystore.jks")
  

• 补充SSL协议启用配置：

  .option("kafka.security.protocol", "SSL")
  .option("kafka.ssl.enabled.protocols", "TLSv1.2,TLSv1.3")
  

2. 使用S3中的JKS文件连接Kafka的配置方法 #

Kafka客户端同样无法直接读取S3路径，需先将文件同步到集群节点本地，再配置本地路径：

方法一：通过Spark --files参数加载S3文件

• 提交任务时指定S3路径的JKS文件（需确保Spark集群已配置S3访问权限，比如AWS密钥环境变量或Hadoop配置）：

  spark-submit \
    --files s3://your-bucket/path/kafka.client.truststore.jks,s3://your-bucket/path/kafka.client.keystore.jks \
    --class your.main.Class \
    your-jar-file.jar
  

• 代码中使用本地文件名配置：

  .option("kafka.ssl.truststore.location", "kafka.client.truststore.jks")
  .option("kafka.ssl.keystore.location", "kafka.client.keystore.jks")
  

方法二：提前同步S3文件到集群节点本地

• 使用集群管理工具（如Ansible）将S3文件下载到所有Spark节点的统一本地路径（比如/opt/spark/ssl/）：

  aws s3 cp s3://your-bucket/path/kafka.client.truststore.jks /opt/spark/ssl/
  aws s3 cp s3://your-bucket/path/kafka.client.keystore.jks /opt/spark/ssl/
  

• 代码中配置本地绝对路径：

  .option("kafka.ssl.truststore.location", "/opt/spark/ssl/kafka.client.truststore.jks")
  .option("kafka.ssl.keystore.location", "/opt/spark/ssl/kafka.client.keystore.jks")
  

注意：无论哪种方法，需确保集群所有节点的JKS文件版本一致、Spark运行用户有读取权限，同时配置好Kafka SSL的必要参数。

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