Flink1.13.1 JobManager无法识别自定义Kafka Metrics Reporter问题求助

• SPI服务注册缺失：Flink 1.13版本通过Java SPI机制加载MetricReporterFactory实现类，你当前没有在jar包中配置SPI服务发现文件，Flink无法扫描到你自定义的KafkaReporterFactory，所以日志里显示的可用工厂列表中没有你自己的实现。
• 配置冗余冲突：你在Flink配置中同时配置了metrics.reporter.kafka.factory.class和metrics.reporter.kafka.class，两类配置会产生优先级冲突，干扰加载逻辑。
• 打包或部署不符合要求：如果jar包打包时没有正确包含类文件、或者部署后没有重启所有集群节点，也会导致类加载失败。

• 补充SPI服务配置：在项目的src/main/resources目录下新建层级目录META-INF/services，在该目录下创建名为org.apache.flink.metrics.reporter.MetricReporterFactory的文本文件，文件内容写入自定义工厂类的全限定名：

org.apache.flink.metrics.kafka.KafkaReporterFactory

重新打包项目，确保该配置文件被正确打入jar包的对应路径下。

• 精简Flink配置：修改配置文件，删除metrics.reporter.kafka.class这一行，只保留以下配置即可：

metrics.reporter.kafka.factory.class: org.apache.flink.metrics.kafka.KafkaReporterFactory
metrics.reporter.kafka.interval: 15 SECONDS

• 校验部署逻辑：如果将jar放在plugins/metrics-kafka目录，确保该目录下仅存放自定义Reporter的jar及其私有依赖，避免和Flink自带的公共依赖产生版本冲突；如果放在lib目录，需要重启所有JobManager和TaskManager节点，让集群类加载器加载新的jar包。
• 校验打包正确性：解压打好的jar包，确认两点：1. org/apache/flink/metrics/kafka路径下存在KafkaReporterFactory.class和KafkaReporter.class文件；2. META-INF/services/org.apache.flink.metrics.reporter.MetricReporterFactory文件存在且内容正确。如果是Scala编写的代码，无需将scala-library打入jar包，Flink集群已自带对应版本的Scala依赖。

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