如何为Kafka-BigQuery Dataflow Flex模板配置安全参数？

问题背景 #

使用Dataflow的Kafka_to_BigQuery Flex模板连接AWS MSK实例时，因缺少SASL_SSL协议、SCRAM认证及信任库配置，出现以下错误：

java.lang.RuntimeException: org.apache.kafka.common.errors.TimeoutException: Timeout expired while fetching topic metadata

以下是三个关键配置的具体实现步骤：

1. 配置security.protocol=SASL_SSL #

直接通过kafkaConsumerProperties参数将Kafka消费者安全协议传递给模板，在--parameters中添加：

kafkaConsumerProperties="security.protocol=SASL_SSL,sasl.mechanism=SCRAM-SHA-512"

（注：sasl.mechanism需匹配AWS MSK配置的SCRAM算法，通常为SHA-256或SHA-512）

2. 配置SCRAM认证信息 #

有两种实现方式：

方式一：直接在参数中指定JAAS配置（无需上传xyz.conf）

在kafkaConsumerProperties中追加JAAS配置：

kafkaConsumerProperties="security.protocol=SASL_SSL,sasl.mechanism=SCRAM-SHA-512,sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"abcd\" password=\"1234\";"

方式二：使用JAAS配置文件

• 先将xyz.conf上传至GCS路径，例如gs://<your-bucket>/kafka-config/xyz.conf
• 在gcloud命令中添加参数，将文件同步到Dataflow Worker：

  --parameters filesToStage=gs://<your-bucket>/kafka-config/xyz.conf
  

• 添加JVM启动参数指定配置文件路径：

  --jvm-flags="-Djava.security.auth.login.config=/dataflow/tmp/xyz.conf"
  

（注：filesToStage指定的GCS文件会被自动复制到Worker的/dataflow/tmp目录）

3. 配置GCS存储的JKS信任库 #

• 将JKS信任库文件上传至GCS路径，例如gs://<your-bucket>/kafka-config/truststore.jks
• 在gcloud命令中添加参数同步文件到Worker（若同时使用JAAS配置文件，需将两个文件路径用逗号分隔）：

  --parameters filesToStage=gs://<your-bucket>/kafka-config/truststore.jks
  

• 添加JVM启动参数指定信任库信息：

  --jvm-flags="-Djavax.net.ssl.trustStore=/dataflow/tmp/truststore.jks -Djavax.net.ssl.trustStorePassword=<your-truststore-password>"
  

完整gcloud命令示例 #

（以直接指定JAAS配置为例）

gcloud dataflow flex-template run first-kafka \
--template-file-gcs-location gs://dataflow-templates-us-east4/latest/flex/Kafka_to_BigQuery \
--region us-east4 \
--worker-region us-east4 \
--subnetwork <subnetwork-url> \
--parameters inputTopics=topic1,bootstrapServers=<bootstrap server>,outputTableSpec=<Bigquery table>,stagingLocation=gs://<xxx>/staging-dataflow,dataflowKmsKey=<CMS Key>,serviceAccount=XYZ@developer.gserviceaccount.com,kafkaConsumerProperties="security.protocol=SASL_SSL,sasl.mechanism=SCRAM-SHA-512,sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username=\"abcd\" password=\"1234\";" \
--jvm-flags="-Djavax.net.ssl.trustStore=/dataflow/tmp/truststore.jks -Djavax.net.ssl.trustStorePassword=<your-truststore-password>" \
--parameters filesToStage=gs://<your-bucket>/kafka-config/truststore.jks

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