Dataflow Python SDK 自定义Docker镜像下Kafka JKS认证失败求助

问题 #

构建基于Python的Apache Beam管道从Kafka读取数据，Kafka要求通过Truststore和Keystore JKS文件认证。配置代码如下：

kafka_consumer_config = {
"bootstrap.servers": "xxx",
"security.protocol": "SSL",
"ssl.truststore.location":  "/template/truststore.jks",
"ssl.truststore.password": "xxx",
"ssl.keystore.location": "/template/keystore.jks",
"ssl.keystore.password": "xxx",
"ssl.key.password": "xxx",
"group.id": "xxx",
"auto.offset.reset": "earliest",
"enable.auto.commit": "false",
"max.poll.records": "100000"
}

with beam.Pipeline(options=pipeline_options) as pipeline:
    kafka_records = (
        pipeline 
        | 'ReadFromKafka' >> ReadFromKafka(
            consumer_config=kafka_consumer_config,
            topics=kafka_topics,
            with_metadata=True,
        )
    )

运行时抛出错误：

Error message from worker: generic::unknown: org.apache.beam.sdk.util.UserCodeException: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
        org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)

Caused by: org.apache.kafka.common.KafkaException: Failed to load SSL keystore /template/keystore.jks of type JKS

Modification time of key store could not be obtained: /template/keystore.jks

已确认JKS文件存在于Dataflow Worker镜像中，/template/目录已设置chmod 666权限。使用的基础镜像是apache/beam_python3.12_sdk:latest，已安装Open JDK 17，Worker镜像类路径包含以下JAR包：

• beam-sdks-java-io-expansion-service-2.66.0.jar
• beam-sdks-java-io-google-cloud-platform-expansion-service-2.66.0.jar
  Dataflow版本为2.66.0。询问遗漏的配置或步骤。

排查与解决方案 #

1. 检查JKS文件自身权限
   仅设置目录权限为666不够，需确保文件本身有可读权限。执行以下命令：

   chmod 644 /template/keystore.jks /template/truststore.jks
   

   保证Worker进程能读取文件内容。

2. 验证JKS文件有效性
   使用keytool命令检查JKS文件是否正常：

   keytool -list -v -keystore /template/keystore.jks -storepass <你的keystore密码>
   keytool -list -v -keystore /template/truststore.jks -storepass <你的truststore密码>
   

   若命令报错，说明JKS文件损坏或密码错误，需重新生成正确的证书文件。

3. 显式声明SSL存储类型
   在kafka_consumer_config中添加以下配置项：

   "ssl.keystore.type": "JKS",
   "ssl.truststore.type": "JKS"
   

   虽然JKS是默认类型，但某些环境下自动检测可能失败，显式声明可避免此类问题。

4. 确认Worker运行用户的访问权限
   Dataflow Worker通常以beam用户运行，需确保该用户拥有目录和文件的访问权限。镜像构建时添加：

   chown beam:beam /template /template/keystore.jks /template/truststore.jks
   

5. 检查Kafka客户端版本兼容性
   Beam 2.66.0依赖的Kafka客户端版本可能与目标集群不匹配，导致SSL握手异常。查看Beam官方文档确认兼容的Kafka版本，必要时在自定义镜像中调整Kafka客户端依赖。

6. 验证JDK SSL配置
   OpenJDK 17对SSL证书要求更严格，检查JKS文件中的证书是否过期、签名算法是否被当前JDK支持，避免因证书规范问题导致加载失败。

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