Repository: strimzi/strimzi-kafka-bridge
Pull request #742 (2023-01-04 09:18:03)

Replaced Vert.x Config component with plain Java config retriever
This trivial PR replaces the Vert.x Config component with a simple `ConfigRetriever` class for getting bridge configuration from both properties file and environment variables.

scholzj: I also wonder ... if we move away to completely our stuff, would it make sense to have it more tightůy integrated into the `*Config` classes?
ppatierno: Our `*Config` classes needs a way to get the application properties file content anyway and then filtering each of them what they need (i.e. `http.`, `producer.` and so on). Loading configuration in a central place into a map and then filtering later makes sense to me instead of each `*Config` class accessing the properties file.

Patch:
diff --git a/pom.xml b/pom.xml
index fe91ef1..eefaeed 100644
--- a/pom.xml
+++ b/pom.xml
@@ -150,11 +150,6 @@
 			<artifactId>vertx-json-schema</artifactId>
 			<version>${vertx.version}</version>
 		</dependency>
-		<dependency>
-			<groupId>io.vertx</groupId>
-			<artifactId>vertx-config</artifactId>
-			<version>${vertx.version}</version>
-		</dependency>
 		<dependency>
 			<groupId>io.vertx</groupId>
 			<artifactId>vertx-micrometer-metrics</artifactId>
diff --git a/src/main/java/io/strimzi/kafka/bridge/AdminClientEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/AdminClientEndpoint.java
index 939dc57..433d34b 100644
--- a/src/main/java/io/strimzi/kafka/bridge/AdminClientEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/AdminClientEndpoint.java
@@ -7,7 +7,7 @@ package io.strimzi.kafka.bridge;
 
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
-import io.vertx.core.Handler;
+import io.strimzi.kafka.bridge.http.HttpBridgeEndpoint;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ListOffsetsResult;
@@ -28,13 +28,13 @@ import java.util.concurrent.CompletionStage;
 /**
  * Base class for admin client endpoint
  */
