Apache Flink JobManager无法找到自定义指标报告

Question

我正在使用Apache Flink版本1.13.1

我写了一个自定义的指标报告，但是JobManager似乎没有识别它。在启动时，JobManager显示以下警告日志:

2021-08-25 14:54:06,243 WARN  org.apache.flink.runtime.metrics.ReporterSetup               [] - The reporter factory (org.apache.flink.metrics.kafka.KafkaReporterFactory) could not be found for reporter kafka. Available factories: [org.apache.flink.metrics.slf4j.Slf4jReporterFactory, org.apache.flink.metrics.datadog.DatadogHttpReporterFactory, org.apache.flink.metrics.prometheus.PrometheusPushGatewayReporterFactory, org.apache.flink.metrics.graphite.GraphiteReporterFactory, org.apache.flink.metrics.statsd.StatsDReporterFactory, org.apache.flink.metrics.prometheus.PrometheusReporterFactory, org.apache.flink.metrics.jmx.JMXReporterFactory, org.apache.flink.metrics.influxdb.InfluxdbReporterFactory].
2021-08-25 14:54:06,245 INFO  org.apache.flink.runtime.metrics.MetricRegistryImpl          [] - No metrics reporter configured, no metrics will be exposed/reported.

我在Flink插件文件夹中有一个名为metrics-kafka的文件夹，其中包含用于度量报告的打包jar。我也复制了这个jar到lib文件夹，这两个都不起作用。请参阅下面使用的配置和代码。

Flink配置文件:

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

指标报告器工厂类:

package org.apache.flink.metrics.kafka
import org.apache.flink.metrics.reporter.{InterceptInstantiationViaReflection, MetricReporter, MetricReporterFactory}
import java.util.Properties
@InterceptInstantiationViaReflection(reporterClassName = "org.apache.flink.metrics.kafka.KafkaReporter")
class KafkaReporterFactory extends MetricReporterFactory{
override def createMetricReporter(properties: Properties): MetricReporter = {
new KafkaReporter()
}
}

度量报告类:

package org.apache.flink.metrics.kafka
import org.apache.flink.metrics.MetricConfig
import org.apache.flink.metrics.reporter.{InstantiateViaFactory, Scheduled}
@InstantiateViaFactory(factoryClassName = "org.apache.flink.metrics.kafka.KafkaReporterFactory")
class KafkaReporter extends MyAbstractReporter with Scheduled{
...
}

Answer 1

我发现我需要添加一个名为org.apache.flink.metrics.reporter.MetricReporterFactory的文件，/resources/META-INF/services/的内容为org.apache.flink.metrics.kafka.KafkaReporterFactory。