Repository: strimzi/strimzi-kafka-bridge
Pull request #954 (2024-12-17 09:26:38)

Add support for Strimzi Metrics Reporter
This patch adds support for metrics types configuration, integrating the [Strimzi Metrics Reporter](https://github.com/strimzi/metrics-reporter), and custom Prometheus JMX Exporter configuration. The custom configuration feature is needed for Cluster Operator integration, but it can also be leveraged by standalone users. See [proposal 43](https://github.com/strimzi/proposals/pull/143) for more details.


fvaleri: @OwenCorrigan76 fyi
fvaleri: > there will be some backward compatibility impact here or in the operators repo that needs to be carefully considered

I agree. Probably the backwards incompatible changes are mostly on the operator code, where we have to introduce `metricsConfig` at the CRD level. @ppatierno wdyt?
fvaleri: @strimzi/maintainers proposal 92 has been accepted, and this PR is aligned, so open for review. Thanks.
fvaleri: @ppatierno @scholzj your comments should be addressed, but there are a few open questions. For the new abstraction I used the `MetricsCollector` name, as `MetricsReporter` would create some confusion IMO. Let me know what you think.
ppatierno: @fvaleri can you resolve the conflicts before I have another pass please?
fvaleri: > Which kind of validation/spotpug error? We don't get any right now in the main. This refactoring seems to be useless from a pure code perspective.

```sh
[INFO] --- spotbugs:4.7.3.0:check (default-cli) @ kafka-bridge ---
[INFO] BugInstance size is 3
[INFO] Error size is 0
[INFO] Total bugs: 3
[ERROR] Low: Private method io.strimzi.kafka.bridge.http.HttpBridge.contentTypeToFormat(String) is never called [io.strimzi.kafka.bridge.http.HttpBridge] At HttpBridge.java:[lines 644-652] UPM_UNCALLED_PRIVATE_METHOD
[ERROR] High: Found reliance on default encoding in io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter.toKafkaRecord(String, Integer, byte[]): String.getBytes() [io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter, io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter] At HttpTextMessageConverter.java:[line 46]Another occurrence at HttpTextMessageConverter.java:[line 53] DM_DEFAULT_ENCODING
[ERROR] High: Found reliance on default encoding in io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter.toMessages(ConsumerRecords): new String(byte[]) [io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter, io.strimzi.kafka.bridge.http.converter.HttpTextMessageConverter] At HttpTextMessageConverter.java:[line 104]Another occurrence at HttpTextMessageConverter.java:[line 105] DM_DEFAULT_ENCODING
```

> Again, why it's not happening today in the main?

I see these errors on the main branch too. Just tried with a fresh clone.

ppatierno: > I see these errors on the main branch too. Just tried with a fresh clone.

Ok so I see another issue here. Our `make all` is not running the `spotbugs` target automatically and it should be fixed.
@fvaleri could you remove these changes from this PR, and opening a new one where you have those fixes and adding spotbugs as target during the make all process so that it also runs in the Azure and Travis CI pipeline please?
fvaleri: The problem is that we run spotbugs before building in the CI. I'll fix that in the spinoff PR.
fvaleri: @ppatierno rebased. Let's wait for the CI to complete.
ppatierno: @scholzj do you have anything to add here? so we can proceed with merging it?
scholzj: > @scholzj do you have anything to add here? so we can proceed with merging it?

You do not have the sufficient number of approvals, so you cannot proceed with merging it regardless of what I think.
ppatierno: I think you totally mislead what I meant. I just meant if you have anything to add which means review, leaving comments and/or approving. Then we can proceed. I wasn't going to merge without another approval, I know the rules. But because you reviewed this PR already I was just asking if you had anything more.
ppatierno: Adding that even if any other maintainer would have approved it, I was anyway going to ask you if you had anything to add despite I had all the rights to merge.
ppatierno: @fvaleri other than resolving the remaining comments could you also rebase please? We have the fix for the Travis CI pipeline in main now. Thanks!
fvaleri: @PaulRMellor thanks for the great suggestions. Let me know if there is still some doc improvement or issue.
ppatierno: @fvaleri thanks again for your work!

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index dde5421..01878d0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,15 @@
   * Both the `/openapi` and `/openapi/v3` endpoints return the OpenAPI v3 definition of the bridge REST API.
 * Removed script to build bridge configuration within the container. 
   It is going to be set up by the Strimzi operator within a ConfigMap and mounted as volume on the bridge pod.
+* Added support for [Strimzi Metrics Reporter](https://github.com/strimzi/metrics-reporter).
+  Set `bridge.metrics=strimziMetricsReporter` to enable it.
+* Added support for custom Prometheus JMX Exporter configuration.
+  Set `bridge.metrics.exporter.config.path=/path/to/my-exporter-config.yml` to use your custom config.
+
+### Changes, deprecations and removals
+
+* `KAFKA_BRIDGE_METRICS_ENABLED` configuration has been deprecated.
+  Set `bridge.metrics=jmxPrometheusExporter` to keep using Prometheus JMX Exporter.
 
 ## 0.31.1
 
diff --git a/config/application.properties b/config/application.properties
index 423b178..3b58783 100644
--- a/config/application.properties
+++ b/config/application.properties
@@ -1,5 +1,16 @@
 #Bridge related settings
 bridge.id=my-bridge
+
+# uncomment the following line to enable Prometheus JMX Exporter, check the Kafka Bridge documentation for more details
+#bridge.metrics=jmxPrometheusExporter
+# optionally, set the file path of your custom configuration
+#bridge.metrics.exporter.config.path=/path/to/my-exporter-config.yaml
+
+# uncomment the following line to enable Strimzi Metrics Reporter, check the Kafka Bridge documentation for more details
+#bridge.metrics=strimziMetricsReporter
+# optionally, filter the exposed metrics of all internal Kafka clients using a comma separated list of regexes
+#kafka.prometheus.metrics.reporter.allowlist=.*
+
 # uncomment the following line (bridge.tracing) to enable OpenTelemetry tracing, check the documentation for more details
 #bridge.tracing=opentelemetry
 
diff --git a/documentation/assemblies/assembly-kafka-bridge-config.adoc b/documentation/assemblies/assembly-kafka-bridge-config.adoc
index 9cb1238..0302a6b 100644
--- a/documentation/assemblies/assembly-kafka-bridge-config.adoc
+++ b/documentation/assemblies/assembly-kafka-bridge-config.adoc
@@ -6,13 +6,14 @@
 = Kafka Bridge configuration
 
 [role="_abstract"]
-Configure a deployment of the Kafka Bridge using configuration properties.
-Configure Kafka and specify the HTTP connection details needed to be able to interact with Kafka.
+Configure a deployment of the Kafka Bridge with Kafka-related properties and specify the HTTP connection details needed to be able to interact with Kafka.
+Additionally, enable metrics in Prometheus format using either the https://github.com/prometheus/jmx_exporter[Prometheus JMX Exporter] or the https://github.com/strimzi/metrics-reporter[Strimzi Metrics Reporter].
 You can also use configuration properties to enable and use distributed tracing with the Kafka Bridge.
 Distributed tracing allows you to track the progress of transactions between applications in a distributed system.
 
 NOTE: Use the `KafkaBridge` resource to configure properties when you are xref:overview-components-running-kafka-bridge-cluster-{context}[running the Kafka Bridge on Kubernetes].
 
 include::modules/proc-configuring-kafka-bridge.adoc[leveloffset=+1]
-include::modules/proc-configuring-kafka-bridge-metrics.adoc[leveloffset=+1]
-include::modules/proc-configuring-kafka-bridge-tracing.adoc[leveloffset=+1]
\ No newline at end of file
+include::modules/proc-configuring-kafka-bridge-jmx-metrics.adoc[leveloffset=+1]
+include::modules/proc-configuring-kafka-bridge-smr-metrics.adoc[leveloffset=+1]
+include::modules/proc-configuring-kafka-bridge-tracing.adoc[leveloffset=+1]
diff --git a/documentation/book/api/.openapi-generator/openapi.json-generate-apidoc.sha256 b/documentation/book/api/.openapi-generator/openapi.json-generate-apidoc.sha256
index 20d0350..393174d 100644
--- a/documentation/book/api/.openapi-generator/openapi.json-generate-apidoc.sha256
+++ b/documentation/book/api/.openapi-generator/openapi.json-generate-apidoc.sha256
@@ -1,1 +1,1 @@
-638e4ea2961241ead6552ccef364f2617c3b602e4bf9dff7786f5c463b07d063
\ No newline at end of file
+4502368c51039e785930e38d1b134385ac0f20259b840e32f3f96ffee9bbe26c
\ No newline at end of file
diff --git a/documentation/book/api/index.adoc b/documentation/book/api/index.adoc
index 7c54138..e7bb275 100644
--- a/documentation/book/api/index.adoc
+++ b/documentation/book/api/index.adoc
@@ -1399,6 +1399,11 @@ Retrieves the bridge metrics in Prometheus format.
 | Metrics in Prometheus format retrieved successfully.
 |  <<String>>
 
+
+| 404
+| The metrics endpoint is not enabled.
+|  <<>>
+
 |===
 
 ===== Samples
diff --git a/documentation/modules/proc-configuring-kafka-bridge-jmx-metrics.adoc b/documentation/modules/proc-configuring-kafka-bridge-jmx-metrics.adoc
new file mode 100644
index 0000000..b954891
--- /dev/null
+++ b/documentation/modules/proc-configuring-kafka-bridge-jmx-metrics.adoc
@@ -0,0 +1,33 @@
+[id='proc-configuring-kafka-bridge-jmx-metrics-{context}']
+= Configuring Prometheus JMX Exporter metrics
+
+[role="_abstract"]
+Enable the Prometheus JMX Exporter to collect Kafka Bridge metrics by setting the `bridge.metrics` option to `jmxPrometheusExporter`.
+
+.Prerequisites
+
+* xref:proc-downloading-kafka-bridge-{context}[The Kafka Bridge installation archive is downloaded].
+
+.Procedure
+
+. Set the `bridge.metrics` configuration to `jmxPrometheusExporter`.
++
+.Configuration for enabling metrics
+
+[source,properties]
+----
+bridge.metrics=jmxPrometheusExporter
+----
++
+Optionally, you can add a custom Prometheus JMX Exporter configuration using the `bridge.metrics.exporter.config.path` property.
+If not configured, a default embedded configuration file is used.
+
+. Run the Kafka Bridge run script.
++
+.Running the Kafka Bridge
+[source,shell]
+----
+./bin/kafka_bridge_run.sh --config-file=<path>/application.properties
+----
++
+With metrics enabled, you can scrape metrics in Prometheus format from the `/metrics` endpoint of the Kafka Bridge.
diff --git a/documentation/modules/proc-configuring-kafka-bridge-metrics.adoc b/documentation/modules/proc-configuring-kafka-bridge-metrics.adoc
deleted file mode 100644
index 94a9c37..0000000
--- a/documentation/modules/proc-configuring-kafka-bridge-metrics.adoc
+++ /dev/null
@@ -1,30 +0,0 @@
-[id='proc-configuring-kafka-bridge-metrics-{context}']
-= Configuring metrics
-
-[role="_abstract"]
-Enable metrics for the Kafka Bridge by setting the `KAFKA_BRIDGE_METRICS_ENABLED` environment variable.
-
-.Prerequisites
-
-* xref:proc-downloading-kafka-bridge-{context}[The Kafka Bridge installation archive is downloaded].
-
-.Procedure
-
-. Set the environment variable for enabling metrics to `true`.
-+
-.Environment variable for enabling metrics
-
-[source,properties]
-----
-KAFKA_BRIDGE_METRICS_ENABLED=true
-----
-
-. Run the Kafka Bridge script to enable metrics.
-+
-.Running the Kafka Bridge to enable metrics
-[source,shell]
-----
-./bin/kafka_bridge_run.sh --config-file=<path>/application.properties
-----
-+
-With metrics enabled, you can use `GET /metrics` with the `/metrics` endpoint to retrieve Kafka Bridge metrics in Prometheus format.
\ No newline at end of file
diff --git a/documentation/modules/proc-configuring-kafka-bridge-smr-metrics.adoc b/documentation/modules/proc-configuring-kafka-bridge-smr-metrics.adoc
new file mode 100644
index 0000000..ce49d32
--- /dev/null
+++ b/documentation/modules/proc-configuring-kafka-bridge-smr-metrics.adoc
@@ -0,0 +1,42 @@
+[id='proc-configuring-kafka-bridge-smr-metrics-{context}']
+= Configuring Strimzi Metrics Reporter metrics
+
+[role="_abstract"]
+Enable the Strimzi Metrics Reporter to collect Kafka Bridge metrics by setting the `bridge.metrics` option to `strimziMetricsReporter`.
+
+.Prerequisites
+
+* xref:proc-downloading-kafka-bridge-{context}[The Kafka Bridge installation archive is downloaded].
+
+.Procedure
+
+. Set the `bridge.metrics` configuration to `strimziMetricsReporter`.
++
+.Configuration for enabling metrics
+
+[source,properties]
+----
+bridge.metrics=strimziMetricsReporter
+----
++
+Optionally, you can configure a comma-separated list of regular expressions to filter exposed metrics using the `kafka.prometheus.metrics.reporter.allowlist` property.
+If not configured, a default set of metrics is exposed.
+
++
+When needed, it is possible to configure the `allowlist` per client type.
+For example, by using the `kafka.admin` prefix and setting `kafka.admin.prometheus.metrics.reporter.allowlist=`, all admin client metrics are excluded.
+
++
+You can add any plugin configuration to the Kafka Bridge properties file using `kafka.`, `kafka.admin.`, `kafka.producer.`, and `kafka.consumer.` prefixes.
+In the event that the same property is configured with multiple prefixes, the most specific prefix takes precedence.
+For example, `kafka.producer.prometheus.metrics.reporter.allowlist` takes precedence over `kafka.prometheus.metrics.reporter.allowlist`.
+
+. Run the Kafka Bridge run script.
++
+.Running the Kafka Bridge
+[source,shell]
+----
+./bin/kafka_bridge_run.sh --config-file=<path>/application.properties
+----
++
+With metrics enabled, you can scrape metrics in Prometheus format from the `/metrics` endpoint of the Kafka Bridge.
diff --git a/pom.xml b/pom.xml
index 0031a79..2d10029 100644
--- a/pom.xml
+++ b/pom.xml
@@ -115,6 +115,7 @@
 		<checkstyle.version>10.12.2</checkstyle.version>
 		<hamcrest.version>2.2</hamcrest.version>
 		<junit.version>5.8.2</junit.version>
+		<mockito.version>4.11.0</mockito.version>
 		<maven-jar-plugin.version>3.3.0</maven-jar-plugin.version>
 		<maven-surefire-plugin.version>3.0.0-M7</maven-surefire-plugin.version>
 		<maven-failsafe-plugin.version>3.0.0-M7</maven-failsafe-plugin.version>
@@ -131,6 +132,7 @@
 		<spotbugs.version>4.7.3</spotbugs.version>
 		<maven.spotbugs.version>4.7.3.0</maven.spotbugs.version>
 		<strimzi-oauth.version>0.15.0</strimzi-oauth.version>
+		<strimzi-metrics-reporter.version>0.1.0</strimzi-metrics-reporter.version>
 		<opentelemetry.version>1.34.1</opentelemetry.version>
 		<opentelemetry-alpha.version>1.34.1-alpha</opentelemetry-alpha.version>
 		<opentelemetry.instrumentation.version>1.32.0-alpha</opentelemetry.instrumentation.version>
@@ -295,6 +297,11 @@
 			<artifactId>micrometer-registry-prometheus</artifactId>
 			<version>${micrometer.version}</version>
 		</dependency>
+		<dependency>
+			<groupId>io.strimzi</groupId>
+			<artifactId>metrics-reporter</artifactId>
+			<version>${strimzi-metrics-reporter.version}</version>
+		</dependency>
 		<dependency>
 			<groupId>io.prometheus.jmx</groupId>
 			<artifactId>collector</artifactId>
@@ -305,6 +312,19 @@
 			<artifactId>prometheus-metrics-model</artifactId>
 			<version>${prometheus-client.version}</version>
 		</dependency>
+		<!-- We include explicitly prometheus-metrics-instrumentation-jvm 
+			and prometheus-metrics-exporter-httpserver to force the use of 
+			Bridge version over metrics-reporter version. -->
+		<dependency>
+			<groupId>io.prometheus</groupId>
+			<artifactId>prometheus-metrics-instrumentation-jvm</artifactId>
+			<version>${prometheus-client.version}</version>
+		</dependency>
+		<dependency>
+			<groupId>io.prometheus</groupId>
+			<artifactId>prometheus-metrics-exporter-httpserver</artifactId>
+			<version>${prometheus-client.version}</version>
+		</dependency>
 		<dependency>
 			<groupId>io.prometheus</groupId>
 			<artifactId>prometheus-metrics-exposition-textformats</artifactId>
@@ -421,6 +441,12 @@
 			<version>${hamcrest.version}</version>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.mockito</groupId>
+			<artifactId>mockito-core</artifactId>
+			<version>${mockito.version}</version>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>io.strimzi</groupId>
 			<artifactId>strimzi-test-container</artifactId>
@@ -584,6 +610,10 @@
 									<!-- OpenTelemetry - used via classpath configuration opentelemetry-exporter-sender-grpc is required at runtime to replace OKHTTP -->
 									<ignoredUnusedDeclaredDependency>io.grpc:grpc-netty-shaded:jar</ignoredUnusedDeclaredDependency>
 									<ignoredUnusedDeclaredDependency>com.google.guava:guava</ignoredUnusedDeclaredDependency>
+									<!-- Strimzi Metrics Reporter - used via classpath configuration -->
+									<ignoredUnusedDeclaredDependency>io.strimzi:metrics-reporter</ignoredUnusedDeclaredDependency>
+									<ignoredUnusedDeclaredDependency>io.prometheus:prometheus-metrics-instrumentation-jvm</ignoredUnusedDeclaredDependency>
+									<ignoredUnusedDeclaredDependency>io.prometheus:prometheus-metrics-exporter-httpserver</ignoredUnusedDeclaredDependency>
 								</ignoredUnusedDeclaredDependencies>
 						</configuration>
 					</execution>
diff --git a/src/main/java/io/strimzi/kafka/bridge/Application.java b/src/main/java/io/strimzi/kafka/bridge/Application.java
index 04a3f6e..edded0d 100644
--- a/src/main/java/io/strimzi/kafka/bridge/Application.java
+++ b/src/main/java/io/strimzi/kafka/bridge/Application.java
@@ -2,10 +2,8 @@
  * Copyright Strimzi authors.
  * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
  */
-
 package io.strimzi.kafka.bridge;
 
-import io.micrometer.core.instrument.MeterRegistry;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.ConfigRetriever;
 import io.strimzi.kafka.bridge.http.HttpBridge;
@@ -18,7 +16,6 @@ import io.vertx.micrometer.Label;
 import io.vertx.micrometer.MetricsDomain;
 import io.vertx.micrometer.MicrometerMetricsOptions;
 import io.vertx.micrometer.VertxPrometheusOptions;
-import io.vertx.micrometer.backends.BackendRegistries;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.DefaultParser;
 import org.apache.commons.cli.Option;
@@ -28,17 +25,11 @@ import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
 import javax.management.MalformedObjectNameException;
-import java.io.BufferedReader;
 import java.io.File;
 import java.io.IOException;
-import java.io.InputStream;
-import java.io.InputStreamReader;
-import java.nio.charset.StandardCharsets;
 import java.util.EnumSet;
-import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 /**
  * Apache Kafka bridge main application class
@@ -46,8 +37,6 @@ import java.util.stream.Collectors;
 public class Application {
     private static final Logger LOGGER = LogManager.getLogger(Application.class);
 
-    private static final String KAFKA_BRIDGE_METRICS_ENABLED = "KAFKA_BRIDGE_METRICS_ENABLED";
-
     /**
      * Bridge entrypoint
      *
@@ -56,27 +45,12 @@ public class Application {
     public static void main(String[] args) {
         LOGGER.info("Strimzi Kafka Bridge {} is starting", Application.class.getPackage().getImplementationVersion());
         try {
-            VertxOptions vertxOptions = new VertxOptions();
-            JmxCollectorRegistry jmxCollectorRegistry = null;
-            if (Boolean.parseBoolean(System.getenv(KAFKA_BRIDGE_METRICS_ENABLED))) {
-                LOGGER.info("Metrics enabled and exposed on the /metrics endpoint");
-                // setup Micrometer metrics options
-                vertxOptions.setMetricsOptions(metricsOptions());
-                jmxCollectorRegistry = getJmxCollectorRegistry();
-            }
-            Vertx vertx = Vertx.vertx(vertxOptions);
-            // MeterRegistry default instance is just null if metrics are not enabled in the VertxOptions instance
-            MeterRegistry meterRegistry = BackendRegistries.getDefaultNow();
-            MetricsReporter metricsReporter = new MetricsReporter(jmxCollectorRegistry, meterRegistry);
-
-
             CommandLine commandLine = new DefaultParser().parse(generateOptions(), args);
-
             Map<String, Object> config = ConfigRetriever.getConfig(absoluteFilePath(commandLine.getOptionValue("config-file")));
             BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
             LOGGER.info("Bridge configuration {}", bridgeConfig);
 
-            deployHttpBridge(vertx, bridgeConfig, metricsReporter).onComplete(done -> {
+            deployHttpBridge(bridgeConfig).onComplete(done -> {
                 if (done.succeeded()) {
                     // register tracing - if set, etc
                     TracingUtil.initialize(bridgeConfig);
@@ -88,37 +62,19 @@ public class Application {
         }
     }
 
-    /**
-     * Set up the Vert.x metrics options
-     *
-     * @return instance of the MicrometerMetricsOptions on Vert.x
-     */
-    private static MicrometerMetricsOptions metricsOptions() {
-        Set<String> set = new HashSet<>();
-        set.add(MetricsDomain.NAMED_POOLS.name());
-        set.add(MetricsDomain.VERTICLES.name());
-        return new MicrometerMetricsOptions()
-                .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
-                // define the labels on the HTTP server related metrics
-                .setLabels(EnumSet.of(Label.HTTP_PATH, Label.HTTP_METHOD, Label.HTTP_CODE))
-                // disable metrics about pool and verticles
-                .setDisabledMetricsCategories(set)
-                .setJvmMetricsEnabled(true)
-                .setEnabled(true);
-    }
-
     /**
      * Deploys the HTTP bridge into a new verticle
      *
-     * @param vertx                 Vertx instance
      * @param bridgeConfig          Bridge configuration
-     * @param metricsReporter       MetricsReporter instance for scraping metrics from different registries
      * @return                      Future for the bridge startup
      */
-    private static Future<HttpBridge> deployHttpBridge(Vertx vertx, BridgeConfig bridgeConfig, MetricsReporter metricsReporter)  {
+    private static Future<HttpBridge> deployHttpBridge(BridgeConfig bridgeConfig) 
+            throws MalformedObjectNameException, IOException {
         Promise<HttpBridge> httpPromise = Promise.promise();
 
-        HttpBridge httpBridge = new HttpBridge(bridgeConfig, metricsReporter);
+        Vertx vertx = createVertxInstance(bridgeConfig);
+        HttpBridge httpBridge = new HttpBridge(bridgeConfig);
+
         vertx.deployVerticle(httpBridge)
                 .onComplete(done -> {
                     if (done.succeeded()) {
@@ -133,26 +89,30 @@ public class Application {
         return httpPromise.future();
     }
 
+    private static Vertx createVertxInstance(BridgeConfig bridgeConfig) {
+        VertxOptions vertxOptions = new VertxOptions();
+        if (bridgeConfig.getMetricsType() != null) {
+            vertxOptions.setMetricsOptions(metricsOptions()); // enable Vertx metrics
+        }
+        Vertx vertx = Vertx.vertx(vertxOptions);
+        return vertx;
+    }
+
     /**
-     * Return a JmxCollectorRegistry instance with the YAML configuration filters
+     * Set up the Vert.x metrics options
      *
-     * @return JmxCollectorRegistry instance
-     * @throws MalformedObjectNameException
-     * @throws IOException
+     * @return instance of the MicrometerMetricsOptions on Vert.x
      */
-    private static JmxCollectorRegistry getJmxCollectorRegistry()
-            throws MalformedObjectNameException, IOException {
-        InputStream is = Application.class.getClassLoader().getResourceAsStream("jmx_metrics_config.yaml");
-        if (is == null) {
-            return null;
-        }
-
-        try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
-            String yaml = reader
-                    .lines()
-                    .collect(Collectors.joining("\n"));
-            return new JmxCollectorRegistry(yaml);
-        }
+    private static MicrometerMetricsOptions metricsOptions() {
+        return new MicrometerMetricsOptions()
+            .setPrometheusOptions(new VertxPrometheusOptions().setEnabled(true))
+            // define the labels on the HTTP server related metrics
+            .setLabels(EnumSet.of(Label.HTTP_PATH, Label.HTTP_METHOD, Label.HTTP_CODE))
+            // disable metrics about pool and verticles
+            .setDisabledMetricsCategories(
+                Set.of(MetricsDomain.NAMED_POOLS.name(), MetricsDomain.VERTICLES.name())
+            ).setJvmMetricsEnabled(true)
+            .setEnabled(true);
     }
 
     /**
@@ -161,7 +121,6 @@ public class Application {
      * @return command line options
      */
     private static Options generateOptions() {
-
         Option configFileOption = Option.builder()
                 .required(true)
                 .hasArg(true)
diff --git a/src/main/java/io/strimzi/kafka/bridge/JmxCollectorRegistry.java b/src/main/java/io/strimzi/kafka/bridge/JmxCollectorRegistry.java
deleted file mode 100644
index 27107d3..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/JmxCollectorRegistry.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
- * Copyright Strimzi authors.
- * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
- */
-
-package io.strimzi.kafka.bridge;
-
-import io.prometheus.jmx.JmxCollector;
-import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter;
-import io.prometheus.metrics.model.registry.PrometheusRegistry;
-
-import javax.management.MalformedObjectNameException;
-import java.io.ByteArrayOutputStream;
-import java.io.File;
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-
-/**
- * Allow to collect JMX metrics exposing them in the Prometheus format
- */
-public class JmxCollectorRegistry {
-    private final PrometheusRegistry collectorRegistry;
-    private final PrometheusTextFormatWriter textFormatter = new PrometheusTextFormatWriter(true);
-
-    /**
-     * Constructor
-     *
-     * @param yamlConfig YAML configuration string with metrics filtering rules
-     * @throws MalformedObjectNameException Throws MalformedObjectNameException
-     */
-    public JmxCollectorRegistry(String yamlConfig) throws MalformedObjectNameException {
-        new JmxCollector(yamlConfig).register();
-        collectorRegistry = PrometheusRegistry.defaultRegistry;
-    }
-
-    /**
-     * Constructor
-     *
-     * @param yamlFileConfig file containing the YAML configuration with metrics filtering rules
-     * @throws MalformedObjectNameException Throws MalformedObjectNameException
-     * @throws IOException Throws IOException
-     */
-    public JmxCollectorRegistry(File yamlFileConfig) throws MalformedObjectNameException, IOException {
-        new JmxCollector(yamlFileConfig).register();
-        collectorRegistry = PrometheusRegistry.defaultRegistry;
-    }
-
-    /**
-     * @return Content that should be included in the response body for an endpoint designated for
-     * Prometheus to scrape from.
-     */
-    public String scrape() {
-        ByteArrayOutputStream stream = new ByteArrayOutputStream();
-        try {
-            textFormatter.write(stream, collectorRegistry.scrape());
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-        return stream.toString(StandardCharsets.UTF_8);
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/MetricsReporter.java b/src/main/java/io/strimzi/kafka/bridge/MetricsReporter.java
deleted file mode 100644
index ab8e2f3..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/MetricsReporter.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/*
- * Copyright Strimzi authors.
- * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
- */
-
-package io.strimzi.kafka.bridge;
-
-import io.micrometer.core.instrument.Meter;
-import io.micrometer.core.instrument.MeterRegistry;
-import io.micrometer.prometheus.PrometheusMeterRegistry;
-import io.micrometer.prometheus.PrometheusNamingConvention;
-
-/**
- * Used for scraping and reporting metrics in Prometheus format
- */
-public class MetricsReporter {
-
-    private final JmxCollectorRegistry jmxCollectorRegistry;
-    private final MeterRegistry meterRegistry;
-
-    /**
-     * Constructor
-     *
-     * @param jmxCollectorRegistry JmxCollectorRegistry instance for scraping metrics from JMX endpoints
-     * @param meterRegistry MeterRegistry instance for scraping metrics exposed through Vert.x
-     */
-    public MetricsReporter(JmxCollectorRegistry jmxCollectorRegistry, MeterRegistry meterRegistry) {
-        this.jmxCollectorRegistry = jmxCollectorRegistry;
-        this.meterRegistry = meterRegistry;
-        if (this.meterRegistry instanceof PrometheusMeterRegistry) {
-            this.meterRegistry.config().namingConvention(new MetricsNamingConvention());
-        }
-    }
-
-    private static class MetricsNamingConvention extends PrometheusNamingConvention {
-        @Override
-        public String name(String name, Meter.Type type, String baseUnit) {
-            String metricName = name.startsWith("vertx.") ? name.replace("vertx.", "strimzi.bridge.") : name;
-            return super.name(metricName, type, baseUnit);
-        }
-    }
-
-    /**
-     * @return JmxCollectorRegistry instance for scraping metrics from JMX endpoints
-     */
-    public JmxCollectorRegistry getJmxCollectorRegistry() {
-        return jmxCollectorRegistry;
-    }
-
-    /**
-     * @return MeterRegistry instance for scraping metrics exposed through Vert.x
-     */
-    public MeterRegistry getMeterRegistry() {
-        return meterRegistry;
-    }
-
-    /**
-     * Scrape metrics on the provided registries returning them in the Prometheus format
-     *
-     * @return metrics in Prometheus format as String
-     */
-    public String scrape() {
-        StringBuilder sb = new StringBuilder();
-        if (jmxCollectorRegistry != null) {
-            sb.append(jmxCollectorRegistry.scrape());
-        }
-        if (meterRegistry instanceof PrometheusMeterRegistry) {
-            sb.append(((PrometheusMeterRegistry) meterRegistry).scrape());
-        }
-        return sb.toString();
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
index 919c7f9..0ff6622 100644
--- a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
+++ b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
@@ -6,14 +6,20 @@
 package io.strimzi.kafka.bridge.config;
 
 import io.strimzi.kafka.bridge.http.HttpConfig;
+import io.strimzi.kafka.bridge.metrics.MetricsType;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
+import java.nio.file.Path;
 import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 
 /**
  * Bridge configuration properties
  */
 public class BridgeConfig extends AbstractConfig {
+    private static final Logger LOGGER = LogManager.getLogger(BridgeConfig.class);
 
     /** Prefix for all the specific bridge configuration parameters */
     public static final String BRIDGE_CONFIG_PREFIX = "bridge.";
@@ -21,8 +27,18 @@ public class BridgeConfig extends AbstractConfig {
     /** Bridge identification number */
     public static final String BRIDGE_ID = BRIDGE_CONFIG_PREFIX + "id";
 
+    /** Metrics system to be used in the bridge */
+    public static final String METRICS_TYPE = BRIDGE_CONFIG_PREFIX + "metrics";
+
+    /** JMX Exporter configuration file path */
+    private static final String JMX_EXPORTER_CONFIG_PATH = METRICS_TYPE + ".exporter.config.path";
+    
     /** Tracing system to be used in the bridge */
-    public static final String TRACING_TYPE = BRIDGE_CONFIG_PREFIX + "tracing";
+    private static final String TRACING_TYPE = BRIDGE_CONFIG_PREFIX + "tracing";
+    
+    /** Default Strimzi Metrics Reporter allow list. */
+    /* test */ static final String DEFAULT_STRIMZI_METRICS_REPORTER_ALLOW_LIST = "kafka_consumer_consumer_metrics.*, " +
+        "kafka_producer_kafka_metrics_count_count, kafka_producer_producer_metrics.*";
 
     private final KafkaConfig kafkaConfig;
     private final HttpConfig httpConfig;
@@ -55,21 +71,36 @@ public class BridgeConfig extends AbstractConfig {
     }
 
     /**
-     * Loads the entire bridge configuration parameters from a related map
+     * Loads the entire bridge configuration parameters from a related map.
+     * Some validation and default values are also applied.
      *
      * @param map map from which loading configuration parameters
      * @return overall bridge configuration
      */
     public static BridgeConfig fromMap(Map<String, Object> map) {
+        validateAndApplyDefaults(map);
         KafkaConfig kafkaConfig = KafkaConfig.fromMap(map);
         HttpConfig httpConfig = HttpConfig.fromMap(map);
-
         return new BridgeConfig(map.entrySet().stream()
                 .filter(e -> e.getKey().startsWith(BridgeConfig.BRIDGE_CONFIG_PREFIX))
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)),
                 kafkaConfig, httpConfig);
     }
 
+    private static void validateAndApplyDefaults(Map<String, Object> map) {
+        if (map.get(METRICS_TYPE) != null) {
+            // get and validate metrics type
+            MetricsType metricsType = MetricsType.fromString((String) map.get(METRICS_TYPE));
+            // apply default Strimzi Metrics Reporter configuration if not present
+            if (metricsType == MetricsType.STRIMZI_REPORTER) {
+                LOGGER.info("Using default metrics configuration");
+                map.putIfAbsent("kafka.metric.reporters", "io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter");
+                map.putIfAbsent("kafka.prometheus.metrics.reporter.listener.enable", "false");
+                map.putIfAbsent("kafka.prometheus.metrics.reporter.allowlist", DEFAULT_STRIMZI_METRICS_REPORTER_ALLOW_LIST);
+            }
+        }
+    }
+
     @Override
     public String toString() {
         return "BridgeConfig(" +
@@ -90,6 +121,31 @@ public class BridgeConfig extends AbstractConfig {
         }
     }
 
+    /**
+     * @return the metric type to be used in the bridge
+     */
+    public MetricsType getMetricsType() {
+        final String envVarValue = System.getenv("KAFKA_BRIDGE_METRICS_ENABLED");
+        if (envVarValue != null) {
+            LOGGER.warn("KAFKA_BRIDGE_METRICS_ENABLED is deprecated, use bridge.metrics configuration");
+        }
+        
+        return Optional.ofNullable((String) config.get(BridgeConfig.METRICS_TYPE))
+            .map(MetricsType::fromString)
+            .orElseGet(() -> Boolean.parseBoolean(envVarValue) ? MetricsType.JMX_EXPORTER : null);
+    }
+
+    /**
+     * @return the Prometheus JMX Exporter configuration file path
+     */
+    public Path getJmxExporterConfigPath() {
+        if (config.get(BridgeConfig.JMX_EXPORTER_CONFIG_PATH) == null) {
+            return null;
+        } else {
+            return Path.of((String) config.get(BridgeConfig.JMX_EXPORTER_CONFIG_PATH));
+        }
+    }
+
     /**
      * @return the tracing system to be used in the bridge
      */
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
index c05d439..af5e6ca 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -2,7 +2,6 @@
  * Copyright Strimzi authors.
  * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
  */
-
 package io.strimzi.kafka.bridge.http;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -15,10 +14,13 @@ import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.ConsumerInstanceId;
 import io.strimzi.kafka.bridge.EmbeddedFormat;
 import io.strimzi.kafka.bridge.IllegalEmbeddedFormatException;
-import io.strimzi.kafka.bridge.MetricsReporter;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.http.converter.JsonUtils;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
+import io.strimzi.kafka.bridge.metrics.JmxMetricsCollector;
+import io.strimzi.kafka.bridge.metrics.MetricsCollector;
+import io.strimzi.kafka.bridge.metrics.MetricsType;
+import io.strimzi.kafka.bridge.metrics.StrimziMetricsCollector;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Promise;
 import io.vertx.core.file.FileSystem;
@@ -43,11 +45,19 @@ import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import javax.management.MalformedObjectNameException;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static io.netty.handler.codec.http.HttpHeaderNames.ACCEPT;
 import static io.netty.handler.codec.http.HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS;
@@ -59,7 +69,7 @@ import static io.netty.handler.codec.http.HttpHeaderNames.ORIGIN;
 /**
  * Main bridge class listening for connections and handling HTTP requests.
  */
-@SuppressWarnings({"checkstyle:MemberName"})
+@SuppressWarnings({"checkstyle:MemberName", "checkstyle:ClassDataAbstractionCoupling", "checkstyle:ClassFanOutComplexity"})
 public class HttpBridge extends AbstractVerticle {
     private static final Logger LOGGER = LogManager.getLogger(HttpBridge.class);
 
@@ -76,17 +86,59 @@ public class HttpBridge extends AbstractVerticle {
 
     private final Map<ConsumerInstanceId, Long> timestampMap = new HashMap<>();
 
-    private final MetricsReporter metricsReporter;
+    private MetricsCollector metricsCollector = null;
 
     /**
      * Constructor
      *
      * @param bridgeConfig bridge configuration
-     * @param metricsReporter MetricsReporter instance for scraping metrics from different registries
      */
-    public HttpBridge(BridgeConfig bridgeConfig, MetricsReporter metricsReporter) {
+    public HttpBridge(BridgeConfig bridgeConfig) {
         this.bridgeConfig = bridgeConfig;
-        this.metricsReporter = metricsReporter;
+        if (bridgeConfig.getMetricsType() != null) {
+            if (bridgeConfig.getMetricsType() == MetricsType.JMX_EXPORTER) {
+                this.metricsCollector = createJmxMetricsCollector(bridgeConfig);
+            } else if (bridgeConfig.getMetricsType() == MetricsType.STRIMZI_REPORTER) {
+                this.metricsCollector = new StrimziMetricsCollector();
+            }
+        }
+        if (metricsCollector != null) {
+            LOGGER.info("Metrics of type '{}' enabled and exposed on /metrics endpoint", bridgeConfig.getMetricsType());
+        }
+    }
+
+    /**
+     * Create a JmxMetricsCollector instance with the YAML configuration filters.
+     * This is loaded from a custom config file if present or from the default configuration file.
+     *
+     * @return JmxCollectorRegistry instance
+     */
+    private static JmxMetricsCollector createJmxMetricsCollector(BridgeConfig bridgeConfig) {
+        try {
+            if (bridgeConfig.getJmxExporterConfigPath() == null) {
+                // load default configuration
+                LOGGER.info("Using default JMX metrics configuration");
+                InputStream is = Application.class.getClassLoader().getResourceAsStream("jmx_metrics_config.yaml");
+                if (is == null) {
+                    return null;
+                }
+                try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
+                    String yaml = reader
+                        .lines()
+                        .collect(Collectors.joining("\n"));
+                    return new JmxMetricsCollector(yaml);
+                }
+            } else if (Files.exists(bridgeConfig.getJmxExporterConfigPath())) {
+                // load custom configuration file
+                LOGGER.info("Loading custom JMX metrics configuration file from {}", bridgeConfig.getJmxExporterConfigPath());
+                String yaml = Files.readString(bridgeConfig.getJmxExporterConfigPath(), StandardCharsets.UTF_8);
+                return new JmxMetricsCollector(yaml);
+            } else {
+                throw new RuntimeException("Custom JMX metrics configuration file not found");
+            }
+        } catch (IOException | MalformedObjectNameException e) {
+            throw new RuntimeException("Failed to initialize JMX metrics collector", e);
+        }
     }
 
     private void bindHttpServer(Promise<Void> startPromise) {
@@ -140,7 +192,6 @@ public class HttpBridge extends AbstractVerticle {
 
     @Override
     public void start(Promise<Void> startPromise) {
-
         RouterBuilder.create(vertx, "openapi.json")
                 .onComplete(ar -> {
                     if (ar.succeeded()) {
@@ -184,9 +235,9 @@ public class HttpBridge extends AbstractVerticle {
                         this.router.errorHandler(HttpResponseStatus.BAD_REQUEST.code(), this::errorHandler);
                         this.router.errorHandler(HttpResponseStatus.NOT_FOUND.code(), this::errorHandler);
 
-                        if (this.metricsReporter.getMeterRegistry() != null) {
+                        if (this.metricsCollector != null && this.metricsCollector.getVertxRegistry() != null) {
                             // exclude to report the HTTP server metrics for the /metrics endpoint itself
-                            this.metricsReporter.getMeterRegistry().config().meterFilter(
+                            this.metricsCollector.getVertxRegistry().config().meterFilter(
                                     MeterFilter.deny(meter -> "/metrics".equals(meter.getTag(Label.HTTP_PATH.toString())))
                             );
                         }
@@ -236,7 +287,6 @@ public class HttpBridge extends AbstractVerticle {
 
     @Override
     public void stop(Promise<Void> stopPromise) {
-
         LOGGER.info("Stopping HTTP-Kafka bridge verticle ...");
 
         this.isReady = false;
@@ -558,10 +608,14 @@ public class HttpBridge extends AbstractVerticle {
     }
 
     private void metrics(RoutingContext routingContext) {
-        routingContext.response()
+        if (metricsCollector != null) {
+            routingContext.response()
                 .putHeader("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
                 .setStatusCode(HttpResponseStatus.OK.code())
-                .end(metricsReporter.scrape());
+                .end(metricsCollector.scrape());
+        } else {
+            HttpUtils.sendResponse(routingContext, HttpResponseStatus.NOT_FOUND.code(), null, null);
+        }
     }
 
     private void information(RoutingContext routingContext) {
diff --git a/src/main/java/io/strimzi/kafka/bridge/metrics/JmxMetricsCollector.java b/src/main/java/io/strimzi/kafka/bridge/metrics/JmxMetricsCollector.java
new file mode 100644
index 0000000..30eff78
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/metrics/JmxMetricsCollector.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+
+package io.strimzi.kafka.bridge.metrics;
+
+import io.prometheus.jmx.JmxCollector;
+import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+
+import javax.management.MalformedObjectNameException;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Collect and scrape JMX metrics in Prometheus format.
+ */
+public class JmxMetricsCollector extends MetricsCollector {
+    private final PrometheusRegistry registry;
+    private final PrometheusTextFormatWriter textFormatter;
+
+    /**
+     * Constructor.
+     *
+     * @param yamlConfig YAML configuration string with metrics filtering rules
+     * @throws MalformedObjectNameException Throws MalformedObjectNameException
+     */
+    public JmxMetricsCollector(String yamlConfig) throws MalformedObjectNameException {
+        // Prometheus default registry is a singleton, so it is shared with JmxCollector
+        this(new JmxCollector(yamlConfig), PrometheusRegistry.defaultRegistry, new PrometheusTextFormatWriter(true));
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param jmxCollector JMX collector registry
+     * @param registry Prometheus collector registry
+     * @param textFormatter Prometheus text formatter
+     */
+    /* test */ JmxMetricsCollector(JmxCollector jmxCollector,
+                                   PrometheusRegistry registry,
+                                   PrometheusTextFormatWriter textFormatter) {
+        super();
+        jmxCollector.register();
+        this.registry = registry;
+        this.textFormatter = textFormatter;
+    }
+
+    @Override
+    public String doScrape() {
+        ByteArrayOutputStream stream = new ByteArrayOutputStream();
+        try {
+            textFormatter.write(stream, registry.scrape());
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        return stream.toString(StandardCharsets.UTF_8);
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/metrics/MetricsCollector.java b/src/main/java/io/strimzi/kafka/bridge/metrics/MetricsCollector.java
new file mode 100644
index 0000000..1ab5c40
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/metrics/MetricsCollector.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.bridge.metrics;
+
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
+import io.micrometer.prometheus.PrometheusNamingConvention;
+import io.vertx.micrometer.backends.BackendRegistries;
+
+/**
+ * Abstract class for collecting and exposing metrics.
+ */
+public abstract class MetricsCollector {
+    private final PrometheusMeterRegistry vertxRegistry;
+    
+    MetricsCollector() {
+        this.vertxRegistry = (PrometheusMeterRegistry) BackendRegistries.getDefaultNow();
+        if (vertxRegistry != null) {
+            // replace the default Prometheus naming convention
+            this.vertxRegistry.config().namingConvention(new MetricsNamingConvention());
+        }
+    }
+
+    private static class MetricsNamingConvention extends PrometheusNamingConvention {
+        @Override
+        public String name(String name, Meter.Type type, String baseUnit) {
+            String metricName = name.startsWith("vertx.") ? name.replace("vertx.", "strimzi.bridge.") : name;
+            return super.name(metricName, type, baseUnit);
+        }
+    }
+
+    /**
+     * @return Registry instance for scraping Vertx metrics.
+     *         This is null if metrics are not enabled in the VertxOptions instance.
+     */
+    public MeterRegistry getVertxRegistry() {
+        return vertxRegistry;
+    }
+
+    /**
+     * Scrape all, including Vertx metrics.
+     * 
+     * @return Raw metrics in Prometheus format.
+     */
+    public String scrape() {
+        StringBuilder sb = new StringBuilder();
+        sb.append(doScrape());
+        if (vertxRegistry != null) {
+            sb.append(vertxRegistry.scrape());
+        }
+        return sb.toString();
+    }
+
+    /**
+     * @return Raw metrics in Prometheus format.
+     */
+    abstract String doScrape();
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/metrics/MetricsType.java b/src/main/java/io/strimzi/kafka/bridge/metrics/MetricsType.java
new file mode 100644
index 0000000..53d8219
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/metrics/MetricsType.java
@@ -0,0 +1,40 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.bridge.metrics;
+
+/**
+ * Metrics type.
+ */
+public enum MetricsType {
+    /** Prometheus JMX Exporter. */
+    JMX_EXPORTER("jmxPrometheusExporter"),
+    
+    /** Strimzi Metrics Reporter. */
+    STRIMZI_REPORTER("strimziMetricsReporter");
+
+    private final String text;
+    
+    MetricsType(final String text) {
+        this.text = text;
+    }
+    
+    @Override
+    public String toString() {
+        return text;
+    }
+
+    /**
+     * @param text Text.
+     * @return Get type from text.
+     */
+    public static MetricsType fromString(String text) {
+        for (MetricsType t : MetricsType.values()) {
+            if (t.text.equalsIgnoreCase(text)) {
+                return t;
+            }
+        }
+        throw new IllegalArgumentException("Metrics type not found: " + text);
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/metrics/StrimziMetricsCollector.java b/src/main/java/io/strimzi/kafka/bridge/metrics/StrimziMetricsCollector.java
new file mode 100644
index 0000000..97d9fb9
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/metrics/StrimziMetricsCollector.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+
+package io.strimzi.kafka.bridge.metrics;
+
+import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Collect and scrape Strimzi Reporter metrics in Prometheus format.
+ */
+public class StrimziMetricsCollector extends MetricsCollector {
+    private final PrometheusRegistry registry;
+    private final PrometheusTextFormatWriter textFormatter;
+
+    /**
+     * Constructor.
+     */
+    public StrimziMetricsCollector() {
+        // Prometheus default registry is a singleton, so it is shared with Strimzi Metrics Reporter
+        this(PrometheusRegistry.defaultRegistry, new PrometheusTextFormatWriter(true));
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param registry Prometheus collector registry
+     * @param textFormatter Prometheus text formatter
+     */
+    /* test */ StrimziMetricsCollector(PrometheusRegistry registry,
+                                       PrometheusTextFormatWriter textFormatter) {
+        super();
+        this.registry = registry;
+        this.textFormatter = textFormatter;
+    }
+
+    @Override
+    public String doScrape() {
+        ByteArrayOutputStream stream = new ByteArrayOutputStream();
+        try {
+            textFormatter.write(stream, registry.scrape());
+        } catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+        return stream.toString(StandardCharsets.UTF_8);
+    }
+}
diff --git a/src/main/resources/openapi.json b/src/main/resources/openapi.json
index dc51164..c342353 100644
--- a/src/main/resources/openapi.json
+++ b/src/main/resources/openapi.json
@@ -1478,6 +1478,9 @@
                             }
                         },
                         "description": "Metrics in Prometheus format retrieved successfully."
+                    },
+                    "404": {
+                        "description": "The metrics endpoint is not enabled."
                     }
                 },
                 "operationId": "metrics",


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java b/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java
index a237e2f..fe7e407 100644
--- a/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java
@@ -2,25 +2,28 @@
  * Copyright Strimzi authors.
  * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
  */
-
 package io.strimzi.kafka.bridge.config;
 
+import io.strimzi.kafka.bridge.metrics.MetricsType;
 import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.junit.jupiter.api.Test;
 
+import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.not;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
  * Some config related classes unit tests
  */
 public class ConfigTest {
-
     @Test
     public void testConfig() {
         Map<String, Object> map = new HashMap<>();
@@ -64,8 +67,8 @@ public class ConfigTest {
         BridgeConfig bridgeConfig = BridgeConfig.fromMap(map);
         assertThat(bridgeConfig.getKafkaConfig().getConfig().size(), is(6));
 
-        assertThat(bridgeConfig.getKafkaConfig().toString().contains("ssl.truststore.password=" + storePassword), is(false));
-        assertThat(bridgeConfig.getKafkaConfig().toString().contains("ssl.truststore.password=[hidden]"), is(true));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), not(containsString("ssl.truststore.password=" + storePassword)));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("ssl.truststore.password=[hidden]"));
     }
 
     @Test
@@ -81,4 +84,72 @@ public class ConfigTest {
         assertThat(bridgeConfig.getHttpConfig().isConsumerEnabled(), is(true));
         assertThat(bridgeConfig.getHttpConfig().isProducerEnabled(), is(true));
     }
+
+    @Test
+    public void testJmxExporterMetricsType() {
+        String configFilePah = "/tmp/my-exporter-config.yaml";
+        
+        Map<String, Object> map = Map.of(
+            "bridge.id", "my-bridge",
+            "kafka.bootstrap.servers", "localhost:9092",
+            "bridge.metrics", "jmxPrometheusExporter",
+            "bridge.metrics.exporter.config.path", configFilePah,
+            "http.host", "0.0.0.0",
+            "http.port", "8080"
+        );
+
+        BridgeConfig bridgeConfig = BridgeConfig.fromMap(map);
+        assertThat(bridgeConfig.getMetricsType(), is(MetricsType.JMX_EXPORTER));
+        assertThat(bridgeConfig.getJmxExporterConfigPath(), is(Path.of(configFilePah)));
+    }
+
+    @Test
+    public void testStrimziReporterMetricsType() {
+        Map<String, Object> map = new HashMap<>(Map.of(
+            "bridge.id", "my-bridge",
+            "kafka.bootstrap.servers", "localhost:9092",
+            "bridge.metrics", "strimziMetricsReporter",
+            "http.host", "0.0.0.0",
+            "http.port", "8080"
+        ));
+
+        BridgeConfig bridgeConfig = BridgeConfig.fromMap(map);
+        assertThat(bridgeConfig.getMetricsType(), is(MetricsType.STRIMZI_REPORTER));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("metric.reporters=io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter"));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("prometheus.metrics.reporter.listener.enable=false"));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("prometheus.metrics.reporter.allowlist=" + BridgeConfig.DEFAULT_STRIMZI_METRICS_REPORTER_ALLOW_LIST));
+    }
+
+    @Test
+    public void testStrimziReporterWithCustomConfig() {
+        Map<String, Object> map = new HashMap<>(Map.of(
+            "bridge.id", "my-bridge",
+            "kafka.bootstrap.servers", "localhost:9092",
+            "bridge.metrics", "strimziMetricsReporter",
+            "kafka.metric.reporters", "my.domain.CustomMetricReporter,io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter",
+            "kafka.prometheus.metrics.reporter.allowlist", "kafka_log.*,kafka_network.*",
+            "http.host", "0.0.0.0",
+            "http.port", "8080"
+        ));
+
+        BridgeConfig bridgeConfig = BridgeConfig.fromMap(map);
+        assertThat(bridgeConfig.getMetricsType(), is(MetricsType.STRIMZI_REPORTER));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("metric.reporters=my.domain.CustomMetricReporter,io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter"));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("prometheus.metrics.reporter.listener.enable=false"));
+        assertThat(bridgeConfig.getKafkaConfig().toString(), containsString("prometheus.metrics.reporter.allowlist=kafka_log.*,kafka_network.*"));
+    }
+
+    @Test
+    public void testInvalidMetricsType() {
+        Map<String, Object> map = Map.of(
+            "bridge.id", "my-bridge",
+            "kafka.bootstrap.servers", "localhost:9092",
+            "bridge.metrics", "invalidReporterType",
+            "http.host", "0.0.0.0",
+            "http.port", "8080"
+        );
+
+        Exception e = assertThrows(IllegalArgumentException.class, () -> BridgeConfig.fromMap(map));
+        assertThat(e.getMessage(), is("Metrics type not found: invalidReporterType"));
+    }
 }
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java
index 2cdd376..377f7c3 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/ConsumerGeneratedNameIT.java
@@ -4,10 +4,7 @@
  */
 package io.strimzi.kafka.bridge.http;
 
-import io.micrometer.core.instrument.MeterRegistry;
 import io.netty.handler.codec.http.HttpResponseStatus;
-import io.strimzi.kafka.bridge.JmxCollectorRegistry;
-import io.strimzi.kafka.bridge.MetricsReporter;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
@@ -79,8 +76,6 @@ public class ConsumerGeneratedNameIT {
     private static HttpBridge httpBridge;
     private static WebClient client;
     private static BridgeConfig bridgeConfig;
-    private static MeterRegistry meterRegistry = null;
-    private static JmxCollectorRegistry jmxCollectorRegistry = null;
 
     ConsumerService consumerService() {
         return ConsumerService.getInstance(client);
@@ -98,7 +93,7 @@ public class ConsumerGeneratedNameIT {
         if ("FALSE".equals(BRIDGE_EXTERNAL_ENV)) {
 
             bridgeConfig = BridgeConfig.fromMap(config);
-            httpBridge = new HttpBridge(bridgeConfig, new MetricsReporter(jmxCollectorRegistry, meterRegistry));
+            httpBridge = new HttpBridge(bridgeConfig);
 
             LOGGER.info("Deploying in-memory bridge");
             vertx.deployVerticle(httpBridge).onComplete(context.succeeding(id -> context.completeNow()));
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java
index 2cc3079..e6f2815 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/DisablingConsumerProducerIT.java
@@ -7,7 +7,6 @@ package io.strimzi.kafka.bridge.http;
 
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.BridgeContentType;
-import io.strimzi.kafka.bridge.MetricsReporter;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
@@ -166,7 +165,7 @@ public class DisablingConsumerProducerIT {
         CompletableFuture<Boolean> startBridge = new CompletableFuture<>();
         if ("FALSE".equals(BRIDGE_EXTERNAL_ENV)) {
             BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
-            HttpBridge httpBridge = new HttpBridge(bridgeConfig, new MetricsReporter(null, null));
+            HttpBridge httpBridge = new HttpBridge(bridgeConfig);
 
             LOGGER.info("Deploying in-memory bridge");
             vertx.deployVerticle(httpBridge).onComplete(context.succeeding(id -> startBridge.complete(true)));
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
index 36f0c83..b190580 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpCorsIT.java
@@ -5,11 +5,8 @@
 
 package io.strimzi.kafka.bridge.http;
 
-import io.micrometer.core.instrument.MeterRegistry;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.BridgeContentType;
-import io.strimzi.kafka.bridge.JmxCollectorRegistry;
-import io.strimzi.kafka.bridge.MetricsReporter;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.facades.AdminClientFacade;
@@ -58,8 +55,6 @@ public class HttpCorsIT {
 
     static BridgeConfig bridgeConfig;
     static StrimziKafkaContainer kafkaContainer;
-    static MeterRegistry meterRegistry = null;
-    static JmxCollectorRegistry jmxCollectorRegistry = null;
     static AdminClientFacade adminClientFacade;
 
     static {
@@ -298,7 +293,7 @@ public class HttpCorsIT {
             config.put(HttpConfig.HTTP_CORS_ALLOWED_METHODS, methodsAllowed != null ? methodsAllowed : "GET,POST,PUT,DELETE,OPTIONS,PATCH");
 
             bridgeConfig = BridgeConfig.fromMap(config);
-            httpBridge = new HttpBridge(bridgeConfig, new MetricsReporter(jmxCollectorRegistry, meterRegistry));
+            httpBridge = new HttpBridge(bridgeConfig);
         }
     }
 }
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/InvalidProducerIT.java b/src/test/java/io/strimzi/kafka/bridge/http/InvalidProducerIT.java
index 6f82392..25b48e5 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/InvalidProducerIT.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/InvalidProducerIT.java
@@ -5,7 +5,6 @@
 package io.strimzi.kafka.bridge.http;
 
 import io.strimzi.kafka.bridge.BridgeContentType;
-import io.strimzi.kafka.bridge.MetricsReporter;
 import io.strimzi.kafka.bridge.clients.BasicKafkaClient;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaProducerConfig;
@@ -76,7 +75,7 @@ public class InvalidProducerIT extends HttpBridgeITAbstract {
 
         if ("FALSE".equals(BRIDGE_EXTERNAL_ENV)) {
             bridgeConfig = BridgeConfig.fromMap(cfg);
-            httpBridge = new HttpBridge(bridgeConfig, new MetricsReporter(jmxCollectorRegistry, meterRegistry));
+            httpBridge = new HttpBridge(bridgeConfig);
 
             LOGGER.info("Deploying in-memory bridge");
             vertx.deployVerticle(httpBridge).onComplete(context.succeeding(id -> context.completeNow()));
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java b/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java
index cac9159..5aa42b7 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/base/HttpBridgeITAbstract.java
@@ -5,9 +5,6 @@
 
 package io.strimzi.kafka.bridge.http.base;
 
-import io.micrometer.core.instrument.MeterRegistry;
-import io.strimzi.kafka.bridge.JmxCollectorRegistry;
-import io.strimzi.kafka.bridge.MetricsReporter;
 import io.strimzi.kafka.bridge.clients.BasicKafkaClient;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
@@ -20,6 +17,7 @@ import io.strimzi.kafka.bridge.http.services.BaseService;
 import io.strimzi.kafka.bridge.http.services.ConsumerService;
 import io.strimzi.kafka.bridge.http.services.ProducerService;
 import io.strimzi.kafka.bridge.http.services.SeekService;
+import io.strimzi.kafka.bridge.metrics.MetricsType;
 import io.strimzi.kafka.bridge.utils.Urls;
 import io.strimzi.test.container.StrimziKafkaContainer;
 import io.vertx.core.Vertx;
@@ -88,6 +86,7 @@ public abstract class HttpBridgeITAbstract {
         config.put(KafkaConsumerConfig.KAFKA_CONSUMER_CONFIG_PREFIX + ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
         config.put(KafkaProducerConfig.KAFKA_PRODUCER_CONFIG_PREFIX + ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000");
         config.put(HttpConfig.HTTP_CONSUMER_TIMEOUT, timeout);
+        config.put(BridgeConfig.METRICS_TYPE, MetricsType.STRIMZI_REPORTER.toString());
         config.put(BridgeConfig.BRIDGE_ID, "my-bridge");
     }
 
@@ -98,9 +97,6 @@ public abstract class HttpBridgeITAbstract {
     protected static HttpBridge httpBridge;
     protected static BridgeConfig bridgeConfig;
 
-    protected static MeterRegistry meterRegistry = null;
-    protected static JmxCollectorRegistry jmxCollectorRegistry = null;
-
     protected BaseService baseService() {
         return BaseService.getInstance(client);
     }
@@ -128,7 +124,8 @@ public abstract class HttpBridgeITAbstract {
 
         if ("FALSE".equals(BRIDGE_EXTERNAL_ENV)) {
             bridgeConfig = BridgeConfig.fromMap(config);
-            httpBridge = new HttpBridge(bridgeConfig, new MetricsReporter(jmxCollectorRegistry, meterRegistry));
+            
+            httpBridge = new HttpBridge(bridgeConfig);
 
             LOGGER.info("Deploying in-memory bridge");
             vertx.deployVerticle(httpBridge).onComplete(context.succeeding(id -> context.completeNow()));
diff --git a/src/test/java/io/strimzi/kafka/bridge/metrics/JmxMetricsCollectorTest.java b/src/test/java/io/strimzi/kafka/bridge/metrics/JmxMetricsCollectorTest.java
new file mode 100644
index 0000000..4a7f357
--- /dev/null
+++ b/src/test/java/io/strimzi/kafka/bridge/metrics/JmxMetricsCollectorTest.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.bridge.metrics;
+
+import io.prometheus.jmx.JmxCollector;
+import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+import io.prometheus.metrics.model.snapshots.MetricSnapshots;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class JmxMetricsCollectorTest {
+    @Test
+    void shouldReturnMetrics() throws IOException {
+        JmxCollector mockJmxCollector = mock(JmxCollector.class);
+            
+        PrometheusRegistry mockPromRegistry = mock(PrometheusRegistry.class);
+        MetricSnapshots mockSnapshots = mock(MetricSnapshots.class);
+        when(mockPromRegistry.scrape()).thenReturn(mockSnapshots);
+        
+        PrometheusTextFormatWriter mockPromFormatter = mock(PrometheusTextFormatWriter.class);
+        doAnswer(invocation -> {
+            ByteArrayOutputStream stream = invocation.getArgument(0);
+            stream.write("test_metric\n".getBytes(StandardCharsets.UTF_8));
+            return null;
+        }).when(mockPromFormatter).write(any(), any());
+
+        MetricsCollector metricsCollector = new JmxMetricsCollector(mockJmxCollector, mockPromRegistry, mockPromFormatter);
+        
+        String result = metricsCollector.scrape();
+        assertThat(result, containsString("test_metric"));
+        assertThat(result.getBytes(StandardCharsets.UTF_8).length, is(result.length()));
+    }
+
+    @Test
+    void shouldHandleIoException() throws IOException {
+        JmxCollector mockJmxCollector = mock(JmxCollector.class);
+        
+        PrometheusRegistry mockPromRegistry = mock(PrometheusRegistry.class);
+        MetricSnapshots mockSnapshots = mock(MetricSnapshots.class);
+        when(mockPromRegistry.scrape()).thenReturn(mockSnapshots);
+
+        PrometheusTextFormatWriter mockPromFormatter = mock(PrometheusTextFormatWriter.class);
+        doThrow(new IOException("Test exception"))
+            .when(mockPromFormatter).write(any(ByteArrayOutputStream.class), Mockito.eq(mockSnapshots));
+
+        MetricsCollector metricsCollector = new JmxMetricsCollector(mockJmxCollector, mockPromRegistry, mockPromFormatter);
+        
+        RuntimeException exception = assertThrows(RuntimeException.class, () -> metricsCollector.doScrape());
+        assertThat(exception.getMessage(), containsString("Test exception"));
+    }
+
+    @Test
+    void shouldThrowWithInvalidYaml() {
+        assertThrows(ClassCastException.class, () -> new JmxMetricsCollector("invalid"));
+    }
+}
diff --git a/src/test/java/io/strimzi/kafka/bridge/metrics/StrimziMetricsCollectorTest.java b/src/test/java/io/strimzi/kafka/bridge/metrics/StrimziMetricsCollectorTest.java
new file mode 100644
index 0000000..5899dd1
--- /dev/null
+++ b/src/test/java/io/strimzi/kafka/bridge/metrics/StrimziMetricsCollectorTest.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.kafka.bridge.metrics;
+
+import io.prometheus.metrics.expositionformats.PrometheusTextFormatWriter;
+import io.prometheus.metrics.model.registry.PrometheusRegistry;
+import io.prometheus.metrics.model.snapshots.MetricSnapshots;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class StrimziMetricsCollectorTest {
+    @Test
+    void shouldReturnMetrics() throws IOException {
+        PrometheusRegistry mockPromRegistry = mock(PrometheusRegistry.class);
+        MetricSnapshots mockSnapshots = mock(MetricSnapshots.class);
+        when(mockPromRegistry.scrape()).thenReturn(mockSnapshots);
+        
+        PrometheusTextFormatWriter mockPromFormatter = mock(PrometheusTextFormatWriter.class);
+        doAnswer(invocation -> {
+            ByteArrayOutputStream stream = invocation.getArgument(0);
+            stream.write("test_metric\n".getBytes(StandardCharsets.UTF_8));
+            return null;
+        }).when(mockPromFormatter).write(any(), any());
+
+        MetricsCollector metricsCollector = new StrimziMetricsCollector(mockPromRegistry, mockPromFormatter);
+        
+        String result = metricsCollector.scrape();
+        assertThat(result, containsString("test_metric"));
+        assertThat(result.getBytes(StandardCharsets.UTF_8).length, is(result.length()));
+    }
+
+    @Test
+    void shouldHandleIoException() throws IOException {
+        PrometheusRegistry mockPromRegistry = mock(PrometheusRegistry.class);
+        MetricSnapshots mockSnapshots = mock(MetricSnapshots.class);
+        when(mockPromRegistry.scrape()).thenReturn(mockSnapshots);
+
+        PrometheusTextFormatWriter mockPromFormatter = mock(PrometheusTextFormatWriter.class);
+        doThrow(new IOException("Test exception"))
+            .when(mockPromFormatter).write(any(ByteArrayOutputStream.class), Mockito.eq(mockSnapshots));
+
+        MetricsCollector metricsCollector = new StrimziMetricsCollector(mockPromRegistry, mockPromFormatter);
+        
+        RuntimeException exception = assertThrows(RuntimeException.class, () -> metricsCollector.doScrape());
+        assertThat(exception.getMessage(), containsString("Test exception"));
+    }
+}