Repository: strimzi/strimzi-kafka-bridge
Pull request #288 (2019-07-29 09:29:24)

Bridge configuration refactoring
This PR applies a refactoring around the bridge configuration which doesn't make more sense as it is today and even because doesn't allow to set overall bridge configuration parameter (like the `id` needed in this PR #279) instead of adding it to every "bridge" configuration AMQP and HTTP.
After this PR I would make a new one trying to remove the overload of the bridge word on AMQP and HTTP which are more "endpoints" or "protocol implementations" of the bridge but not bridge themselves.


Patch:
diff --git a/src/main/java/io/strimzi/kafka/bridge/Application.java b/src/main/java/io/strimzi/kafka/bridge/Application.java
index c497e50..1f520c1 100644
--- a/src/main/java/io/strimzi/kafka/bridge/Application.java
+++ b/src/main/java/io/strimzi/kafka/bridge/Application.java
@@ -6,9 +6,8 @@
 package io.strimzi.kafka.bridge;
 
 import io.strimzi.kafka.bridge.amqp.AmqpBridge;
-import io.strimzi.kafka.bridge.amqp.AmqpBridgeConfig;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.http.HttpBridge;
-import io.strimzi.kafka.bridge.http.HttpBridgeConfig;
 import io.vertx.config.ConfigRetriever;
 import io.vertx.config.ConfigRetrieverOptions;
 import io.vertx.config.ConfigStoreOptions;
@@ -65,19 +64,18 @@ public class Application {
         retriever.getConfig(ar -> {
 
             Map<String, Object> config = ar.result().getMap();
-            AmqpBridgeConfig amqpBridgeConfig = AmqpBridgeConfig.fromMap(config);
-            HttpBridgeConfig httpBridgeConfig = HttpBridgeConfig.fromMap(config);
+            BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
 
             int healthServerPort = Integer.valueOf(config.getOrDefault(HEALTH_SERVER_PORT, DEFAULT_HEALTH_SERVER_PORT).toString());
 
-            if (amqpBridgeConfig.getEndpointConfig().isEnabled() && amqpBridgeConfig.getEndpointConfig().getPort() == healthServerPort) {
+            if (bridgeConfig.getAmqpConfig().isEnabled() && bridgeConfig.getAmqpConfig().getPort() == healthServerPort) {
                 log.error("Health server port {} conflicts with configured AMQP port", healthServerPort);
                 System.exit(1);
             }
 
             List<Future> futures = new ArrayList<>();
-            futures.add(deployAmqpBridge(vertx, amqpBridgeConfig));
-            futures.add(deployHttpBridge(vertx, httpBridgeConfig));
+            futures.add(deployAmqpBridge(vertx, bridgeConfig));
+            futures.add(deployHttpBridge(vertx, bridgeConfig));
 
             CompositeFuture.join(futures).setHandler(done -> {
                 if (done.succeeded()) {
@@ -95,7 +93,7 @@ public class Application {
                     
                     // when HTTP protocol is enabled, it handles healthy/ready endpoints as well,
                     // so no need for a standalone HTTP health server
-                    if (!httpBridgeConfig.getEndpointConfig().isEnabled()) {
+                    if (!bridgeConfig.getHttpConfig().isEnabled()) {
                         healthChecker.startHealthServer(vertx, healthServerPort);
                     }
                 }
@@ -107,14 +105,14 @@ public class Application {
      * Deploys the AMQP bridge into a new verticle
      *
      * @param vertx                 Vertx instance
-     * @param amqpBridgeConfig      AMQP Bridge configuration
+     * @param bridgeConfig          Bridge configuration
      * @return                      Future for the bridge startup
      */
-    private static Future<AmqpBridge> deployAmqpBridge(Vertx vertx, AmqpBridgeConfig amqpBridgeConfig)  {
+    private static Future<AmqpBridge> deployAmqpBridge(Vertx vertx, BridgeConfig bridgeConfig)  {
         Future<AmqpBridge> amqpFuture = Future.future();
 
-        if (amqpBridgeConfig.getEndpointConfig().isEnabled()) {
-            AmqpBridge amqpBridge = new AmqpBridge(amqpBridgeConfig);
+        if (bridgeConfig.getAmqpConfig().isEnabled()) {
+            AmqpBridge amqpBridge = new AmqpBridge(bridgeConfig);
 
             vertx.deployVerticle(amqpBridge, done -> {
                 if (done.succeeded()) {
@@ -136,14 +134,14 @@ public class Application {
      * Deploys the HTTP bridge into a new verticle
      *
      * @param vertx                 Vertx instance
-     * @param httpBridgeConfig      HTTP Bridge configuration
+     * @param bridgeConfig          Bridge configuration
      * @return                      Future for the bridge startup
      */
-    private static Future<HttpBridge> deployHttpBridge(Vertx vertx, HttpBridgeConfig httpBridgeConfig)  {
+    private static Future<HttpBridge> deployHttpBridge(Vertx vertx, BridgeConfig bridgeConfig)  {
         Future<HttpBridge> httpFuture = Future.future();
 
-        if (httpBridgeConfig.getEndpointConfig().isEnabled()) {
-            HttpBridge httpBridge = new HttpBridge(httpBridgeConfig);
+        if (bridgeConfig.getHttpConfig().isEnabled()) {
+            HttpBridge httpBridge = new HttpBridge(bridgeConfig);
             
             vertx.deployVerticle(httpBridge, done -> {
                 if (done.succeeded()) {
diff --git a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
index a0bec86..4832436 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
@@ -52,7 +52,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
     protected final Deserializer<V> valueDeserializer;
     protected final Vertx vertx;
 
-    protected final BridgeConfig bridgeConfigProperties;
+    protected final BridgeConfig bridgeConfig;
 
     private Handler<BridgeEndpoint> closeHandler;
 
@@ -96,15 +96,15 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
      * Constructor
      *
      * @param vertx Vert.x instance
-     * @param bridgeConfigProperties Bridge configuration
+     * @param bridgeConfig Bridge configuration
      * @param format embedded format for the key/value in the Kafka message
      * @param keyDeserializer Kafka deserializer for the message key
      * @param valueDeserializer Kafka deserializer for the message value
      */
-    public SinkBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfigProperties,
+    public SinkBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfig,
                               EmbeddedFormat format, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
         this.vertx = vertx;
-        this.bridgeConfigProperties = bridgeConfigProperties;
+        this.bridgeConfig = bridgeConfig;
         this.topicSubscriptions = new ArrayList<>();
         this.format = format;
         this.keyDeserializer = keyDeserializer;
@@ -146,7 +146,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
     protected void initConsumer(boolean shouldAttachBatchHandler, Properties config) {
 
         // create a consumer
-        KafkaConfig kafkaConfig = this.bridgeConfigProperties.getKafkaConfig();
+        KafkaConfig kafkaConfig = this.bridgeConfig.getKafkaConfig();
         Properties props = new Properties();
         props.putAll(kafkaConfig.getConfig());
         props.putAll(kafkaConfig.getConsumerConfig().getConfig());
diff --git a/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
index 4e51493..b4a5c41 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
@@ -33,7 +33,7 @@ public abstract class SourceBridgeEndpoint<K, V> implements BridgeEndpoint {
     protected final Serializer<V> valueSerializer;
     protected final Vertx vertx;
 
-    protected final BridgeConfig bridgeConfigProperties;
+    protected final BridgeConfig bridgeConfig;
 
     private Handler<BridgeEndpoint> closeHandler;
 
@@ -44,15 +44,15 @@ public abstract class SourceBridgeEndpoint<K, V> implements BridgeEndpoint {
      * Constructor
      *
      * @param vertx Vert.x instance
-     * @param bridgeConfigProperties Bridge configuration
+     * @param bridgeConfig Bridge configuration
      * @param format embedded format for the key/value in the Kafka message
      * @param keySerializer Kafka serializer for the message key
      * @param valueSerializer Kafka serializer for the message value
      */
-    public SourceBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfigProperties,
+    public SourceBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfig,
                                 EmbeddedFormat format, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
         this.vertx = vertx;
-        this.bridgeConfigProperties = bridgeConfigProperties;
+        this.bridgeConfig = bridgeConfig;
         this.format = format;
         this.keySerializer = keySerializer;
         this.valueSerializer = valueSerializer;
@@ -98,7 +98,7 @@ public abstract class SourceBridgeEndpoint<K, V> implements BridgeEndpoint {
     @Override
     public void open() {
 
-        KafkaConfig kafkaConfig = this.bridgeConfigProperties.getKafkaConfig();
+        KafkaConfig kafkaConfig = this.bridgeConfig.getKafkaConfig();
         Properties props = new Properties();
         props.putAll(kafkaConfig.getConfig());
         props.putAll(kafkaConfig.getProducerConfig().getConfig());
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridge.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridge.java
index a983244..2669099 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridge.java
@@ -11,6 +11,7 @@ import io.strimzi.kafka.bridge.HealthCheckable;
 import io.strimzi.kafka.bridge.SinkBridgeEndpoint;
 import io.strimzi.kafka.bridge.SourceBridgeEndpoint;
 import io.strimzi.kafka.bridge.amqp.converter.AmqpDefaultMessageConverter;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.AsyncResult;
@@ -72,7 +73,7 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
     public static final String AMQP_PARTITION_FILTER = "io.strimzi:partition-filter:int";
     public static final String AMQP_OFFSET_FILTER = "io.strimzi:offset-filter:long";
 
-    private final AmqpBridgeConfig amqpBridgeConfig;
+    private final BridgeConfig bridgeConfig;
 
     // container-id needed for working in "client" mode
     private static final String CONTAINER_ID = "kafka-bridge-service";
@@ -92,10 +93,10 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
     /**
      * Constructor
      *
-     * @param amqpBridgeConfig bridge configuration for AMQP support
+     * @param bridgeConfig bridge configuration
      */
-    public AmqpBridge(AmqpBridgeConfig amqpBridgeConfig) {
-        this.amqpBridgeConfig = amqpBridgeConfig;
+    public AmqpBridge(BridgeConfig bridgeConfig) {
+        this.bridgeConfig = bridgeConfig;
     }
 
     /**
@@ -115,7 +116,7 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
 
                         log.info("AMQP-Kafka Bridge started and listening on port {}", ar.result().actualPort());
                         log.info("AMQP-Kafka Bridge bootstrap servers {}",
-                                this.amqpBridgeConfig.getKafkaConfig().getConfig()
+                                this.bridgeConfig.getKafkaConfig().getConfig()
                                         .get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)
                         );
 
@@ -138,8 +139,8 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
 
         this.client = ProtonClient.create(this.vertx);
 
-        String host = this.amqpBridgeConfig.getEndpointConfig().getHost();
-        int port = this.amqpBridgeConfig.getEndpointConfig().getPort();
+        String host = this.bridgeConfig.getAmqpConfig().getHost();
+        int port = this.bridgeConfig.getAmqpConfig().getPort();
 
         ProtonClientOptions options = this.createClientOptions();
 
@@ -154,7 +155,7 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
 
                 log.info("AMQP-Kafka Bridge started and connected in client mode to {}:{}", host, port);
                 log.info("AMQP-Kafka Bridge bootstrap servers {}",
-                        this.amqpBridgeConfig.getKafkaConfig().getConfig()
+                        this.bridgeConfig.getKafkaConfig().getConfig()
                                 .get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)
                 );
 
@@ -176,7 +177,7 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
 
         this.endpoints = new HashMap<>();
 
-        AmqpMode mode = this.amqpBridgeConfig.getEndpointConfig().getMode();
+        AmqpMode mode = this.bridgeConfig.getAmqpConfig().getMode();
         log.info("AMQP-Kafka Bridge configured in {} mode", mode);
         if (mode == AmqpMode.SERVER) {
             this.bindAmqpServer(startFuture);
@@ -230,11 +231,11 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
     private ProtonServerOptions createServerOptions() {
 
         ProtonServerOptions options = new ProtonServerOptions();
-        options.setHost(this.amqpBridgeConfig.getEndpointConfig().getHost());
-        options.setPort(this.amqpBridgeConfig.getEndpointConfig().getPort());
+        options.setHost(this.bridgeConfig.getAmqpConfig().getHost());
+        options.setPort(this.bridgeConfig.getAmqpConfig().getPort());
 
-        if (this.amqpBridgeConfig.getEndpointConfig().getCertDir() != null && this.amqpBridgeConfig.getEndpointConfig().getCertDir().length() > 0) {
-            String certDir = this.amqpBridgeConfig.getEndpointConfig().getCertDir();
+        if (this.bridgeConfig.getAmqpConfig().getCertDir() != null && this.bridgeConfig.getAmqpConfig().getCertDir().length() > 0) {
+            String certDir = this.bridgeConfig.getAmqpConfig().getCertDir();
             log.info("Enabling SSL configuration for AMQP with TLS certificates from {}", certDir);
             options.setSsl(true)
                     .setPemTrustOptions(new PemTrustOptions()
@@ -258,8 +259,8 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
         options.setConnectTimeout(1000);
         options.setReconnectAttempts(-1).setReconnectInterval(1000); // reconnect forever, every 1000 millisecs
 
-        if (this.amqpBridgeConfig.getEndpointConfig().getCertDir() != null && this.amqpBridgeConfig.getEndpointConfig().getCertDir().length() > 0) {
-            String certDir = this.amqpBridgeConfig.getEndpointConfig().getCertDir();
+        if (this.bridgeConfig.getAmqpConfig().getCertDir() != null && this.bridgeConfig.getAmqpConfig().getCertDir().length() > 0) {
+            String certDir = this.bridgeConfig.getAmqpConfig().getCertDir();
             log.info("Enabling SSL configuration for AMQP with TLS certificates from {}", certDir);
             options.setSsl(true)
                     .addEnabledSaslMechanism("EXTERNAL")
@@ -294,7 +295,7 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
                     this.processOpenSender(connection, sender);
                 });
 
-        if (this.amqpBridgeConfig.getEndpointConfig().getMode() == AmqpMode.CLIENT) {
+        if (this.bridgeConfig.getAmqpConfig().getMode() == AmqpMode.CLIENT) {
             connection.open();
         }
     }
@@ -397,7 +398,7 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
         // the source endpoint is only one, handling more AMQP receiver links internally
         if (source == null) {
             // TODO: the AMQP client should be able to specify the format during link attachment
-            source = new AmqpSourceBridgeEndpoint<>(this.vertx, this.amqpBridgeConfig,
+            source = new AmqpSourceBridgeEndpoint<>(this.vertx, this.bridgeConfig,
                     EmbeddedFormat.JSON, new StringSerializer(), new ByteArraySerializer());
 
             source.closeHandler(s -> {
@@ -422,7 +423,7 @@ public class AmqpBridge extends AbstractVerticle implements HealthCheckable {
 
         // create and add a new sink to the map
         // TODO: the AMQP client should be able to specify the format during link attachment
-        SinkBridgeEndpoint<?, ?> sink = new AmqpSinkBridgeEndpoint<>(this.vertx, this.amqpBridgeConfig,
+        SinkBridgeEndpoint<?, ?> sink = new AmqpSinkBridgeEndpoint<>(this.vertx, this.bridgeConfig,
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
 
         sink.closeHandler(s -> {
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfig.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfig.java
deleted file mode 100644
index c362cf8..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfig.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Copyright 2016, 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.amqp;
-
-import io.strimzi.kafka.bridge.config.BridgeConfig;
-import io.strimzi.kafka.bridge.config.KafkaConfig;
-
-import java.util.Map;
-
-/**
- * Bridge configuration for AMQP support
- */
-public class AmqpBridgeConfig extends BridgeConfig<AmqpConfig> {
-
-    /**
-     * Constructor
-     *
-     * @param kafkaConfig Kafka related configuration
-     * @param amqpConfig AMQP related configuration
-     */
-    private AmqpBridgeConfig(KafkaConfig kafkaConfig, AmqpConfig amqpConfig) {
-        super(kafkaConfig);
-        this.endpointConfig = amqpConfig;
-    }
-
-    /**
-     * Loads all AMQP bridge configuration parameters from a related map
-     *
-     * @param map map from which loading configuration parameters
-     * @return AMQP bridge related configuration
-     */
-    public static AmqpBridgeConfig fromMap(Map<String, Object> map) {
-        KafkaConfig kafkaConfig = KafkaConfig.fromMap(map);
-        AmqpConfig amqpConfig = AmqpConfig.fromMap(map);
-
-        return new AmqpBridgeConfig(kafkaConfig, amqpConfig);
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpoint.java
index 6e6a9fe..2816970 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpoint.java
@@ -10,6 +10,7 @@ import io.strimzi.kafka.bridge.Endpoint;
 import io.strimzi.kafka.bridge.QoSEndpoint;
 import io.strimzi.kafka.bridge.SinkBridgeEndpoint;
 import io.strimzi.kafka.bridge.SinkTopicSubscription;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.strimzi.kafka.bridge.tracker.SimpleOffsetTracker;
 import io.vertx.core.AsyncResult;
@@ -50,9 +51,9 @@ public class AmqpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
     // sender link for handling outgoing message
     private ProtonSender sender;
 
-    public AmqpSinkBridgeEndpoint(Vertx vertx, AmqpBridgeConfig bridgeConfigProperties,
+    public AmqpSinkBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfig,
                                   EmbeddedFormat format, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
-        super(vertx, bridgeConfigProperties, format, keyDeserializer, valueDeserializer);
+        super(vertx, bridgeConfig, format, keyDeserializer, valueDeserializer);
     }
 
     @Override
@@ -78,8 +79,7 @@ public class AmqpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
     public void handle(Endpoint<?> endpoint) {
 
         ProtonLink<?> link = (ProtonLink<?>) endpoint.get();
-        AmqpConfig amqpConfigProperties =
-                (AmqpConfig) this.bridgeConfigProperties.getEndpointConfig();
+        AmqpConfig amqpConfig = (AmqpConfig) this.bridgeConfig.getAmqpConfig();
 
         // Note: This is only called once for each instance
         if (!(link instanceof ProtonSender)) {
@@ -88,7 +88,7 @@ public class AmqpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
         try {
 
             if (this.converter == null) {
-                this.converter = (MessageConverter<K, V, Message, Collection<Message>>) AmqpBridge.instantiateConverter(amqpConfigProperties.getMessageConverter());
+                this.converter = (MessageConverter<K, V, Message, Collection<Message>>) AmqpBridge.instantiateConverter(amqpConfig.getMessageConverter());
             }
 
             this.sender = (ProtonSender) link;
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSourceBridgeEndpoint.java
index 87123d7..e1d86aa 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSourceBridgeEndpoint.java
@@ -8,6 +8,7 @@ package io.strimzi.kafka.bridge.amqp;
 import io.strimzi.kafka.bridge.EmbeddedFormat;
 import io.strimzi.kafka.bridge.Endpoint;
 import io.strimzi.kafka.bridge.SourceBridgeEndpoint;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
@@ -40,9 +41,9 @@ public class AmqpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
     // receiver link for handling incoming message
     private Map<String, ProtonReceiver> receivers;
 
-    public AmqpSourceBridgeEndpoint(Vertx vertx, AmqpBridgeConfig bridgeConfigProperties,
+    public AmqpSourceBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfig,
                                     EmbeddedFormat format, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
-        super(vertx, bridgeConfigProperties, format, keySerializer, valueSerializer);
+        super(vertx, bridgeConfig, format, keySerializer, valueSerializer);
         this.receivers = new HashMap<>();
     }
 
@@ -64,8 +65,7 @@ public class AmqpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
     public void handle(Endpoint<?> endpoint) {
 
         ProtonLink<?> link = (ProtonLink<?>) endpoint.get();
-        AmqpConfig amqpConfigProperties =
-                (AmqpConfig) this.bridgeConfigProperties.getEndpointConfig();
+        AmqpConfig amqpConfig = (AmqpConfig) this.bridgeConfig.getAmqpConfig();
 
         if (!(link instanceof ProtonReceiver)) {
             throw new IllegalArgumentException("This Proton link must be a receiver");
@@ -73,7 +73,7 @@ public class AmqpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
 
         if (this.converter == null) {
             try {
-                this.converter = (MessageConverter<K, V, Message, Collection<Message>>) AmqpBridge.instantiateConverter(amqpConfigProperties.getMessageConverter());
+                this.converter = (MessageConverter<K, V, Message, Collection<Message>>) AmqpBridge.instantiateConverter(amqpConfig.getMessageConverter());
             } catch (AmqpErrorConditionException e) {
                 AmqpBridge.detachWithError(link, e.toCondition());
                 return;
@@ -101,12 +101,12 @@ public class AmqpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
         if (receiver.getRemoteQoS() == ProtonQoS.AT_MOST_ONCE) {
             // sender settle mode is SETTLED (so AT_MOST_ONCE QoS), we assume Apache Kafka
             // no problem in throughput terms so use prefetch due to no ack from Kafka server
-            receiver.setPrefetch(amqpConfigProperties.getFlowCredit());
+            receiver.setPrefetch(amqpConfig.getFlowCredit());
         } else {
             // sender settle mode is UNSETTLED (or MIXED) (so AT_LEAST_ONCE QoS).
             // Thanks to the ack from Kafka server we can modulate flow control
             receiver.setPrefetch(0)
-                    .flow(amqpConfigProperties.getFlowCredit());
+                    .flow(amqpConfig.getFlowCredit());
         }
 
         receiver.open();
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 e81bcea..2ef1a9d 100644
--- a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
+++ b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
@@ -5,22 +5,36 @@
 
 package io.strimzi.kafka.bridge.config;
 
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import io.strimzi.kafka.bridge.amqp.AmqpConfig;
+import io.strimzi.kafka.bridge.http.HttpConfig;
+
 /**
  * Bridge configuration properties
- * @param <T>   type for configuring specific properties on the endpoint side
  */
-public abstract class BridgeConfig<T> {
+public class BridgeConfig extends AbstractConfig {
 
-    protected KafkaConfig kafkaConfig;
-    protected T endpointConfig;
+    public static final String BRIDGE_CONFIG_PREFIX = "bridge.";
+
+    private KafkaConfig kafkaConfig;
+    private AmqpConfig amqpConfig;
+    private HttpConfig httpConfig;
 
     /**
      * Constructor
-     *
+     * 
+     * @param config bridge common configuration parameters map
      * @param kafkaConfig Kafka related configuration
+     * @param amqpConfig AMQP endpoint related configuration
+     * @param httpConfig HTTP endpoint related configuration
      */
-    public BridgeConfig(KafkaConfig kafkaConfig) {
+    private BridgeConfig(Map<String, Object> config, KafkaConfig kafkaConfig, AmqpConfig amqpConfig, HttpConfig httpConfig) {
+        super(config);
         this.kafkaConfig = kafkaConfig;
+        this.amqpConfig = amqpConfig;
+        this.httpConfig = httpConfig;
     }
 
     /**
@@ -31,17 +45,43 @@ public abstract class BridgeConfig<T> {
     }
 
     /**
-     * @return the endpoint configuration
+     * @return the AMQP endpoint related configuration
+     */
+    public AmqpConfig getAmqpConfig() {
+        return this.amqpConfig;
+    }
+
+    /**
+     * @return the HTTP endpoint related configuration
      */
-    public T getEndpointConfig() {
-        return this.endpointConfig;
+    public HttpConfig getHttpConfig() {
+        return this.httpConfig;
+    }
+
+    /**
+     * Loads the entire bridge configuration parameters from a related map
+     *
+     * @param map map from which loading configuration parameters
+     * @return overall bridge configuration
+     */
+    public static BridgeConfig fromMap(Map<String, Object> map) {
+        KafkaConfig kafkaConfig = KafkaConfig.fromMap(map);
+        AmqpConfig amqpConfig = AmqpConfig.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, amqpConfig, httpConfig);
     }
 
     @Override
     public String toString() {
         return "BridgeConfig(" +
-                "kafkaConfig=" + this.kafkaConfig +
-                ",endpointConfig=" + this.endpointConfig +
+                "config=" + this.config +
+                ",kafkaConfig=" + this.kafkaConfig +
+                ",amqpConfig=" + this.amqpConfig +
+                ",httpConfig=" + this.httpConfig +
                 ")";
     }
 }
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 2b692ab..1d175f1 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -12,6 +12,7 @@ import io.strimzi.kafka.bridge.HealthCheckable;
 import io.strimzi.kafka.bridge.HealthChecker;
 import io.strimzi.kafka.bridge.SinkBridgeEndpoint;
 import io.strimzi.kafka.bridge.SourceBridgeEndpoint;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
 import io.vertx.core.AbstractVerticle;
 import io.vertx.core.Future;
@@ -40,7 +41,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
 
     private static final Logger log = LoggerFactory.getLogger(HttpBridge.class);
 
-    private final HttpBridgeConfig httpBridgeConfig;
+    private final BridgeConfig bridgeConfig;
 
     private HttpServer httpServer;
 
@@ -56,10 +57,10 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
     /**
      * Constructor
      *
-     * @param httpBridgeConfig bridge configuration for HTTP support
+     * @param bridgeConfig bridge configuration
      */
-    public HttpBridge(HttpBridgeConfig httpBridgeConfig) {
-        this.httpBridgeConfig = httpBridgeConfig;
+    public HttpBridge(BridgeConfig bridgeConfig) {
+        this.bridgeConfig = bridgeConfig;
     }
 
     private void bindHttpServer(Future<Void> startFuture) {
@@ -72,7 +73,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
                     if (httpServerAsyncResult.succeeded()) {
                         log.info("HTTP-Kafka Bridge started and listening on port {}", httpServerAsyncResult.result().actualPort());
                         log.info("HTTP-Kafka Bridge bootstrap servers {}",
-                                this.httpBridgeConfig.getKafkaConfig().getConfig()
+                                this.bridgeConfig.getKafkaConfig().getConfig()
                                         .get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)
                         );
 
@@ -166,8 +167,8 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
 
     private HttpServerOptions httpServerOptions() {
         HttpServerOptions httpServerOptions = new HttpServerOptions();
-        httpServerOptions.setHost(this.httpBridgeConfig.getEndpointConfig().getHost());
-        httpServerOptions.setPort(this.httpBridgeConfig.getEndpointConfig().getPort());
+        httpServerOptions.setHost(this.bridgeConfig.getHttpConfig().getHost());
+        httpServerOptions.setPort(this.bridgeConfig.getHttpConfig().getPort());
         return httpServerOptions;
     }
 
@@ -187,7 +188,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
         JsonObject body = routingContext.getBodyAsJson();
         EmbeddedFormat format = EmbeddedFormat.from(body.getString("format", "binary"));
 
-        final SinkBridgeEndpoint sink = new HttpSinkBridgeEndpoint<>(this.vertx, this.httpBridgeConfig, this.httpBridgeContext,
+        final SinkBridgeEndpoint sink = new HttpSinkBridgeEndpoint<>(this.vertx, this.bridgeConfig, this.httpBridgeContext,
                 format, new ByteArrayDeserializer(), new ByteArrayDeserializer());
 
         sink.closeHandler(s -> {
@@ -307,7 +308,7 @@ public class HttpBridge extends AbstractVerticle implements HealthCheckable {
 
         try {
             if (source == null) {
-                source = new HttpSourceBridgeEndpoint<>(this.vertx, this.httpBridgeConfig,
+                source = new HttpSourceBridgeEndpoint<>(this.vertx, this.bridgeConfig,
                         contentTypeToFormat(contentType), new ByteArraySerializer(), new ByteArraySerializer());
 
                 source.closeHandler(s -> {
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfig.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfig.java
deleted file mode 100644
index ac6f39e..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfig.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * Copyright 2018, 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.config.BridgeConfig;
-import io.strimzi.kafka.bridge.config.KafkaConfig;
-
-import java.util.Map;
-
-/**
- * Bridge configuration for HTTP support
- */
-public class HttpBridgeConfig extends BridgeConfig<HttpConfig> {
-
-    /**
-     * Constructor
-     *
-     * @param kafkaConfig Kafka related configuration
-     * @param httpConfig HTTP related configuration
-     */
-    public HttpBridgeConfig(KafkaConfig kafkaConfig, HttpConfig httpConfig) {
-        super(kafkaConfig);
-        this.endpointConfig = httpConfig;
-    }
-
-    /**
-     * Loads all HTTP bridge configuration parameters from a related map
-     *
-     * @param map map from which loading configuration parameters
-     * @return HTTP bridge related configuration
-     */
-    public static HttpBridgeConfig fromMap(Map<String, Object> map) {
-        KafkaConfig kafkaConfig = KafkaConfig.fromMap(map);
-        HttpConfig httpConfig = HttpConfig.fromMap(map);
-
-        return new HttpBridgeConfig(kafkaConfig, httpConfig);
-    }
-}
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 c31bffa..0c0a382 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -11,6 +11,7 @@ import io.strimzi.kafka.bridge.EmbeddedFormat;
 import io.strimzi.kafka.bridge.Endpoint;
 import io.strimzi.kafka.bridge.SinkBridgeEndpoint;
 import io.strimzi.kafka.bridge.SinkTopicSubscription;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.strimzi.kafka.bridge.http.converter.HttpBinaryMessageConverter;
 import io.strimzi.kafka.bridge.http.converter.HttpJsonMessageConverter;
@@ -51,9 +52,9 @@ public class HttpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
 
     private HttpBridgeContext httpBridgeContext;
 
-    HttpSinkBridgeEndpoint(Vertx vertx, HttpBridgeConfig httpBridgeConfigProperties, HttpBridgeContext context,
+    HttpSinkBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfig, HttpBridgeContext context,
                            EmbeddedFormat format, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer) {
-        super(vertx, httpBridgeConfigProperties, format, keyDeserializer, valueDeserializer);
+        super(vertx, bridgeConfig, format, keyDeserializer, valueDeserializer);
         this.httpBridgeContext = context;
     }
 
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 a7899c9..1356ccf 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -10,6 +10,7 @@ import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.EmbeddedFormat;
 import io.strimzi.kafka.bridge.Endpoint;
 import io.strimzi.kafka.bridge.SourceBridgeEndpoint;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
 import io.strimzi.kafka.bridge.http.converter.HttpBinaryMessageConverter;
 import io.strimzi.kafka.bridge.http.converter.HttpJsonMessageConverter;
@@ -35,9 +36,9 @@ public class HttpSourceBridgeEndpoint<K, V> extends SourceBridgeEndpoint<K, V> {
 
     private MessageConverter<K, V, Buffer, Buffer> messageConverter;
 
-    public HttpSourceBridgeEndpoint(Vertx vertx, HttpBridgeConfig bridgeConfigProperties,
+    public HttpSourceBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfig,
                                     EmbeddedFormat format, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
-        super(vertx, bridgeConfigProperties, format, keySerializer, valueSerializer);
+        super(vertx, bridgeConfig, format, keySerializer, valueSerializer);
     }
 
     @Override


Test patch:
diff --git a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
index 705a810..2bce02f 100644
--- a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
@@ -9,6 +9,7 @@ import io.strimzi.kafka.bridge.KafkaClusterTestBase;
 import io.strimzi.kafka.bridge.amqp.converter.AmqpDefaultMessageConverter;
 import io.strimzi.kafka.bridge.amqp.converter.AmqpJsonMessageConverter;
 import io.strimzi.kafka.bridge.amqp.converter.AmqpRawMessageConverter;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
 import io.strimzi.kafka.bridge.converter.DefaultDeserializer;
@@ -93,14 +94,14 @@ class AmqpBridgeTest extends KafkaClusterTestBase {
     private Vertx vertx;
     private AmqpBridge bridge;
 
-    private AmqpBridgeConfig bridgeConfigProperties;
+    private BridgeConfig bridgeConfig;
 
     @BeforeEach
     void before(VertxTestContext context) {
         this.vertx = Vertx.vertx();
 
-        this.bridgeConfigProperties = AmqpBridgeConfig.fromMap(config);
-        this.bridge = new AmqpBridge(this.bridgeConfigProperties);
+        this.bridgeConfig = BridgeConfig.fromMap(config);
+        this.bridge = new AmqpBridge(this.bridgeConfig);
 
         vertx.deployVerticle(this.bridge, context.succeeding(id -> context.completeNow()));
     }
@@ -624,7 +625,7 @@ class AmqpBridgeTest extends KafkaClusterTestBase {
                         consume.flag();
                     }
                 })
-                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfig.getAmqpConfig().getFlowCredit()).open();
 
                 ProtonSender sender = connection.createSender(null);
                 sender.open();
@@ -696,7 +697,7 @@ class AmqpBridgeTest extends KafkaClusterTestBase {
                         consume.flag();
                     }
                 })
-                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfig.getAmqpConfig().getFlowCredit()).open();
             } else {
                 context.failNow(ar.cause());
             }
@@ -766,7 +767,7 @@ class AmqpBridgeTest extends KafkaClusterTestBase {
                         consume.flag();
                     }
                 })
-                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfig.getAmqpConfig().getFlowCredit()).open();
             } else {
                 context.failNow(ar.cause());
             }
@@ -836,7 +837,7 @@ class AmqpBridgeTest extends KafkaClusterTestBase {
                         consume.flag();
                     }
                 })
-                        .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                        .setPrefetch(this.bridgeConfig.getAmqpConfig().getFlowCredit()).open();
             } else {
                 context.failNow(ar.cause());
             }
@@ -859,7 +860,7 @@ class AmqpBridgeTest extends KafkaClusterTestBase {
                 connection.open();
 
                 ProtonReceiver receiver = connection.createReceiver(topic + "/group.id/my_group");
-                receiver.setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit())
+                receiver.setPrefetch(this.bridgeConfig.getAmqpConfig().getFlowCredit())
                         .open();
 
                 this.vertx.setTimer(2000, t -> {
@@ -875,7 +876,7 @@ class AmqpBridgeTest extends KafkaClusterTestBase {
                             noPartition.flag();
                         }
                     })
-                            .setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit()).open();
+                            .setPrefetch(this.bridgeConfig.getAmqpConfig().getFlowCredit()).open();
                 });
 
 
diff --git a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
index d2555b6..cd8f5d1 100644
--- a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
@@ -6,6 +6,7 @@
 package io.strimzi.kafka.bridge.amqp;
 
 import io.strimzi.kafka.bridge.SinkTopicSubscription;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
 import io.strimzi.kafka.bridge.converter.MessageConverter;
@@ -157,7 +158,7 @@ class AmqpSinkBridgeEndpointMockTest {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
         MockRecordProducer recordProducer = new MockRecordProducer(topic, 0, 0L);
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
 
@@ -224,7 +225,7 @@ class AmqpSinkBridgeEndpointMockTest {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
         MockRecordProducer recordProducer = new MockRecordProducer(topic, 0, 0L);
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
 
@@ -289,7 +290,7 @@ class AmqpSinkBridgeEndpointMockTest {
     @Test
     public <K, V> void address_badAddressNoGroupId() throws Exception {
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, "missing group id delimiter");
@@ -305,7 +306,7 @@ class AmqpSinkBridgeEndpointMockTest {
     @Test
     public <K, V> void address_badAddressEmptyTopic() throws Exception {
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, "/group.id/blah");
@@ -322,7 +323,7 @@ class AmqpSinkBridgeEndpointMockTest {
     public <K, V> void address_badAddressEmptyGroup() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic + "/group.id/");
@@ -339,7 +340,7 @@ class AmqpSinkBridgeEndpointMockTest {
     public <K, V> void filters_nonIntegerPartitionFilter() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic + "/group.id/blah");
@@ -360,7 +361,7 @@ class AmqpSinkBridgeEndpointMockTest {
     public <K, V> void filters_nonLongOffsetFilter() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic + "/group.id/blah");
@@ -382,7 +383,7 @@ class AmqpSinkBridgeEndpointMockTest {
     public <K, V> void filters_negativeIntegerPartitionFilter() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic + "/group.id/blah");
@@ -407,7 +408,7 @@ class AmqpSinkBridgeEndpointMockTest {
     public <K, V> void filters_negativeLongOffsetFilter() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic + "/group.id/blah");
@@ -428,7 +429,7 @@ class AmqpSinkBridgeEndpointMockTest {
     public <K, V> void filters_offsetFilterButNoPartitionFilter() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
         ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic + "/group.id/blah");
@@ -447,8 +448,8 @@ class AmqpSinkBridgeEndpointMockTest {
     @Test
     public <K, V> void config_NoSuchConverterClass() throws AmqpErrorConditionException {
         Vertx vertx = Vertx.vertx();
-        AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
-        config.getEndpointConfig().setMessageConverter("foo.bar.Baz");
+        BridgeConfig config = BridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
+        config.getAmqpConfig().setMessageConverter("foo.bar.Baz");
         AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, config,
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
 
@@ -465,8 +466,8 @@ class AmqpSinkBridgeEndpointMockTest {
     @Test
     public <K, V> void config_ConverterWrongType() throws AmqpErrorConditionException {
         Vertx vertx = Vertx.vertx();
-        AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
-        config.getEndpointConfig().setMessageConverter("java.util.HashSet");
+        BridgeConfig config = BridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
+        config.getAmqpConfig().setMessageConverter("java.util.HashSet");
         AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, config,
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
@@ -508,8 +509,8 @@ class AmqpSinkBridgeEndpointMockTest {
     @Test
     public <K, V> void config_ConverterNoDefaultConstructor() throws AmqpErrorConditionException {
         Vertx vertx = Vertx.vertx();
-        AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
-        config.getEndpointConfig().setMessageConverter(NoNullaryCtor.class.getName());
+        BridgeConfig config = BridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
+        config.getAmqpConfig().setMessageConverter(NoNullaryCtor.class.getName());
         AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, config,
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
@@ -552,8 +553,8 @@ class AmqpSinkBridgeEndpointMockTest {
     @Test
     public <K, V> void config_ConverterDefaultConstructorThrows() throws AmqpErrorConditionException {
         Vertx vertx = Vertx.vertx();
-        AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
-        config.getEndpointConfig().setMessageConverter(CtorThrows.class.getName());
+        BridgeConfig config = BridgeConfig.fromMap(AmqpSinkBridgeEndpointMockTest.config);
+        config.getAmqpConfig().setMessageConverter(CtorThrows.class.getName());
         AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, config,
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
@@ -577,7 +578,7 @@ class AmqpSinkBridgeEndpointMockTest {
     public <K, V> void partitionsForFails() throws Exception {
         String topic = "my_topic";
         Vertx vertx = Vertx.vertx();
-        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, AmqpBridgeConfig.fromMap(config),
+        AmqpSinkBridgeEndpoint<K, V> endpoint = (AmqpSinkBridgeEndpoint) new AmqpSinkBridgeEndpoint<>(vertx, BridgeConfig.fromMap(config),
                 EmbeddedFormat.JSON, new StringDeserializer(), new ByteArrayDeserializer());
         endpoint.open();
 
diff --git a/src/test/java/io/strimzi/kafka/bridge/example/AmqpBridgeServer.java b/src/test/java/io/strimzi/kafka/bridge/example/AmqpBridgeServer.java
index 11f6449..8d1bfd0 100644
--- a/src/test/java/io/strimzi/kafka/bridge/example/AmqpBridgeServer.java
+++ b/src/test/java/io/strimzi/kafka/bridge/example/AmqpBridgeServer.java
@@ -6,7 +6,7 @@
 package io.strimzi.kafka.bridge.example;
 
 import io.strimzi.kafka.bridge.amqp.AmqpBridge;
-import io.strimzi.kafka.bridge.amqp.AmqpBridgeConfig;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.vertx.core.Vertx;
 
 import java.io.IOException;
@@ -23,9 +23,9 @@ public class AmqpBridgeServer {
         Vertx vertx = Vertx.vertx();
 
         Map<String, Object> config = new HashMap<>();
-        AmqpBridgeConfig bridgeConfigProperties = AmqpBridgeConfig.fromMap(config);
-
-        AmqpBridge bridge = new AmqpBridge(bridgeConfigProperties);
+        BridgeConfig bridgeConfig = BridgeConfig.fromMap(config);
+        
+        AmqpBridge bridge = new AmqpBridge(bridgeConfig);
 
         vertx.deployVerticle(bridge);
 
diff --git a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
index d84d0cd..ab33fce 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
@@ -9,6 +9,7 @@ import io.netty.handler.codec.http.HttpResponseStatus;
 import io.strimzi.kafka.bridge.BridgeContentType;
 import io.strimzi.kafka.bridge.KafkaClusterTestBase;
 import io.strimzi.kafka.bridge.amqp.AmqpConfig;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.config.KafkaConsumerConfig;
 import io.strimzi.kafka.bridge.http.model.HttpBridgeError;
@@ -88,7 +89,7 @@ class HttpBridgeTest extends KafkaClusterTestBase {
     private HttpBridge httpBridge;
     private WebClient client;
 
-    private HttpBridgeConfig bridgeConfigProperties;
+    private BridgeConfig bridgeConfig;
 
     //HTTP methods with configured Response timeout
     private HttpRequest<Buffer> postRequest(String requestURI) {
@@ -109,8 +110,8 @@ class HttpBridgeTest extends KafkaClusterTestBase {
     @BeforeEach
     void before(VertxTestContext context) {
         this.vertx = Vertx.vertx();
-        this.bridgeConfigProperties = HttpBridgeConfig.fromMap(config);
-        this.httpBridge = new HttpBridge(this.bridgeConfigProperties);
+        this.bridgeConfig = BridgeConfig.fromMap(config);
+        this.httpBridge = new HttpBridge(this.bridgeConfig);
 
         vertx.deployVerticle(this.httpBridge, context.succeeding(id -> context.completeNow()));