PyFlink 1.16.1连接安全Kafka集群报错，求助排查配置问题

以下是针对TimeoutException: Topic flink-test not present in metadata after 60000 ms错误的排查与修复方案：

1. 校验JAAS配置的准确性 #

• 确保JAAS配置格式完全匹配Kafka要求，核心是KafkaClient模块的参数正确，示例如下：

  KafkaClient {
      org.apache.kafka.common.security.plain.PlainLoginModule required
      username="your-kafka-username"
      password="your-kafka-password";
  };
  

• 在PyFlink中传递JAAS配置时，要么将多行配置用\n拼接成单行字符串，要么通过java.security.auth.login.config系统属性指定外部JAAS文件路径（集群模式下更推荐后者，避免配置转义问题）。

2. 确认SSL信任库配置完整性 #

• 检查ssl.truststore.location和ssl.truststore.password参数，确保信任库文件路径在Flink所有任务节点上一致且可访问（集群模式下可通过分布式缓存上传文件）。
• 用keytool验证信任库是否包含Kafka Broker的SSL证书：

  keytool -list -v -keystore /path/to/your/truststore.jks -storepass your-truststore-password
  

3. 补充核心Kafka生产者配置 #

必须显式配置以下参数，缺一不可：

• bootstrap.servers: Kafka Broker的SSL端口地址（默认9093），多个地址用逗号分隔
• security.protocol: 固定设为SASL_SSL，开启SASL认证+SSL加密
• sasl.mechanism: 与JAAS配置匹配的认证机制（如PLAIN、SCRAM-SHA-256）

4. 排查网络与权限问题 #

• 测试Flink节点与Kafka Broker的SSL端口连通性：

  nc -zv kafka-broker-host 9093
  

• 检查Kafka ACL权限：确保当前认证用户拥有flink-test主题的WRITE权限，以及集群级别的DESCRIBE权限（获取元数据必须）。

5. 正确配置的PyFlink代码示例 #

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors import FlinkKafkaProducer
from pyflink.common.serialization import JsonRowSerializationSchema
from pyflink.common import Types

env = StreamExecutionEnvironment.get_execution_environment()
# 若使用集群模式，需开启checkpoint保证Exactly-Once语义
env.enable_checkpointing(5000)

# 定义数据类型（示例）
row_type_info = Types.ROW([Types.STRING(), Types.INT()])

# Kafka生产者核心配置
kafka_props = {
    "bootstrap.servers": "kafka-broker-1:9093,kafka-broker-2:9093",
    "security.protocol": "SASL_SSL",
    "sasl.mechanism": "PLAIN",
    "sasl.jaas.config": "org.apache.kafka.common.security.plain.PlainLoginModule required username='test-user' password='test-pass';",
    "ssl.truststore.location": "/opt/flink/conf/kafka-truststore.jks",
    "ssl.truststore.password": "truststore123",
    "acks": "all"
}

# 构建JSON序列化器与生产者
serialization_schema = JsonRowSerializationSchema.builder().with_type_info(row_type_info).build()
kafka_producer = FlinkKafkaProducer(
    topic="flink-test",
    serialization_schema=serialization_schema,
    producer_config=kafka_props,
    delivery_guarantee=FlinkKafkaProducer.DeliveryGuarantee.EXACTLY_ONCE
)

# 生成测试数据并写入Kafka
test_data = env.from_collection([("test-key", 1), ("test-key2", 2)])
test_data.add_sink(kafka_producer)

env.execute("PyFlink Kafka Secure Producer")

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