[ISSUE#173]Connect Exporter for Prometheus - #200
Conversation
|
|
||
|
|
||
| public class MetricsExportSinkTask extends SinkTask { | ||
| private List<MetricsExporter> metricsExporters; |
There was a problem hiding this comment.
Has the implementation of MetricsExporter been submitted yet?
There was a problem hiding this comment.
Sorry not yet :( . But I am trying my best to finish a exporter for Prometheus which gets the metrics from the log files, and some preliminary knowledge is got. And I hear the sftp source connector (my upstream task) is designing in the last weekly meeting. So I'm waiting for the implementation ConnectRecord in it in order to complete my parsing.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR modifies 6 file(s) with 198 lines of diff. No test changes detected — consider adding test coverage.
Automated review by github-manager-bot
| @@ -0,0 +1,41 @@ | |||
| package org.apache.rocket.connect.metrics.export.sink.connector; | |||
There was a problem hiding this comment.
No test changes detected alongside source modifications. Consider adding tests to cover the changes.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR introduces the interface scaffolding for a Prometheus metrics exporter connector for RocketMQ Connect. It defines the MetricsExporter SPI interface, a SinkConnector/SinkTask pair that delegates to discovered exporters via ServiceLoader, and a utility class for service discovery.
The design is sound (SPI-based extensibility), but the implementation has several issues that need attention before this can be merged:
- Critical: The SPI service file is empty and has a truncated filename, so no implementations will be discovered at runtime
- Warnings: Class name typo, instance initializer ordering, and missing null safety
Additionally, this PR only adds interfaces — there is no actual Prometheus exporter implementation included. Is this intentional (to be followed by another PR), or should the implementation be part of this PR?
Note: This PR has been open since July 2022. If it is still actively being worked on, please address the above issues. If it has been abandoned, consider closing it.
Automated review by "github-manager-bot"
Additional notes (not anchored to a changed line)
- [CRITICAL]
connectors/rocketmq-connect-metrics-exporter/src/main/resources/META-INF/service/org.apache.rocket.connect.metrics.export.sink.connector.MetricsExport:1— This SPI service file is empty and has a truncated filename (should end with MetricsExporter, not MetricsExport). Without a valid SPI entry, ServiceLoader will find no implementations. (line outside diff)
|
|
||
| /** | ||
| * @author: ming | ||
| */ |
There was a problem hiding this comment.
Typo in class name: ServiceProvicerUtil should be ServiceProviderUtil (missing d in Provider).
|
|
||
| private List<MetricsExporter> metricsExporters; | ||
| { | ||
| metricsExporters = ServiceProvicerUtil.getMetricsExporterServices(); |
There was a problem hiding this comment.
Instance initializer block runs at construction time, before start(). Consider moving ServiceLoader initialization to start() to avoid potential classpath issues.
| private List<MetricsExporter> metricsExporters; | ||
|
|
||
| @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException { | ||
| for (MetricsExporter exporter : metricsExporters) { |
There was a problem hiding this comment.
metricsExporters is initialized in init() but used in put() without null check. If put() is somehow called before init(), this will NPE. Consider adding a null guard or lazy initialization.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
PR received and logged for review. This PR requires detailed code review by a maintainer.
Diff size: 198 lines
Author: yhx-coder (CONTRIBUTOR)
Automated review by RockteMQ-AI
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
Review of PR #200: [ISSUE#173]Connect Exporter for Prometheus
Findings: 8 issue(s) identified (1 critical).
CLA: unknown
Please address the inline comments above.
Automated review by github-manager-bot
Additional notes (not anchored to a changed line)
- [CRITICAL]
connectors/rocketmq-connect-metrics-exporter/src/main/resources/META-INF/service/org.apache.rocket.connect.metrics.export.sink.connector.MetricsExport— SPI service file has two fatal errors: (1) directory must be META-INF/services/ (plural), not META-INF/service/; (2) filename must be the fully-qualified interface name 'org.apache.rocket.connect.metrics.export.sink.connector.MetricsExporter' (missing trailing 'r'). Additionally the file is empty — no implementation classes are registered. ServiceLoader.load(MetricsExporter.class) will find zero implementations, making the entire connector a silent no-op. (missing file/line/body)
| private List<MetricsExporter> metricsExporters; | ||
| { | ||
| metricsExporters = ServiceProvicerUtil.getMetricsExporterServices(); | ||
| } |
There was a problem hiding this comment.
metricsExporters is loaded in an instance initializer block but never closed/stopped in stop(). If exporters hold resources (e.g., HTTP clients, thread pools for Prometheus scraping), they will leak because the connector's stop() only nulls out config. Consider iterating metricsExporters in stop() and calling stop() on each, or documenting that lifecycle is fully managed by the task.
| private List<MetricsExporter> metricsExporters; | ||
|
|
||
| @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException { | ||
| for (MetricsExporter exporter : metricsExporters) { |
There was a problem hiding this comment.
In put(), if one exporter throws an exception, all subsequent exporters in the loop are skipped and the entire batch is lost. For a multi-exporter setup this means one faulty exporter blocks all others. Consider catching per-exporter, logging, and continuing — or aggregating exceptions before rethrowing.
| private List<MetricsExporter> metricsExporters; | ||
|
|
||
| @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException { | ||
| for (MetricsExporter exporter : metricsExporters) { |
There was a problem hiding this comment.
If metricsExporters is empty (which it will be given the broken SPI file), put() silently discards all incoming ConnectRecords with no log or error. A warning or exception when the list is empty during start() or init() would prevent silent data loss.
| * @author: ming | ||
| */ | ||
| public class ServiceProvicerUtil { | ||
|
|
There was a problem hiding this comment.
Class name 'ServiceProvicerUtil' contains a typo — should be 'ServiceProviderUtil'. This is a public API surface; fixing it later would be a breaking change.
| List<KeyValue> configs = new ArrayList<>(); | ||
| configs.add(config); | ||
| return configs; | ||
| } |
There was a problem hiding this comment.
taskConfigs(int maxTasks) always returns a single-element list regardless of maxTasks. If the framework distributes work across multiple tasks for parallelism, this silently caps it at one. Either honor maxTasks or explicitly document that this connector supports only a single task.
| xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" | ||
| xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> | ||
| <modelVersion>4.0.0</modelVersion> | ||
|
|
There was a problem hiding this comment.
pom.xml has no parent POM and uses a standalone hardcoded version '1.0-SNAPSHOT' with a standalone groupId. Other connector modules in this repo likely inherit from a parent POM for consistent dependency management. This may cause version drift and build inconsistency.
| @@ -0,0 +1,47 @@ | |||
| <?xml version="1.0" encoding="UTF-8"?> | |||
There was a problem hiding this comment.
No test files are included in this PR. There is no unit or integration test coverage for the connector, task, or SPI loading logic — critical paths like put(), start()/stop() lifecycle, and validate() are completely untested.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
8 finding(s) to address.
Findings
- [CRITICAL]
connectors/rocketmq-connect-metrics-exporter/src/main/resources/META-INF/service/org.apache.rocket.connect.metrics.export.sink.connector.MetricsExport:0— SPI service file is in the wrong directory and has the wrong filename. Java'sServiceLoaderrequires the pathMETA-INF/services/(plural) and the filename must exactly match the fully-qualified interface name:org.apache.rocket.connect.metrics.export.sink.connector.MetricsExporter. The current path usesservice/(singular) and the filename isMetricsExport(missing the trailing 'r'). This meansServiceLoader.load(MetricsExporter.class)will silently return zero implementations at runtime, making the entire connector non-functional. - [CRITICAL]
connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExportSinkTask.java:13—metricsExportersis not initialized at declaration — it is only assigned insideinit(). If the framework ever callsstart()orput()beforeinit(), this will throw a NullPointerException. InMetricsExportSinkConnector, the equivalent field is initialized via an instance-initializer block. The task should do the same, or at minimum initialize at declaration:private List<MetricsExporter> metricsExporters = ServiceProvicerUtil.getMetricsExporterServices(); - [WARNING]
connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/util/ServiceProvicerUtil.java:12— Class name contains a typo:ServiceProvicerUtilshould beServiceProviderUtil. This typo propagates to every call site (connector and task classes) and will be a permanent API blemish if not fixed before merge. - [WARNING]
connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExportSinkConnector.java:21—taskConfigs(int maxTasks)always returns a single-element list regardless ofmaxTasks. This caps the connector to exactly one task even when the runtime requests more. For a metrics exporter this may be intentional, but it should be documented or explicitly guarded — silently ignoringmaxTaskscan confuse operators who configure a higher parallelism. - [WARNING]
connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExportSinkTask.java:15— No error handling input(). If any singleMetricsExporter.export()throws an exception, the remaining exporters in the list are skipped and the entire batch is lost. Consider catching per-exporter exceptions so one faulty exporter does not block the others, and log or route failures appropriately. - [INFO]
connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExportSinkConnector.java:15— ThemetricsExportersfield is populated in an instance-initializer block, meaning ServiceLoader runs at construction time — beforestart()orvalidate(). If the connector is constructed but never started (e.g., validation fails), the SPI scan is wasted work. Consider lazy initialization instart()for consistency with the connector lifecycle. - [INFO]
connectors/rocketmq-connect-metrics-exporter/pom.xml:8— This module defines its own<groupId>and<version>instead of inheriting from the parent POM (<parent>block is absent). This means it won't be built as part of the reactor, won't inherit dependency management, and won't receive the project-wide license/enforcer plugins. It will be an orphaned module that must be built and released independently. - [INFO]
connectors/rocketmq-connect-metrics-exporter/src/main/java/org/apache/rocket/connect/metrics/export/sink/connector/MetricsExportSinkTask.java:21—start()does not callsuper.start(config). While the currentSinkTaskbase may not require it, omitting the super call can break lifecycle contracts in future framework versions. Same applies tostop().
Automated review by github-manager-bot
Additional notes (not anchored to a changed line)
- [CRITICAL]
connectors/rocketmq-connect-metrics-exporter/src/main/resources/META-INF/service/org.apache.rocket.connect.metrics.export.sink.connector.MetricsExport— SPI service file is in the wrong directory and has the wrong filename. Java'sServiceLoaderrequires the pathMETA-INF/services/(plural) and the filename must exactly match the fully-qualified interface name:org.apache.rocket.connect.metrics.export.sink.connector.MetricsExporter. The current path usesservice/(singular) and the filename isMetricsExport(missing the trailing 'r'). This meansServiceLoader.load(MetricsExporter.class)will silently return zero implementations at runtime, making the entire connector non-functional. (missing file/line/body)
|
|
||
|
|
||
| public class MetricsExportSinkTask extends SinkTask { | ||
| private List<MetricsExporter> metricsExporters; |
There was a problem hiding this comment.
metricsExporters is not initialized at declaration — it is only assigned inside init(). If the framework ever calls start() or put() before init(), this will throw a NullPointerException. In MetricsExportSinkConnector, the equivalent field is initialized via an instance-initializer block. The task should do the same, or at minimum initialize at declaration: private List<MetricsExporter> metricsExporters = ServiceProvicerUtil.getMetricsExporterServices();
| /** | ||
| * @author: ming | ||
| */ | ||
| public class ServiceProvicerUtil { |
There was a problem hiding this comment.
Class name contains a typo: ServiceProvicerUtil should be ServiceProviderUtil. This typo propagates to every call site (connector and task classes) and will be a permanent API blemish if not fixed before merge.
| @Override public List<KeyValue> taskConfigs(int maxTasks) { | ||
| List<KeyValue> configs = new ArrayList<>(); | ||
| configs.add(config); | ||
| return configs; |
There was a problem hiding this comment.
taskConfigs(int maxTasks) always returns a single-element list regardless of maxTasks. This caps the connector to exactly one task even when the runtime requests more. For a metrics exporter this may be intentional, but it should be documented or explicitly guarded — silently ignoring maxTasks can confuse operators who configure a higher parallelism.
| public class MetricsExportSinkTask extends SinkTask { | ||
| private List<MetricsExporter> metricsExporters; | ||
|
|
||
| @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException { |
There was a problem hiding this comment.
No error handling in put(). If any single MetricsExporter.export() throws an exception, the remaining exporters in the list are skipped and the entire batch is lost. Consider catching per-exporter exceptions so one faulty exporter does not block the others, and log or route failures appropriately.
|
|
||
| private List<MetricsExporter> metricsExporters; | ||
| { | ||
| metricsExporters = ServiceProvicerUtil.getMetricsExporterServices(); |
There was a problem hiding this comment.
The metricsExporters field is populated in an instance-initializer block, meaning ServiceLoader runs at construction time — before start() or validate(). If the connector is constructed but never started (e.g., validation fails), the SPI scan is wasted work. Consider lazy initialization in start() for consistency with the connector lifecycle.
| <modelVersion>4.0.0</modelVersion> | ||
|
|
||
| <groupId>org.apache.rocketmq</groupId> | ||
| <artifactId>rocketmq-connect-metrics-exporter</artifactId> |
There was a problem hiding this comment.
This module defines its own <groupId> and <version> instead of inheriting from the parent POM (<parent> block is absent). This means it won't be built as part of the reactor, won't inherit dependency management, and won't receive the project-wide license/enforcer plugins. It will be an orphaned module that must be built and released independently.
| } | ||
| } | ||
|
|
||
| @Override public void start(KeyValue config) { |
There was a problem hiding this comment.
start() does not call super.start(config). While the current SinkTask base may not require it, omitting the super call can break lifecycle contracts in future framework versions. Same applies to stop().
3 participants
ISSUE#173 What is the purpose of the change
Some simple interfaces designed for metric exposure.
Brief changelog
Design some interfaces to ensure exposure behavior of metrics provided by dynamic loading.Thus,we can export the metrics to anywhere we want, e.g. prometheus.