-public abstract class AdminClientEndpoint implements BridgeEndpoint {
+public abstract class AdminClientEndpoint implements HttpBridgeEndpoint {
     protected final Logger log = LoggerFactory.getLogger(AdminClientEndpoint.class);
 
     protected String name;
     protected final BridgeConfig bridgeConfig;
 
-    private Handler<BridgeEndpoint> closeHandler;
+    private Handler<HttpBridgeEndpoint> closeHandler;
 
     private AdminClient kAdminClient;
 
@@ -54,7 +54,7 @@ public abstract class AdminClientEndpoint implements BridgeEndpoint {
     }
 
     @Override
-    public BridgeEndpoint closeHandler(Handler<BridgeEndpoint> endpointCloseHandler) {
+    public HttpBridgeEndpoint closeHandler(Handler<HttpBridgeEndpoint> endpointCloseHandler) {
         this.closeHandler = endpointCloseHandler;
         return this;
     }
diff --git a/src/main/java/io/strimzi/kafka/bridge/Application.java b/src/main/java/io/strimzi/kafka/bridge/Application.java
index 8459e8b..8785ef4 100644
--- a/src/main/java/io/strimzi/kafka/bridge/Application.java
+++ b/src/main/java/io/strimzi/kafka/bridge/Application.java
@@ -7,16 +7,13 @@ 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;
 import io.strimzi.kafka.bridge.tracing.TracingUtil;
-import io.vertx.config.ConfigRetriever;
-import io.vertx.config.ConfigRetrieverOptions;
-import io.vertx.config.ConfigStoreOptions;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
 import io.vertx.core.Vertx;
 import io.vertx.core.VertxOptions;
-import io.vertx.core.json.JsonObject;
 import io.vertx.micrometer.Label;
 import io.vertx.micrometer.MetricsDomain;
 import io.vertx.micrometer.MicrometerMetricsOptions;
@@ -78,35 +75,14 @@ public class Application {
 
             CommandLine commandLine = new DefaultParser().parse(generateOptions(), args);
 
-            ConfigStoreOptions fileStore = new ConfigStoreOptions()
-                .setType("file")
-                .setFormat("properties")
-                .setConfig(new JsonObject().put("path", absoluteFilePath(commandLine.getOptionValue("config-file"))).put("raw-data", true));
-
-            ConfigStoreOptions envStore = new ConfigStoreOptions()
-                .setType("env")
-                .setConfig(new JsonObject().put("raw-data", true));
-
-            ConfigRetrieverOptions options = new ConfigRetrieverOptions()
-                    .addStore(fileStore)
-                    .addStore(envStore);
-
-            ConfigRetriever retriever = ConfigRetriever.create(vertx, options);
-            retriever.getConfig(ar -> {
-
-                if (ar.succeeded()) {
-                    Map<String, Object> config = ar.result().getMap();
-                    BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
-
-                    deployHttpBridge(vertx, bridgeConfig, metricsReporter).onComplete(done -> {
-                        if (done.succeeded()) {
-                            // register tracing - if set, etc
-                            TracingUtil.initialize(bridgeConfig);
-                        }
-                    });
-                } else {
-                    log.error("Error starting the bridge", ar.cause());
-                    System.exit(1);
+            Map<String, Object> config = ConfigRetriever.getConfig(absoluteFilePath(commandLine.getOptionValue("config-file")));
+            BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
+            log.info("Bridge configuration {}", bridgeConfig);
+
+            deployHttpBridge(vertx, bridgeConfig, metricsReporter).onComplete(done -> {
+                if (done.succeeded()) {
+                    // register tracing - if set, etc
+                    TracingUtil.initialize(bridgeConfig);
                 }
             });
         } catch (RuntimeException | MalformedObjectNameException | IOException | ParseException e) {
diff --git a/src/main/java/io/strimzi/kafka/bridge/ConnectionEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/ConnectionEndpoint.java
deleted file mode 100644
index f4d5c77..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/ConnectionEndpoint.java
+++ /dev/null
@@ -1,57 +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 java.util.ArrayList;
-import java.util.List;
-
-/**
- * Wrapper around sinks and source for an the endpoint connection
- */
-public class ConnectionEndpoint {
-
-    // more sink endpoints per connection, each of them handling a sender internally
-    private List<SinkBridgeEndpoint> sinks;
-    // only one source endpoint per connection, handling more receiver internally
-    private SourceBridgeEndpoint source;
-
-    /**
-     * Constructor
-     */
-    public ConnectionEndpoint() {
-        this.sinks = new ArrayList<>();
-        this.source = null;
-    }
-
-    /**
-     * Return the sink endpoints collection
-     *
-     * @return  Returns the sink endpoint collection
-     */
-    public List<SinkBridgeEndpoint> getSinks() {
-        return this.sinks;
-    }
-
-    /**
-     * Set the source endpoint for this connection
-     *
-     * @param source    source endpoint to set
-     * @return  current connection endpoint instance
-     */
-    public ConnectionEndpoint setSource(SourceBridgeEndpoint source) {
-        this.source = source;
-        return this;
-    }
-
-    /**
-     * Return the source endpoint
-     *
-     * @return  Returns the source endpoint
-     */
-    public SourceBridgeEndpoint getSource() {
-        return this.source;
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/Endpoint.java b/src/main/java/io/strimzi/kafka/bridge/Endpoint.java
deleted file mode 100644
index 7fb124f..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/Endpoint.java
+++ /dev/null
@@ -1,18 +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;
-
-/**
- * Interface for remote protocol endpoint
- * @param <T>   class representing the actual protocol endpoint
- */
-public interface Endpoint<T> {
-
-    /**
-     * @return  instance of the actual protocol endpoint
-     */
-    T get();
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/Handler.java b/src/main/java/io/strimzi/kafka/bridge/Handler.java
new file mode 100644
index 0000000..201cf3b
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/Handler.java
@@ -0,0 +1,22 @@
+/*
+ * 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;
+
+/**
+ * Represents a functional interface for handling callback when an asynchronous operation ends
+ *
+ * @param <T> type of the data to be handled
+ */
+@FunctionalInterface
+public interface Handler<T> {
+
+    /**
+     * Called to handle the result of the asynchronous operation with the provided data
+     *
+     * @param data data result to handle
+     */
+    void handle(T data);
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
index e82738e..cb28db3 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
@@ -7,7 +7,7 @@ package io.strimzi.kafka.bridge;
 
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
-import io.vertx.core.Handler;
+import io.strimzi.kafka.bridge.http.HttpBridgeEndpoint;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
@@ -35,7 +35,7 @@ import java.util.stream.Collectors;
  * @param <K>   type of Kafka message key
  * @param <V>   type of Kafka message payload
  */
-public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
+public abstract class SinkBridgeEndpoint<K, V> implements HttpBridgeEndpoint {
 
     protected final Logger log = LoggerFactory.getLogger(getClass());
 
@@ -46,7 +46,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
 
     protected final BridgeConfig bridgeConfig;
 
-    private Handler<BridgeEndpoint> closeHandler;
+    private Handler<HttpBridgeEndpoint> closeHandler;
 
     private Consumer<K, V> consumer;
     protected ConsumerInstanceId consumerInstanceId;
@@ -90,7 +90,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
     }
 
     @Override
-    public BridgeEndpoint closeHandler(Handler<BridgeEndpoint> endpointCloseHandler) {
+    public HttpBridgeEndpoint closeHandler(Handler<HttpBridgeEndpoint> endpointCloseHandler) {
         this.closeHandler = endpointCloseHandler;
         return this;
     }
diff --git a/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
index 7491f31..b8e5e81 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
@@ -7,9 +7,9 @@ package io.strimzi.kafka.bridge;
 
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
+import io.strimzi.kafka.bridge.http.HttpBridgeEndpoint;
 import io.strimzi.kafka.bridge.tracing.TracingHandle;
 import io.strimzi.kafka.bridge.tracing.TracingUtil;
-import io.vertx.core.Handler;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -25,7 +25,7 @@ import java.util.concurrent.CompletionStage;
 /**
  * Base class for source bridge endpoints
  */
-public abstract class SourceBridgeEndpoint<K, V> implements BridgeEndpoint {
+public abstract class SourceBridgeEndpoint<K, V> implements HttpBridgeEndpoint {
 
     protected final Logger log = LoggerFactory.getLogger(getClass());
 
@@ -36,7 +36,7 @@ public abstract class SourceBridgeEndpoint<K, V> implements BridgeEndpoint {
 
     protected final BridgeConfig bridgeConfig;
 
-    private Handler<BridgeEndpoint> closeHandler;
+    private Handler<HttpBridgeEndpoint> closeHandler;
 
     private Producer<K, V> producer;
 
@@ -62,7 +62,7 @@ public abstract class SourceBridgeEndpoint<K, V> implements BridgeEndpoint {
     }
 
     @Override
-    public BridgeEndpoint closeHandler(Handler<BridgeEndpoint> endpointCloseHandler) {
+    public HttpBridgeEndpoint closeHandler(Handler<HttpBridgeEndpoint> endpointCloseHandler) {
         this.closeHandler = endpointCloseHandler;
         return this;
     }
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/ConfigRetriever.java b/src/main/java/io/strimzi/kafka/bridge/config/ConfigRetriever.java
new file mode 100644
index 0000000..dfa2319
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/config/ConfigRetriever.java
@@ -0,0 +1,60 @@
+/*
+ * 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 java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+/**
+ * Retrieve the bridge configuration from properties file and environment variables
+ */
+public class ConfigRetriever {
+
+    /**
+     * Retrieve the bridge configuration from the properties file provided as parameter
+     * and adding environment variables
+     * If a parameter is defined in both properties file and environment variables, the latter wins
+     *
+     * @param path path to the properties file
+     * @return configuration as key-value pairs
+     * @throws IOException when not possible to get the properties file
+     */
+    public static Map<String, Object> getConfig(String path) throws IOException {
+        return getConfig(path, System.getenv());
+    }
+
+    /**
+     * Retrieve the bridge configuration from the properties file provided as parameter
+     * and adding the additional configuration parameter provided as well
+     * If a parameter is defined in both properties file and additional configuration, the latter wins
+     *
+     * @param path path to the properties file
+     * @param additionalConfig additional configuration to add
+     * @return configuration as key-value pairs
+     * @throws IOException when not possible to get the properties file
+     */
+    public static Map<String, Object> getConfig(String path, Map<String, String> additionalConfig) throws IOException {
+        Map<String, Object> configuration;
+        try (InputStream is = new FileInputStream(path)) {
+            Properties props = new Properties();
+            props.load(is);
+            configuration =
+                    props.entrySet().stream().collect(
+                            Collectors.toMap(
+                                    e -> String.valueOf(e.getKey()),
+                                    e -> String.valueOf(e.getValue()),
+                                    (prev, next) -> next, HashMap::new
+                            ));
+        }
+        configuration.putAll(additionalConfig);
+        return configuration;
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminClientEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminClientEndpoint.java
index 8aa23e2..e0cbd48 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminClientEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpAdminClientEndpoint.java
@@ -11,11 +11,10 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.AdminClientEndpoint;
 import io.strimzi.kafka.bridge.BridgeContentType;
-import io.strimzi.kafka.bridge.Endpoint;
+import io.strimzi.kafka.bridge.Handler;
 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.vertx.core.Handler;
 import io.vertx.ext.web.RoutingContext;
 import org.apache.kafka.clients.admin.Config;
 import org.apache.kafka.clients.admin.ConfigEntry;
@@ -59,8 +58,7 @@ public class HttpAdminClientEndpoint extends AdminClientEndpoint {
     }
 
     @Override
-    public void handle(Endpoint<?> endpoint, Handler<?> handler) {
-        RoutingContext routingContext = (RoutingContext) endpoint.get();
+    public void handle(RoutingContext routingContext, Handler<HttpBridgeEndpoint> handler) {
         log.trace("HttpAdminClientEndpoint handle thread {}", Thread.currentThread());
         switch (this.httpBridgeContext.getOpenApiOperation()) {
             case LIST_TOPICS:
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 50c8bd7..063fc96 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -303,7 +303,7 @@ public class HttpBridge extends AbstractVerticle {
             });        
             sink.open();
 
-            sink.handle(new HttpEndpoint(routingContext), endpoint -> {
+            sink.handle(routingContext, endpoint -> {
                 HttpSinkBridgeEndpoint<byte[], byte[]> httpEndpoint = (HttpSinkBridgeEndpoint<byte[], byte[]>) endpoint;
                 httpBridgeContext.getHttpSinkEndpoints().put(httpEndpoint.consumerInstanceId(), httpEndpoint);
                 timestampMap.put(httpEndpoint.consumerInstanceId(), System.currentTimeMillis());
@@ -334,7 +334,7 @@ public class HttpBridge extends AbstractVerticle {
         SinkBridgeEndpoint<byte[], byte[]> deleteSinkEndpoint = this.httpBridgeContext.getHttpSinkEndpoints().get(kafkaConsumerInstanceId);
 
         if (deleteSinkEndpoint != null) {
-            deleteSinkEndpoint.handle(new HttpEndpoint(routingContext));
+            deleteSinkEndpoint.handle(routingContext);
 
             this.httpBridgeContext.getHttpSinkEndpoints().remove(kafkaConsumerInstanceId);
             timestampMap.remove(kafkaConsumerInstanceId);
@@ -432,7 +432,7 @@ public class HttpBridge extends AbstractVerticle {
 
         if (sinkEndpoint != null) {
             timestampMap.replace(kafkaConsumerInstanceId, System.currentTimeMillis());
-            sinkEndpoint.handle(new HttpEndpoint(routingContext));
+            sinkEndpoint.handle(routingContext);
         } else {
             HttpBridgeError error = new HttpBridgeError(
                     HttpResponseStatus.NOT_FOUND.code(),
@@ -475,7 +475,7 @@ public class HttpBridge extends AbstractVerticle {
                 source.open();
                 this.httpBridgeContext.getHttpSourceEndpoints().put(httpServerRequest.connection(), source);
             }
-            source.handle(new HttpEndpoint(routingContext));
+            source.handle(routingContext);
 
         } catch (Exception ex) {
             if (source != null) {
@@ -493,7 +493,7 @@ public class HttpBridge extends AbstractVerticle {
     private void processAdminClient(RoutingContext routingContext) {
         AdminClientEndpoint adminClientEndpoint = this.httpBridgeContext.getAdminClientEndpoint();
         if (adminClientEndpoint != null) {
-            adminClientEndpoint.handle(new HttpEndpoint(routingContext));
+            adminClientEndpoint.handle(routingContext);
         } else {
             HttpBridgeError error = new HttpBridgeError(
                     HttpResponseStatus.INTERNAL_SERVER_ERROR.code(),
diff --git a/src/main/java/io/strimzi/kafka/bridge/BridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeEndpoint.java
similarity index 51%
rename from src/main/java/io/strimzi/kafka/bridge/BridgeEndpoint.java
rename to src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeEndpoint.java
index 75fb0e0..5e52e0b 100644
--- a/src/main/java/io/strimzi/kafka/bridge/BridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeEndpoint.java
@@ -3,15 +3,16 @@
  * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
  */
 
-package io.strimzi.kafka.bridge;
+package io.strimzi.kafka.bridge.http;
 
-import io.vertx.core.Handler;
+import io.strimzi.kafka.bridge.Handler;
+import io.vertx.ext.web.RoutingContext;
 
 /**
  * Interface for classes which acts as endpoints
  * bridging traffic between a protocol and Apache Kafka
  */
-public interface BridgeEndpoint {
+public interface HttpBridgeEndpoint {
 
     /**
      * Name of the bridge endpoint
@@ -31,19 +32,19 @@ public interface BridgeEndpoint {
     void close();
 
     /**
-     * Handler for the remote protocol endpoint
-     * @param endpoint Remote protocol endpoint to handle
+     * Handler for the HTTP routing context
+     * @param routingContext HTTP routing context to handle
      */
-    default void handle(Endpoint<?> endpoint) {
-        this.handle(endpoint, null);
+    default void handle(RoutingContext routingContext) {
+        this.handle(routingContext, null);
     }
 
     /**
-     * Handler for the remote protocol endpoint
-     * @param endpoint Remote protocol endpoint to handle
-     * @param handler handler for result
+     * Handler for the HTTP routing context
+     * @param routingContext HTTP routing context to handle
+     * @param handler handler for the corresponding bridge endpoint
      */
-    void handle(Endpoint<?> endpoint, Handler<?> handler);
+    void handle(RoutingContext routingContext, Handler<HttpBridgeEndpoint> handler);
 
     /**
      * Sets an handler called when a bridge endpoint is closed due to internal processing
@@ -51,5 +52,5 @@ public interface BridgeEndpoint {
      * @param endpointCloseHandler The handler
      * @return The bridge endpoint
      */
-    BridgeEndpoint closeHandler(Handler<BridgeEndpoint> endpointCloseHandler);
+    HttpBridgeEndpoint closeHandler(Handler<HttpBridgeEndpoint> endpointCloseHandler);
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpEndpoint.java
deleted file mode 100644
index 291a4e8..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpEndpoint.java
+++ /dev/null
@@ -1,30 +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.http;
-
-import io.strimzi.kafka.bridge.Endpoint;
-import io.vertx.ext.web.RoutingContext;
-
-/**
- * HTTP implementation for an endpoint
- */
-public class HttpEndpoint implements Endpoint<RoutingContext> {
-
-    private RoutingContext routingContext;
-
-    /**
-     * Constructor
-     *
-     * @param routingContext the routing context
-     */
-    public HttpEndpoint(RoutingContext routingContext) {
-        this.routingContext = routingContext;
-    }
-    @Override
-    public RoutingContext get() {
-        return routingContext;
-    }
-}
\ No newline at end of file
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
index a365c69..c3f3cfb 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -13,7 +13,7 @@ import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.ConsumerInstanceId;
 import io.strimzi.kafka.bridge.EmbeddedFormat;
-import io.strimzi.kafka.bridge.Endpoint;
+import io.strimzi.kafka.bridge.Handler;
 import io.strimzi.kafka.bridge.SinkBridgeEndpoint;
 import io.strimzi.kafka.bridge.SinkTopicSubscription;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
@@ -26,7 +26,6 @@ import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.strimzi.kafka.bridge.tracing.SpanHandle;
 import io.strimzi.kafka.bridge.tracing.TracingHandle;
 import io.strimzi.kafka.bridge.tracing.TracingUtil;
-import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.RoutingContext;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -88,7 +87,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
      * @param handler handler for the request
      */
     @SuppressWarnings("checkstyle:NPathComplexity")
-    public void doCreateConsumer(RoutingContext routingContext, JsonNode bodyAsJson, Handler<SinkBridgeEndpoint<K, V>> handler) {
+    public void doCreateConsumer(RoutingContext routingContext, JsonNode bodyAsJson, Handler<HttpBridgeEndpoint> handler) {
         // get the consumer group-id
         this.groupId = routingContext.pathParam("groupid");
 
@@ -515,9 +514,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
 
     @SuppressWarnings({"checkstyle:CyclomaticComplexity"})
     @Override
-    public void handle(Endpoint<?> endpoint, Handler<?> handler) {
-        RoutingContext routingContext = (RoutingContext) endpoint.get();
-
+    public void handle(RoutingContext routingContext, Handler<HttpBridgeEndpoint> handler) {
         JsonNode bodyAsJson = EMPTY_JSON;
         try {
             // check for an empty body
@@ -536,7 +533,7 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         switch (this.httpBridgeContext.getOpenApiOperation()) {
 
             case CREATE_CONSUMER:
-                doCreateConsumer(routingContext, bodyAsJson, (Handler<SinkBridgeEndpoint<K, V>>) handler);
+                doCreateConsumer(routingContext, bodyAsJson, handler);
                 break;
 
             case SUBSCRIBE:
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
index b1814d4..9905d82 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -10,7 +10,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.EmbeddedFormat;
-import io.strimzi.kafka.bridge.Endpoint;
+import io.strimzi.kafka.bridge.Handler;
 import io.strimzi.kafka.bridge.SourceBridgeEndpoint;
 import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
@@ -22,7 +22,6 @@ import io.strimzi.kafka.bridge.http.model.HttpBridgeResult;
 import io.strimzi.kafka.bridge.tracing.SpanHandle;
 import io.strimzi.kafka.bridge.tracing.TracingHandle;
 import io.strimzi.kafka.bridge.tracing.TracingUtil;
-import io.vertx.core.Handler;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.ext.web.RoutingContext;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -71,9 +70,7 @@ public class HttpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
 
     @Override
     @SuppressWarnings("checkstyle:NPathComplexity")
-    public void handle(Endpoint<?> endpoint, Handler<?> handler) {
-        RoutingContext routingContext = (RoutingContext) endpoint.get();
-
+    public void handle(RoutingContext routingContext, Handler<HttpBridgeEndpoint> handler) {
         String topic = routingContext.pathParam("topicname");
 
         List<ProducerRecord<K, V>> records;


Test patch:
diff --git a/perftest/README.md b/perftest/README.md
index 0ad7321..9d211ba 100644
--- a/perftest/README.md
+++ b/perftest/README.md
@@ -1,8 +1,12 @@
 # Performance tests
 
-This folder contains a [JMeter](https://jmeter.apache.org/) configuration file decribing a test plan with:
+This folder contains a [JMeter](https://jmeter.apache.org/) JMX configuration file describing a test plan with the following operations:
 
 * consumers creation, topics subscription, polling for getting records in a loop, and final consumers deletion
 * producers sending records to topics in a loop
 
-The test plan is configurable changing the number of consumers/producers (JMeter threads) and the number of loop cycles for sending/receiving records.
\ No newline at end of file
+The test plan is configurable. You can change the number of consumers/producers (JMeter threads) and the number of loop cycles for sending/receiving records.
+
+It needs a set of plugins in order to show some graphs.
+For this reason, you need to download the JMeter Plugins Manager from the [jmeter-plugins.org](https://jmeter-plugins.org/get/) website and put it into the `lib/ext` folder.
+When opening the JMX configuration file for the first time, JMeter will ask to install the plugins to run the test plan.
\ No newline at end of file
diff --git a/perftest/strimzi-kafka-bridge.jmx b/perftest/strimzi-kafka-bridge.jmx
index 00c0736..2426b3b 100644
--- a/perftest/strimzi-kafka-bridge.jmx
+++ b/perftest/strimzi-kafka-bridge.jmx
@@ -1,5 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<jmeterTestPlan version="1.2" properties="5.0" jmeter="5.1.1 r1855137">
+<jmeterTestPlan version="1.2" properties="5.0" jmeter="5.5">
   <hashTree>
     <TestPlan guiclass="TestPlanGui" testclass="TestPlan" testname="Strimzi HTTP Bridge Test Plan" enabled="true">
       <stringProp name="TestPlan.comments"></stringProp>
@@ -23,6 +23,7 @@
         <boolProp name="ThreadGroup.scheduler">false</boolProp>
         <stringProp name="ThreadGroup.duration"></stringProp>
         <stringProp name="ThreadGroup.delay"></stringProp>
+        <boolProp name="ThreadGroup.same_user_on_next_iteration">true</boolProp>
       </ThreadGroup>
       <hashTree>
         <ConfigTestElement guiclass="HttpDefaultsGui" testclass="ConfigTestElement" testname="HTTP Request Defaults" enabled="true">
@@ -335,6 +336,7 @@
             </value>
           </objProp>
           <stringProp name="filename"></stringProp>
+          <boolProp name="useGroupName">true</boolProp>
         </ResultCollector>
         <hashTree/>
         <kg.apc.jmeter.vizualizers.CorrectedResultCollector guiclass="kg.apc.jmeter.vizualizers.TransactionsPerSecondGui" testclass="kg.apc.jmeter.vizualizers.CorrectedResultCollector" testname="jp@gc - Transactions per Second" enabled="true">
@@ -394,6 +396,7 @@
         <boolProp name="ThreadGroup.scheduler">false</boolProp>
         <stringProp name="ThreadGroup.duration"></stringProp>
         <stringProp name="ThreadGroup.delay"></stringProp>
+        <boolProp name="ThreadGroup.same_user_on_next_iteration">true</boolProp>
       </ThreadGroup>
       <hashTree>
         <ConfigTestElement guiclass="HttpDefaultsGui" testclass="ConfigTestElement" testname="HTTP Request Defaults" enabled="true">
@@ -556,6 +559,7 @@
               </value>
             </objProp>
             <stringProp name="filename"></stringProp>
+            <boolProp name="useGroupName">true</boolProp>
           </ResultCollector>
           <hashTree/>
           <kg.apc.jmeter.vizualizers.CorrectedResultCollector guiclass="kg.apc.jmeter.vizualizers.TransactionsPerSecondGui" testclass="kg.apc.jmeter.vizualizers.CorrectedResultCollector" testname="jp@gc - Transactions per Second" enabled="true">
diff --git a/src/test/java/io/strimzi/kafka/bridge/ConfigRetrieverTest.java b/src/test/java/io/strimzi/kafka/bridge/ConfigRetrieverTest.java
new file mode 100644
index 0000000..c02d67f
--- /dev/null
+++ b/src/test/java/io/strimzi/kafka/bridge/ConfigRetrieverTest.java
@@ -0,0 +1,81 @@
+/*
+ * 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.strimzi.kafka.bridge.config.BridgeConfig;
+import io.strimzi.kafka.bridge.config.ConfigRetriever;
+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.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import static io.strimzi.kafka.bridge.config.BridgeConfig.BRIDGE_ID;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Unit tests for the ConfigRetriever class
+ */
+public class ConfigRetrieverTest {
+
+    @Test
+    public void testApplicationPropertiesFile() throws IOException {
+        String path = getClass().getClassLoader().getResource("application.properties").getPath();
+        Map<String, Object> config = ConfigRetriever.getConfig(path);
+        BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
+
+        assertThat(bridgeConfig.getBridgeID(), is("my-bridge"));
+
+        assertThat(bridgeConfig.getKafkaConfig().getConfig().size(), is(1));
+        assertThat(bridgeConfig.getKafkaConfig().getConfig().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG), is("localhost:9092"));
+
+        assertThat(bridgeConfig.getKafkaConfig().getAdminConfig().getConfig().size(), is(0));
+
+        assertThat(bridgeConfig.getKafkaConfig().getProducerConfig().getConfig().size(), is(1));
+        assertThat(bridgeConfig.getKafkaConfig().getProducerConfig().getConfig().get(ProducerConfig.ACKS_CONFIG), is("1"));
+
+        assertThat(bridgeConfig.getKafkaConfig().getConsumerConfig().getConfig().size(), is(1));
+        assertThat(bridgeConfig.getKafkaConfig().getConsumerConfig().getConfig().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG), is("earliest"));
+
+        assertThat(bridgeConfig.getHttpConfig().getConfig().size(), is(2));
+        assertThat(bridgeConfig.getHttpConfig().getHost(), is("0.0.0.0"));
+        assertThat(bridgeConfig.getHttpConfig().getPort(), is(8080));
+    }
+
+    @Test
+    public void testEnvVarOverride() throws IOException {
+        // "simulating" an addition to the current environment variables
+        Map<String, String> env = new HashMap<>();
+        env.putAll(System.getenv());
+        env.put(BRIDGE_ID, "different-bridge-id");
+
+        String path = getClass().getClassLoader().getResource("application.properties").getPath();
+        Map<String, Object> config = ConfigRetriever.getConfig(path, env);
+        BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
+
+        assertThat(bridgeConfig.getBridgeID(), is(env.get(BRIDGE_ID)));
+    }
+
+    @Test
+    public void testNoApplicationPropertiesFile() throws IOException {
+        assertThrows(FileNotFoundException.class, () -> ConfigRetriever.getConfig("no-existing.properties"));
+    }
+
+    @Test
+    public void testWrongApplicationPropertiesFile() throws IOException {
+        String path = getClass().getClassLoader().getResource("wrong.properties").getPath();
+        Map<String, Object> config = ConfigRetriever.getConfig(path);
+        BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
+
+        assertThat(bridgeConfig.getConfig().size(), is(0));
+    }
+}
diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties
new file mode 100644
index 0000000..97e099d
--- /dev/null
+++ b/src/test/resources/application.properties
@@ -0,0 +1,15 @@
+#Bridge related settings
+bridge.id=my-bridge
+
+#Apache Kafka common
+kafka.bootstrap.servers=localhost:9092
+
+#Apache Kafka producer
+kafka.producer.acks=1
+
+#Apache Kafka consumer
+kafka.consumer.auto.offset.reset=earliest
+
+#HTTP related settings
+http.host=0.0.0.0
+http.port=8080
\ No newline at end of file
diff --git a/src/test/resources/wrong.properties b/src/test/resources/wrong.properties
new file mode 100644
index 0000000..3a77e81
--- /dev/null
+++ b/src/test/resources/wrong.properties
@@ -0,0 +1,1 @@
+no.meaningful.parameter=its-value
\ No newline at end of file