Flink Kafka Consumer获取主题元数据失败问题求助

问题概述 #

将单文件Flink Kafka读取代码拆分为类结构后，运行时报错Failed to get metadata for topics [alerts_consumer_group_prod]，底层异常为UnknownTopicOrPartitionException，但确认Kafka集群中存在目标主题及对应分区。

关键报错片段：

Caused by: java.lang.RuntimeException: Failed to get metadata for topics [alerts_consumer_group_prod].
    at org.apache.flink.connector.kafka.source.enumerator.subscriber.KafkaSubscriberUtils.getTopicMetadata(KafkaSubscriberUtils.java:47)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.

错误根源 #

参数传递顺序不匹配导致主题名称被错误赋值为消费者组ID：

1. Main类中实例化AlertRouter的参数顺序：

AlertRouter alertRouter = new AlertRouter(bootstrapServer, alertTopic, alertConsumerGroupID, alertTopicParallelism);

传递的参数依次是：bootstrapServer、alertTopic（实际主题名）、alertConsumerGroupID（消费者组ID）、alertTopicParallelism。

2. AlertRouter构造函数的参数定义：

public AlertRouter(String bootstrapServer, String consumerGroupID, String topic, int parallelism) {
    this.bootstrapServer = bootstrapServer;
    this.consumerGroupID = consumerGroupID;
    this.topic = topic;
    this.parallelism = parallelism;
}

构造函数期望的参数顺序是：bootstrapServer、consumerGroupID、topic、parallelism。

两者顺序完全颠倒，导致AlertRouter中的topic变量被赋值为消费者组IDalerts_consumer_group_prod，Flink Kafka Source尝试订阅这个不存在的主题，因此抛出UnknownTopicOrPartitionException。

修复方案 #

调整Main类中AlertRouter实例化的参数顺序，匹配构造函数的定义：

// 原错误代码
// AlertRouter alertRouter = new AlertRouter(bootstrapServer, alertTopic, alertConsumerGroupID, alertTopicParallelism);

// 修改后正确代码
AlertRouter alertRouter = new AlertRouter(bootstrapServer, alertConsumerGroupID, alertTopic, alertTopicParallelism);

额外优化建议：

• 清理AlertRouter中冗余的config.properties加载代码（已经在Main类中加载过）；
• KafkaSource构建时无需重复设置setBootstrapServers、setGroupId（已通过setProperties传入），避免配置冲突。

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