Customized Metric Reporter can not be found by Flink

[Fan Xie]

Hi Flink Community,

Recently I implemented a customized metric reporter (named: DiagnosticsMessageReporter) to report Flink metrics to a Kafka topic. I built this reporter into a jar file and copy it to /opt/flink/plugins/DiagnosticsMessageReporter/DiagnosticsMessageReporter.jar for both the Job Manager and task manager's containers. But later on I found the following logs indicated that the metric reporter can not be loaded:

^{2021-05-11 23:08:31,523 WARN  org.apache.flink.runtime.metrics.ReporterSetup               [] - The reporter factory (org.apache.flink.metrics.reporter.DiagnosticsMessageReporterFactory) could not be found for reporter DiagnosticsMessageReporter. Available factories: [org.apache.flink.metrics.datadog.DatadogHttpReporterFactory, org.apache.flink.metrics.slf4j.Slf4jReporterFactory, 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-05-11 23:21:55,698 INFO  org.apache.flink.runtime.metrics.MetricRegistryImpl          [] - No metrics reporter configured, no metrics will be exposed/reported.}

The Flink configs I used are as following:

#DiagnosticsMessageReporter configs
metrics.reporters: DiagnosticsMessageReporter
metrics.reporter.DiagnosticsMessageReporter.factory.class: org.apache.flink.metrics.reporter.DiagnosticsMessageReporterFactory
metrics.reporter.DiagnosticsMessageReporter.bootstrap.servers: kafka:9092
metrics.reporter.DiagnosticsMessageReporter.topic: flink-metrics
metrics.reporter.DiagnosticsMessageReporter.keyBy: task_attempt_id
metrics.reporter.DiagnosticsMessageReporter.interval: 1 SECONDS

Does anyone have any idea about what happened here? Am I missing some of the steps to load the customized reporter as a plugin? Really appreciate if someone can help to take a look at this! 

Best,
Fan

[Yangze Guo]

Hi, Fan

Flink loaded the custom reporter through the service loader mechanism.[1]
Do you add the service file in the "resources/META-INF/services" directory?

[1] https://docs.oracle.com/javase/9/docs/api/java/util/ServiceLoader.html

Best,
Yangze Guo

On Wed, May 12, 2021 at 7:53 AM Fan Xie <[hidden email]> wrote:

>
> Hi Flink Community,
>
> Recently I implemented a customized metric reporter (named: DiagnosticsMessageReporter) to report Flink metrics to a Kafka topic. I built this reporter into a jar file and copy it to /opt/flink/plugins/DiagnosticsMessageReporter/DiagnosticsMessageReporter.jar for both the Job Manager and task manager's containers. But later on I found the following logs indicated that the metric reporter can not be loaded:
>
> 2021-05-11 23:08:31,523 WARN  org.apache.flink.runtime.metrics.ReporterSetup               [] - The reporter factory (org.apache.flink.metrics.reporter.DiagnosticsMessageReporterFactory) could not be found for reporter DiagnosticsMessageReporter. Available factories: [org.apache.flink.metrics.datadog.DatadogHttpReporterFactory, org.apache.flink.metrics.slf4j.Slf4jReporterFactory, 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-05-11 23:21:55,698 INFO  org.apache.flink.runtime.metrics.MetricRegistryImpl          [] - No metrics reporter configured, no metrics will be exposed/reported.
>
> The Flink configs I used are as following:
>
> #DiagnosticsMessageReporter configs
> metrics.reporters: DiagnosticsMessageReporter
> metrics.reporter.DiagnosticsMessageReporter.factory.class: org.apache.flink.metrics.reporter.DiagnosticsMessageReporterFactory
> metrics.reporter.DiagnosticsMessageReporter.bootstrap.servers: kafka:9092
> metrics.reporter.DiagnosticsMessageReporter.topic: flink-metrics
> metrics.reporter.DiagnosticsMessageReporter.keyBy: task_attempt_id
> metrics.reporter.DiagnosticsMessageReporter.interval: 1 SECONDS
>
> Does anyone have any idea about what happened here? Am I missing some of the steps to load the customized reporter as a plugin? Really appreciate if someone can help to take a look at this!
>
> Best,
> Fan
>

[Fan Xie]

Hi Yangze,

I add that file and it works! Thanks for helping with this!

Best,

Fan

---

From: Yangze Guo <[hidden email]>
Sent: Tuesday, May 11, 2021 7:12 PM
To: Fan Xie <[hidden email]>
Cc: [hidden email] <[hidden email]>
Subject: Re: Customized Metric Reporter can not be found by Flink

 

Hi, Fan

Flink loaded the custom reporter through the service loader mechanism.[1]
Do you add the service file in the "resources/META-INF/services" directory?

[1] https://nam06.safelinks.protection.outlook.com/?url=https%3A%2F%2Fdocs.oracle.com%2Fjavase%2F9%2Fdocs%2Fapi%2Fjava%2Futil%2FServiceLoader.html&amp;data=04%7C01%7Cfxie%40linkedin.com%7C75fc9e69c88a4104b8cc08d914eb6f14%7C72f988bf86f141af91ab2d7cd011db47%7C1%7C0%7C637563823676666959%7CUnknown%7CTWFpbGZsb3d8eyJWIjoiMC4wLjAwMDAiLCJQIjoiV2luMzIiLCJBTiI6Ik1haWwiLCJXVCI6Mn0%3D%7C1000&amp;sdata=5RRDNm6Acc8ezXYLu3hBls60po7p5lD3PpU0jG8UhL0%3D&amp;reserved=0

Best,
Yangze Guo

On Wed, May 12, 2021 at 7:53 AM Fan Xie <[hidden email]> wrote:
>
> Hi Flink Community,
>
> Recently I implemented a customized metric reporter (named: DiagnosticsMessageReporter) to report Flink metrics to a Kafka topic. I built this reporter into a jar file and copy it to /opt/flink/plugins/DiagnosticsMessageReporter/DiagnosticsMessageReporter.jar for both the Job Manager and task manager's containers. But later on I found the following logs indicated that the metric reporter can not be loaded:
>
> 2021-05-11 23:08:31,523 WARN  org.apache.flink.runtime.metrics.ReporterSetup               [] - The reporter factory (org.apache.flink.metrics.reporter.DiagnosticsMessageReporterFactory) could not be found for reporter DiagnosticsMessageReporter. Available factories: [org.apache.flink.metrics.datadog.DatadogHttpReporterFactory, org.apache.flink.metrics.slf4j.Slf4jReporterFactory, 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-05-11 23:21:55,698 INFO  org.apache.flink.runtime.metrics.MetricRegistryImpl          [] - No metrics reporter configured, no metrics will be exposed/reported.
>
> The Flink configs I used are as following:
>
> #DiagnosticsMessageReporter configs
> metrics.reporters: DiagnosticsMessageReporter
> metrics.reporter.DiagnosticsMessageReporter.factory.class: org.apache.flink.metrics.reporter.DiagnosticsMessageReporterFactory
> metrics.reporter.DiagnosticsMessageReporter.bootstrap.servers: kafka:9092
> metrics.reporter.DiagnosticsMessageReporter.topic: flink-metrics
> metrics.reporter.DiagnosticsMessageReporter.keyBy: task_attempt_id
> metrics.reporter.DiagnosticsMessageReporter.interval: 1 SECONDS
>
> Does anyone have any idea about what happened here? Am I missing some of the steps to load the customized reporter as a plugin? Really appreciate if someone can help to take a look at this!
>
> Best,
> Fan
>