Repository: strimzi/strimzi-kafka-bridge
Pull request #131 (2019-03-26 15:52:16)

Removed SpringBoot dependency
Removed SpringBoot dependency
Refactored configuration classes to deal with env vars for configuration
Refactored bridges core classes and tests for dealing with new
configuration

ppatierno: The Travis CI has to be revisited completely for building, running tests and pushing the image to Docker.

Patch:
diff --git a/pom.xml b/pom.xml
index 7a70d4f..745ab9f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -16,7 +16,6 @@
 		<log4j.version>1.2.17</log4j.version>
 		<junit.version>4.12</junit.version>
 		<vertx.version>3.6.2</vertx.version>
-		<spring-boot.version>1.4.2.RELEASE</spring-boot.version>
 		<debezium.version>0.8.2</debezium.version>
 	</properties>
 
@@ -31,16 +30,6 @@
 			<artifactId>vertx-kafka-client</artifactId>
 			<version>${vertx.version}</version>
 		</dependency>
-		<dependency>
-		  <groupId>org.springframework.boot</groupId>
-		  <artifactId>spring-boot</artifactId>
-		  <version>${spring-boot.version}</version>
-		</dependency>
-		<dependency>
-		  <groupId>org.springframework.boot</groupId>
-		  <artifactId>spring-boot-autoconfigure</artifactId>
-		  <version>${spring-boot.version}</version>
-		</dependency>
 		<dependency>
 			<groupId>org.slf4j</groupId>
 			<artifactId>slf4j-log4j12</artifactId>
@@ -108,17 +97,27 @@
 	<build>
 		<plugins>
 			<plugin>
-				<groupId>org.springframework.boot</groupId>
-				<artifactId>spring-boot-maven-plugin</artifactId>
-				<version>${spring-boot.version}</version>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-assembly-plugin</artifactId>
 				<executions>
 					<execution>
-						<goals>
-							<goal>repackage</goal>
-						</goals>
+						<id>build-strimzi-kafka-bridge</id>
 						<configuration>
+							<archive>
+								<manifest>
+									<mainClass>io.strimzi.kafka.bridge.Application</mainClass>
+								</manifest>
+							</archive>
+							<descriptorRefs>
+								<descriptorRef>jar-with-dependencies</descriptorRef>
+							</descriptorRefs>
 							<finalName>kafka-bridge</finalName>
+							<appendAssemblyId>false</appendAssemblyId>
 						</configuration>
+						<phase>package</phase>
+						<goals>
+							<goal>single</goal>
+						</goals>
 					</execution>
 				</executions>
 			</plugin>
diff --git a/src/main/java/io/strimzi/kafka/bridge/Application.java b/src/main/java/io/strimzi/kafka/bridge/Application.java
index 1632f1e..007602b 100644
--- a/src/main/java/io/strimzi/kafka/bridge/Application.java
+++ b/src/main/java/io/strimzi/kafka/bridge/Application.java
@@ -17,38 +17,29 @@
 package io.strimzi.kafka.bridge;
 
 import io.strimzi.kafka.bridge.amqp.AmqpBridge;
+import io.strimzi.kafka.bridge.amqp.AmqpBridgeConfig;
 import io.strimzi.kafka.bridge.http.HttpBridge;
+import io.strimzi.kafka.bridge.http.HttpBridgeConfig;
 import io.vertx.core.Vertx;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
 
 /**
- * AMQP - Apache Kafka bridge main application class
+ * Apache Kafka bridge main application class
  */
-@SpringBootApplication
 public class Application {
 
     private static final Logger log = LoggerFactory.getLogger(Application.class);
 
-    private final Vertx vertx = Vertx.vertx();
-
-    @Autowired
-    private AmqpBridge amqpBridge;
-
-    @Autowired
-    private HttpBridge httpBridge;
-
-    @PostConstruct
-    public void start() {
+    public static void main(String[] args) {
+        Vertx vertx = Vertx.vertx();
 
-        this.vertx.deployVerticle(this.amqpBridge, done -> {
+        AmqpBridgeConfig amqpBridgeConfig = AmqpBridgeConfig.fromMap(System.getenv());
+        AmqpBridge amqpBridge = new AmqpBridge(amqpBridgeConfig);
+        HttpBridgeConfig httpBridgeConfig = HttpBridgeConfig.fromMap(System.getenv());
+        HttpBridge httpBridge = new HttpBridge(httpBridgeConfig);
 
+        vertx.deployVerticle(amqpBridge, done -> {
             if (done.succeeded()) {
                 log.debug("AMQP verticle instance deployed [{}]", done.result());
             } else {
@@ -56,28 +47,12 @@ public class Application {
             }
         });
 
-        this.vertx.deployVerticle(this.httpBridge, done -> {
-
+        vertx.deployVerticle(httpBridge, done -> {
             if (done.succeeded()) {
                 log.debug("HTTP verticle instance deployed [{}]", done.result());
             } else {
                 log.debug("Failed to deploy HTTP verticle instance", done.cause());
             }
         });
-
-    }
-
-    @PreDestroy
-    public void stop() {
-
-        this.vertx.close(done -> {
-            if (done.failed()) {
-                log.error("Could not shut down Kafka bridge cleanly", done.cause());
-            }
-        });
-    }
-
-    public static void main(String[] args) {
-        SpringApplication.run(Application.class, args);
     }
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
index c62d0a3..aeacbf6 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SinkBridgeEndpoint.java
@@ -16,8 +16,8 @@
 
 package io.strimzi.kafka.bridge;
 
-import io.strimzi.kafka.bridge.config.BridgeConfigProperties;
-import io.strimzi.kafka.bridge.config.KafkaConfigProperties;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
+import io.strimzi.kafka.bridge.config.KafkaConfig;
 import io.strimzi.kafka.bridge.tracker.OffsetTracker;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Future;
@@ -53,7 +53,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
 
     protected Vertx vertx;
 
-    protected BridgeConfigProperties bridgeConfigProperties;
+    protected BridgeConfig bridgeConfigProperties;
 
     private Handler<BridgeEndpoint> closeHandler;
 
@@ -103,7 +103,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
      * @param vertx	Vert.x instance
      * @param bridgeConfigProperties	Bridge configuration
      */
-    public SinkBridgeEndpoint(Vertx vertx, BridgeConfigProperties bridgeConfigProperties) {
+    public SinkBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfigProperties) {
         this.vertx = vertx;
         this.bridgeConfigProperties = bridgeConfigProperties;
     }
@@ -137,7 +137,7 @@ public abstract class SinkBridgeEndpoint<K, V> implements BridgeEndpoint {
     protected void initConsumer(boolean shouldAttachBatchHandler) {
 
         // create a consumer
-        KafkaConfigProperties consumerConfig = this.bridgeConfigProperties.getKafkaConfigProperties();
+        KafkaConfig consumerConfig = this.bridgeConfigProperties.getKafkaConfig();
         Properties props = new Properties();
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, consumerConfig.getBootstrapServers());
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, consumerConfig.getConsumerConfig().getKeyDeserializer());
diff --git a/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java b/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
index f80853f..38e47f3 100644
--- a/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/SourceBridgeEndpoint.java
@@ -16,7 +16,7 @@
 
 package io.strimzi.kafka.bridge;
 
-import io.strimzi.kafka.bridge.config.BridgeConfigProperties;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
@@ -38,7 +38,7 @@ public abstract class SourceBridgeEndpoint implements BridgeEndpoint {
 
     protected Vertx vertx;
 
-    protected BridgeConfigProperties bridgeConfigProperties;
+    protected BridgeConfig bridgeConfigProperties;
 
     private Handler<BridgeEndpoint> closeHandler;
 
@@ -51,7 +51,7 @@ public abstract class SourceBridgeEndpoint implements BridgeEndpoint {
      * @param vertx	Vert.x instance
      * @param bridgeConfigProperties	Bridge configuration
      */
-    public SourceBridgeEndpoint(Vertx vertx, BridgeConfigProperties bridgeConfigProperties) {
+    public SourceBridgeEndpoint(Vertx vertx, BridgeConfig bridgeConfigProperties) {
         this.vertx = vertx;
         this.bridgeConfigProperties = bridgeConfigProperties;
     }
@@ -92,17 +92,17 @@ public abstract class SourceBridgeEndpoint implements BridgeEndpoint {
     public void open() {
 
         Properties props = new Properties();
-        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bridgeConfigProperties.getKafkaConfigProperties().getBootstrapServers());
-        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfigProperties().getProducerConfig().getKeySerializer());
-        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfigProperties().getProducerConfig().getValueSerializer());
-        props.put(ProducerConfig.ACKS_CONFIG, this.bridgeConfigProperties.getKafkaConfigProperties().getProducerConfig().getAcks());
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bridgeConfigProperties.getKafkaConfig().getBootstrapServers());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfig().getProducerConfig().getKeySerializer());
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfig().getProducerConfig().getValueSerializer());
+        props.put(ProducerConfig.ACKS_CONFIG, this.bridgeConfigProperties.getKafkaConfig().getProducerConfig().getAcks());
 
         this.producerUnsettledMode = KafkaProducer.create(this.vertx, props);
 
         props.clear();
-        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bridgeConfigProperties.getKafkaConfigProperties().getBootstrapServers());
-        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfigProperties().getProducerConfig().getKeySerializer());
-        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfigProperties().getProducerConfig().getValueSerializer());
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.bridgeConfigProperties.getKafkaConfig().getBootstrapServers());
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfig().getProducerConfig().getKeySerializer());
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, this.bridgeConfigProperties.getKafkaConfig().getProducerConfig().getValueSerializer());
         props.put(ProducerConfig.ACKS_CONFIG, "0");
 
         this.producerSettledMode = KafkaProducer.create(this.vertx, props);
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 1cc0979..d651b81 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridge.java
@@ -40,8 +40,6 @@ import org.apache.qpid.proton.amqp.Symbol;
 import org.apache.qpid.proton.amqp.transport.ErrorCondition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
 
 import java.io.File;
 import java.util.HashMap;
@@ -51,7 +49,6 @@ import java.util.Map;
  * Main bridge class listening for connections
  * and handling AMQP senders and receivers
  */
-@Component
 public class AmqpBridge extends AbstractVerticle {
 	
 	private static final Logger log = LoggerFactory.getLogger(AmqpBridge.class);
@@ -75,11 +72,12 @@ public class AmqpBridge extends AbstractVerticle {
 	public static final String AMQP_ERROR_KAFKA_COMMIT = "io.strimzi:kafka-commit";
 	public static final String AMQP_ERROR_CONFIGURATION = "io.strimzi:configuration";
 
-	
 	// AMQP filters
 	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;
+
 	// container-id needed for working in "client" mode
 	private static final String CONTAINER_ID = "kafka-bridge-service";
 
@@ -92,14 +90,16 @@ public class AmqpBridge extends AbstractVerticle {
 	// endpoints for handling incoming and outcoming messages
 	private Map<ProtonConnection, ConnectionEndpoint> endpoints;
 
-	private AmqpBridgeConfigProperties bridgeConfigProperties;
-
 	// if the bridge is ready to handle requests
 	private boolean isReady = false;
 
-	@Autowired
-	public void setBridgeConfigProperties(AmqpBridgeConfigProperties bridgeConfigProperties) {
-		this.bridgeConfigProperties = bridgeConfigProperties;
+    /**
+     * Constructor
+     *
+     * @param amqpBridgeConfig bridge configuration for AMQP support
+     */
+	public AmqpBridge(AmqpBridgeConfig amqpBridgeConfig) {
+	    this.amqpBridgeConfig = amqpBridgeConfig;
 	}
 
 	/**
@@ -119,7 +119,7 @@ public class AmqpBridge extends AbstractVerticle {
 
 						log.info("AMQP-Kafka Bridge started and listening on port {}", ar.result().actualPort());
 						log.info("Kafka bootstrap servers {}",
-								this.bridgeConfigProperties.getKafkaConfigProperties().getBootstrapServers());
+								this.amqpBridgeConfig.getKafkaConfig().getBootstrapServers());
 
 						this.isReady = true;
 						this.startHealthServer();
@@ -145,8 +145,8 @@ public class AmqpBridge extends AbstractVerticle {
 
 		this.client = ProtonClient.create(this.vertx);
 
-		String host = this.bridgeConfigProperties.getEndpointConfigProperties().getHost();
-		int port = this.bridgeConfigProperties.getEndpointConfigProperties().getPort();
+		String host = this.amqpBridgeConfig.getEndpointConfig().getHost();
+		int port = this.amqpBridgeConfig.getEndpointConfig().getPort();
 
 		ProtonClientOptions options = this.createClientOptions();
 
@@ -161,7 +161,7 @@ public class AmqpBridge extends AbstractVerticle {
 
 				log.info("AMQP-Kafka Bridge started and connected in client mode to {}:{}", host, port);
 				log.info("Kafka bootstrap servers {}",
-						this.bridgeConfigProperties.getKafkaConfigProperties().getBootstrapServers());
+						this.amqpBridgeConfig.getKafkaConfig().getBootstrapServers());
 
 				this.isReady = true;
 
@@ -181,7 +181,7 @@ public class AmqpBridge extends AbstractVerticle {
 
 		this.endpoints = new HashMap<>();
 
-		AmqpMode mode = this.bridgeConfigProperties.getEndpointConfigProperties().getMode();
+		AmqpMode mode = this.amqpBridgeConfig.getEndpointConfig().getMode();
 		log.info("AMQP-Kafka Bridge configured in {} mode", mode);
 		if (mode == AmqpMode.SERVER) {
 			this.bindAmqpServer(startFuture);
@@ -255,11 +255,11 @@ public class AmqpBridge extends AbstractVerticle {
 	private ProtonServerOptions createServerOptions(){
 
 		ProtonServerOptions options = new ProtonServerOptions();
-		options.setHost(this.bridgeConfigProperties.getEndpointConfigProperties().getHost());
-		options.setPort(this.bridgeConfigProperties.getEndpointConfigProperties().getPort());
+		options.setHost(this.amqpBridgeConfig.getEndpointConfig().getHost());
+		options.setPort(this.amqpBridgeConfig.getEndpointConfig().getPort());
 
-		if (this.bridgeConfigProperties.getEndpointConfigProperties().getCertDir() != null && this.bridgeConfigProperties.getEndpointConfigProperties().getCertDir().length() > 0) {
-			String certDir = this.bridgeConfigProperties.getEndpointConfigProperties().getCertDir();
+		if (this.amqpBridgeConfig.getEndpointConfig().getCertDir() != null && this.amqpBridgeConfig.getEndpointConfig().getCertDir().length() > 0) {
+			String certDir = this.amqpBridgeConfig.getEndpointConfig().getCertDir();
 			log.info("Enabling SSL configuration for AMQP with TLS certificates from {}", certDir);
 			options.setSsl(true)
 					.setPemTrustOptions(new PemTrustOptions()
@@ -283,8 +283,8 @@ public class AmqpBridge extends AbstractVerticle {
 		options.setConnectTimeout(1000);
 		options.setReconnectAttempts(-1).setReconnectInterval(1000); // reconnect forever, every 1000 millisecs
 
-		if (this.bridgeConfigProperties.getEndpointConfigProperties().getCertDir() != null && this.bridgeConfigProperties.getEndpointConfigProperties().getCertDir().length() > 0) {
-			String certDir = this.bridgeConfigProperties.getEndpointConfigProperties().getCertDir();
+		if (this.amqpBridgeConfig.getEndpointConfig().getCertDir() != null && this.amqpBridgeConfig.getEndpointConfig().getCertDir().length() > 0) {
+			String certDir = this.amqpBridgeConfig.getEndpointConfig().getCertDir();
 			log.info("Enabling SSL configuration for AMQP with TLS certificates from {}", certDir);
 			options.setSsl(true)
 					.addEnabledSaslMechanism("EXTERNAL")
@@ -319,7 +319,7 @@ public class AmqpBridge extends AbstractVerticle {
 			this.processOpenSender(connection, sender);
 		});
 
-		if (this.bridgeConfigProperties.getEndpointConfigProperties().getMode() == AmqpMode.CLIENT) {
+		if (this.amqpBridgeConfig.getEndpointConfig().getMode() == AmqpMode.CLIENT) {
 			connection.open();
 		}
 	}
@@ -421,7 +421,7 @@ public class AmqpBridge extends AbstractVerticle {
 		SourceBridgeEndpoint source = endpoint.getSource();
 		// the source endpoint is only one, handling more AMQP receiver links internally
 		if (source == null) {
-			source = new AmqpSourceBridgeEndpoint(this.vertx, this.bridgeConfigProperties);
+			source = new AmqpSourceBridgeEndpoint(this.vertx, this.amqpBridgeConfig);
 
 			source.closeHandler(s -> {
 				endpoint.setSource(null);
@@ -444,7 +444,7 @@ public class AmqpBridge extends AbstractVerticle {
 		log.info("Remote receiver attached {}", sender.getName());
 		
 		// create and add a new sink to the map
-		SinkBridgeEndpoint<?,?> sink = new AmqpSinkBridgeEndpoint<>(this.vertx, this.bridgeConfigProperties);
+		SinkBridgeEndpoint<?,?> sink = new AmqpSinkBridgeEndpoint<>(this.vertx, this.amqpBridgeConfig);
 
 		sink.closeHandler(s -> {
 			this.endpoints.get(connection).getSinks().remove(s);
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfig.java
similarity index 50%
rename from src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfigProperties.java
rename to src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfig.java
index 9ad3c1d..8937fbb 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfigProperties.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeConfig.java
@@ -16,17 +16,25 @@
 
 package io.strimzi.kafka.bridge.amqp;
 
-import io.strimzi.kafka.bridge.config.BridgeConfigProperties;
-import org.springframework.stereotype.Component;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
+import io.strimzi.kafka.bridge.config.KafkaConfig;
+
+import java.util.Map;
 
 /**
- * Bridge configuration properties for AMQP support
+ * Bridge configuration for AMQP support
  */
-@Component
-public class AmqpBridgeConfigProperties extends BridgeConfigProperties<AmqpConfigProperties> {
+public class AmqpBridgeConfig extends BridgeConfig<AmqpConfig> {
+
+    public AmqpBridgeConfig(KafkaConfig kafkaConfig, AmqpConfig amqpConfig) {
+        super(kafkaConfig);
+        this.endpointConfig = amqpConfig;
+    }
+
+    public static AmqpBridgeConfig fromMap(Map<String, String> map) {
+        KafkaConfig kafkaConfig = KafkaConfig.fromMap(map);
+        AmqpConfig amqpConfig = AmqpConfig.fromMap(map);
 
-    public AmqpBridgeConfigProperties() {
-        super();
-        this.endpointConfigProperties = new AmqpConfigProperties();
+        return new AmqpBridgeConfig(kafkaConfig, amqpConfig);
     }
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpConfig.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpConfig.java
new file mode 100644
index 0000000..d7e85ea
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpConfig.java
@@ -0,0 +1,160 @@
+/*
+ * Copyright 2016 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.amqp;
+
+import java.util.Map;
+
+/**
+ * AMQP related configuration
+ */
+public class AmqpConfig {
+
+    private static final String AMQP_MODE = "AMQP_MODE";
+    private static final String AMQP_HOST = "AMQP_HOST";
+    private static final String AMQP_PORT = "AMQP_PORT";
+    private static final String AMQP_FLOW_CREDIT = "AMQP_FLOW_CREDIT";
+    private static final String AMQP_MESSAGE_CONVERTER = "AMQP_MESSAGE_CONVERTER";
+    private static final String AMQP_CERT_DIR = "AMQP_CERT_DIR";
+
+    private static final String DEFAULT_AMQP_MODE = "SERVER";
+    private static final String DEFAULT_HOST = "0.0.0.0";
+    private static final int DEFAULT_PORT = 5672;
+    private static final int DEFAULT_FLOW_CREDIT = 1024;
+    private static final String DEFAULT_MESSAGE_CONVERTER = "io.strimzi.kafka.bridge.amqp.converter.AmqpDefaultMessageConverter";
+    private static final String DEFAULT_CERT_DIR = null;
+
+    private AmqpMode mode;
+    private int flowCredit;
+    private String host;
+    private int port;
+    private String messageConverter;
+    private String certDir;
+
+    /**
+     * Constructor
+     *
+     * @param mode the AMQP bridge working mode (client or server)
+     * @param flowCredit the AMQP receiver flow credit
+     * @param host the host for AMQP client (to connect) or server (to bind)
+     * @param port the port for AMQP client (to connect) or server (to bind)
+     * @param messageConverter the AMQP message converter
+     * @param certDir the directory with the TLS certificates files
+     */
+    public AmqpConfig(AmqpMode mode, int flowCredit, String host, int port, String messageConverter, String certDir) {
+        this.mode = mode;
+        this.flowCredit = flowCredit;
+        this.host = host;
+        this.port = port;
+        this.messageConverter = messageConverter;
+        this.certDir = certDir;
+    }
+
+    /**
+     * @return the AMQP bridge working mode (client or server)
+     */
+    public AmqpMode getMode() {
+        return this.mode;
+    }
+
+    /**
+     * @return the AMQP receiver flow credit
+     */
+    public int getFlowCredit() {
+        return this.flowCredit;
+    }
+
+    /**
+     * @return the host for AMQP client (to connect) or server (to bind)
+     */
+    public String getHost() {
+        return this.host;
+    }
+
+    /**
+     * @return the port for AMQP client (to connect) or server (to bind)
+     */
+    public int getPort() {
+        return this.port;
+    }
+
+    /**
+     * @return the AMQP message converter
+     */
+    public String getMessageConverter() {
+        return this.messageConverter;
+    }
+
+    /**
+     * Set the AMQP message converter
+     *
+     * @param messageConverter  AMQP message converter
+     * @return this instance for setter chaining
+     */
+    /* test */ AmqpConfig setMessageConverter(String messageConverter) {
+        this.messageConverter = messageConverter;
+        return this;
+    }
+
+    /**
+     * @return the directory with the TLS certificates files
+     */
+    public String getCertDir() {
+        return this.certDir;
+    }
+
+    /**
+     * Loads AMQP related configuration parameters from a related map
+     *
+     * @param map map from which loading configuration parameters
+     * @return AMQP related configuration
+     */
+    public static AmqpConfig fromMap(Map<String, String> map) {
+
+        AmqpMode mode = AmqpMode.from(map.getOrDefault(AmqpConfig.AMQP_MODE, AmqpConfig.DEFAULT_AMQP_MODE));
+
+        int flowCredit = AmqpConfig.DEFAULT_FLOW_CREDIT;
+        String flowCreditEnvVar = map.get(AmqpConfig.AMQP_FLOW_CREDIT);
+        if (flowCreditEnvVar != null) {
+            flowCredit = Integer.parseInt(flowCreditEnvVar);
+        }
+
+        String host = map.getOrDefault(AmqpConfig.AMQP_HOST, AmqpConfig.DEFAULT_HOST);
+
+        int port = AmqpConfig.DEFAULT_PORT;
+        String portEnvVar = map.get(AmqpConfig.AMQP_PORT);
+        if (portEnvVar != null) {
+            port = Integer.parseInt(portEnvVar);
+        }
+
+        String messageConverter = map.getOrDefault(AmqpConfig.AMQP_MESSAGE_CONVERTER, AmqpConfig.DEFAULT_MESSAGE_CONVERTER);
+        String certDir = map.getOrDefault(AmqpConfig.AMQP_CERT_DIR, AmqpConfig.DEFAULT_CERT_DIR);
+
+        return new AmqpConfig(mode, flowCredit, host, port, messageConverter, certDir);
+    }
+
+    @Override
+    public String toString() {
+        return "AmqpConfig(" +
+                "mode=" + this.mode +
+                ",flowCredit=" + this.flowCredit +
+                ",host=" + this.host +
+                ",port=" + this.port +
+                ",messageConverter=" + this.messageConverter +
+                ",certDir=" + this.certDir +
+                ")";
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpConfigProperties.java
deleted file mode 100644
index 2f7ce15..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpConfigProperties.java
+++ /dev/null
@@ -1,161 +0,0 @@
-/*
- * Copyright 2016 Red Hat Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.strimzi.kafka.bridge.amqp;
-
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
-
-/**
- * AMQP configuration properties
- */
-@Component
-@ConfigurationProperties(prefix = "amqp")
-public class AmqpConfigProperties {
-
-    private static final AmqpMode DEFAULT_AMQP_MODE = AmqpMode.SERVER;
-    private static final String DEFAULT_HOST = "0.0.0.0";
-    private static final int DEFAULT_PORT = 5672;
-    private static final int DEFAULT_FLOW_CREDIT = 1024;
-    private static final String DEFAULT_MESSAGE_CONVERTER = "io.strimzi.kafka.bridge.amqp.converter.AmqpDefaultMessageConverter";
-    private static final String DEFAULT_CERT_DIR = null;
-
-    private AmqpMode mode = DEFAULT_AMQP_MODE;
-    private int flowCredit = DEFAULT_FLOW_CREDIT;
-    private String host = DEFAULT_HOST;
-    private int port = DEFAULT_PORT;
-    private String messageConverter = DEFAULT_MESSAGE_CONVERTER;
-    private String certDir = DEFAULT_CERT_DIR;
-
-    /**
-     * Get the AMQP receiver flow credit
-     *
-     * @return
-     */
-    public int getFlowCredit() {
-        return this.flowCredit;
-    }
-
-    /**
-     * Set the AMQP receiver flow credit
-     *
-     * @param flowCredit    AMQP receiver flow credit
-     * @return  this instance for setter chaining
-     */
-    public AmqpConfigProperties setFlowCredit(int flowCredit) {
-        this.flowCredit = flowCredit;
-        return this;
-    }
-
-    /**
-     * Get the bridge working mode (client or server)
-     *
-     * @return
-     */
-    public AmqpMode getMode() {
-        return this.mode;
-    }
-
-    /**
-     * Set the bridge working mode
-     *
-     * @param mode  bridge working mode
-     * @return  this instance for setter chaining
-     */
-    public AmqpConfigProperties setMode(AmqpMode mode) {
-        this.mode = mode;
-        return this;
-    }
-
-    /**
-     * Get the host for AMQP client (to connect) or server (to bind)
-     *
-     * @return
-     */
-    public String getHost() {
-        return this.host;
-    }
-
-    /**
-     * Set the host for AMQP client (to connect) or server (to bind)
-     *
-     * @param host  AMQP host
-     * @return  this instance for setter chaining
-     */
-    public AmqpConfigProperties setHost(String host) {
-        this.host = host;
-        return this;
-    }
-
-    /**
-     * Get the port for AMQP client (to connect) or server (to bind)
-     *
-     * @return
-     */
-    public int getPort() {
-        return this.port;
-    }
-
-    /**
-     * Set the port for AMQP client (to connect) or server (to bind)
-     *
-     * @param port  AMQP port
-     * @return  this instance for setter chaining
-     */
-    public void setPort(int port) {
-        this.port = port;
-    }
-
-    /**
-     * Get the AMQP message converter
-     *
-     * @return
-     */
-    public String getMessageConverter() {
-        return this.messageConverter;
-    }
-
-    /**
-     * Set the AMQP message converter
-     *
-     * @param messageConverter  AMQP message converter
-     * @return  this instance for setter chaining
-     */
-    public AmqpConfigProperties setMessageConverter(String messageConverter) {
-        this.messageConverter = messageConverter;
-        return this;
-    }
-
-    /**
-     * Get the directory with the TLS certificates files
-     *
-     * @return
-     */
-    public String getCertDir() {
-        return this.certDir;
-    }
-
-    /**
-     * Set the directory with the TLS certificates files
-     *
-     * @param certDir  Path to the TLS certificate files
-     * @return  this instance for setter chaining
-     */
-    public AmqpConfigProperties setCertDir(String certDir) {
-        this.certDir = certDir;
-        return this;
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpMode.java b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpMode.java
index 45a3841..93733f7 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpMode.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpMode.java
@@ -21,8 +21,8 @@ package io.strimzi.kafka.bridge.amqp;
  */
 public enum AmqpMode {
 
-    CLIENT("client"),
-    SERVER("server");
+    CLIENT("CLIENT"),
+    SERVER("SERVER");
 
     private final String mode;
 
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 4606c01..946a0ab 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpoint.java
@@ -63,7 +63,7 @@ public class AmqpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
 	 * @param vertx	Vert.x instance
 	 * @param bridgeConfigProperties	Bridge configuration
 	 */
-	public AmqpSinkBridgeEndpoint(Vertx vertx, AmqpBridgeConfigProperties bridgeConfigProperties) {
+	public AmqpSinkBridgeEndpoint(Vertx vertx, AmqpBridgeConfig bridgeConfigProperties) {
 		super(vertx, bridgeConfigProperties);
 	}
 	
@@ -90,8 +90,8 @@ public class AmqpSinkBridgeEndpoint<K, V> extends SinkBridgeEndpoint<K, V> {
 	public void handle(Endpoint<?> endpoint) {
 
 		ProtonLink<?> link = (ProtonLink<?>) endpoint.get();
-		AmqpConfigProperties amqpConfigProperties =
-				(AmqpConfigProperties) this.bridgeConfigProperties.getEndpointConfigProperties();
+		AmqpConfig amqpConfigProperties =
+				(AmqpConfig) this.bridgeConfigProperties.getEndpointConfig();
 
 		// Note: This is only called once for each instance
 		if (!(link instanceof ProtonSender)) {
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 17da2fe..991cef9 100644
--- a/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/amqp/AmqpSourceBridgeEndpoint.java
@@ -55,7 +55,7 @@ public class AmqpSourceBridgeEndpoint extends SourceBridgeEndpoint {
 	 * @param vertx	Vert.x instance
 	 * @param bridgeConfigProperties	Bridge configuration
 	 */
-	public AmqpSourceBridgeEndpoint(Vertx vertx, AmqpBridgeConfigProperties bridgeConfigProperties) {
+	public AmqpSourceBridgeEndpoint(Vertx vertx, AmqpBridgeConfig bridgeConfigProperties) {
 		super(vertx, bridgeConfigProperties);
 		this.receivers = new HashMap<>();
 	}
@@ -76,8 +76,8 @@ public class AmqpSourceBridgeEndpoint extends SourceBridgeEndpoint {
 	public void handle(Endpoint<?> endpoint) {
 
 		ProtonLink<?> link = (ProtonLink<?>) endpoint.get();
-		AmqpConfigProperties amqpConfigProperties =
-				(AmqpConfigProperties) this.bridgeConfigProperties.getEndpointConfigProperties();
+		AmqpConfig amqpConfigProperties =
+				(AmqpConfig) this.bridgeConfigProperties.getEndpointConfig();
 
 		if (!(link instanceof ProtonReceiver)) {
 			throw new IllegalArgumentException("This Proton link must be a receiver");
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
new file mode 100644
index 0000000..90ab0d4
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright 2016 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.config;
+
+/**
+ * Bridge configuration properties
+ * @param <T>   type for configuring specific properties on the endpoint side
+ */
+public abstract class BridgeConfig<T> {
+
+    protected KafkaConfig kafkaConfig;
+    protected T endpointConfig;
+
+    public BridgeConfig(KafkaConfig kafkaConfigProperties) {
+        this.kafkaConfig = kafkaConfigProperties;
+    }
+
+    /**
+     * @return the Kafka related configuration
+     */
+    public KafkaConfig getKafkaConfig() {
+        return this.kafkaConfig;
+    }
+
+    /**
+     * @return the endpoint configuration
+     */
+    public T getEndpointConfig() {
+        return this.endpointConfig;
+    }
+
+    @Override
+    public String toString() {
+        return "BridgeConfig(" +
+                "kafkaConfig=" + this.kafkaConfig +
+                ",endpointConfig=" + this.endpointConfig +
+                ")";
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfigProperties.java
deleted file mode 100644
index 7a3dcbe..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfigProperties.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Copyright 2016 Red Hat Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.strimzi.kafka.bridge.config;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-/**
- * Bridge configuration properties
- * @param <T>   type for configuring specific properties on the endpoint side
- */
-@Component
-public class BridgeConfigProperties<T> {
-
-    protected KafkaConfigProperties kafkaConfigProperties;
-    protected T endpointConfigProperties;
-
-    public BridgeConfigProperties() {
-        this.kafkaConfigProperties = new KafkaConfigProperties();
-    }
-
-    /**
-     * Get the Kafka configuration
-     *
-     * @return
-     */
-    public KafkaConfigProperties getKafkaConfigProperties() {
-        return this.kafkaConfigProperties;
-    }
-
-    /**
-     * Set the Kafka configuration
-     *
-     * @param kafkaConfigProperties  Kafka configuration
-     * @return  this instance for setter chaining
-     */
-    @Autowired
-    public BridgeConfigProperties setKafkaConfigProperties(KafkaConfigProperties kafkaConfigProperties) {
-        this.kafkaConfigProperties = kafkaConfigProperties;
-        return this;
-    }
-
-    /**
-     * Get the endpoint configuration
-     *
-     * @return
-     */
-    public T getEndpointConfigProperties() {
-        return this.endpointConfigProperties;
-    }
-
-    /**
-     * Set the endpoint configuration
-     *
-     * @param endpointConfigProperties  endpoint configuration
-     * @return  this instance for setter chaining
-     */
-    @Autowired
-    public BridgeConfigProperties setEndpointConfigProperties(T endpointConfigProperties) {
-        this.endpointConfigProperties = endpointConfigProperties;
-        return this;
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/KafkaConfig.java b/src/main/java/io/strimzi/kafka/bridge/config/KafkaConfig.java
new file mode 100644
index 0000000..09360f0
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/config/KafkaConfig.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2016 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.config;
+
+import java.util.Map;
+
+/**
+ * Apache Kafka related configuration
+ */
+public class KafkaConfig {
+
+    private static final String KAFKA_BOOTSTRAP_SERVERS = "KAFKA_BOOTSTRAP_SERVERS";
+
+    private static final String DEFAULT_BOOTSTRAP_SERVERS = "localhost:9092";
+
+    private String bootstrapServers;
+    private KafkaProducerConfig producerConfig;
+    private KafkaConsumerConfig consumerConfig;
+
+    /**
+     * Constructor
+     *
+     * @param bootstrapServers the Kafka bootstrap servers
+     * @param producerConfig the Kafka producer configuration
+     * @param consumerConfig the Kafka consumer configuration
+     */
+    public KafkaConfig(String bootstrapServers,
+                       KafkaProducerConfig producerConfig,
+                       KafkaConsumerConfig consumerConfig) {
+        this.bootstrapServers = bootstrapServers;
+        this.producerConfig = producerConfig;
+        this.consumerConfig = consumerConfig;
+    }
+
+    /**
+     * @return the Kafka bootstrap servers
+     */
+    public String getBootstrapServers() {
+        return this.bootstrapServers;
+    }
+
+    /**
+     * @return the Kafka producer configuration
+     */
+    public KafkaProducerConfig getProducerConfig() {
+        return this.producerConfig;
+    }
+
+    /**
+     * @return the Kafka consumer configuration
+     */
+    public KafkaConsumerConfig getConsumerConfig() {
+        return this.consumerConfig;
+    }
+
+    /**
+     * Loads Kafka related configuration parameters from a related map
+     *
+     * @param map map from which loading configuration parameters
+     * @return Kafka related configuration
+     */
+    public static KafkaConfig fromMap(Map<String, String> map) {
+
+        String bootstrapServers = map.getOrDefault(KafkaConfig.KAFKA_BOOTSTRAP_SERVERS, KafkaConfig.DEFAULT_BOOTSTRAP_SERVERS);
+        KafkaProducerConfig producerConfig = KafkaProducerConfig.fromMap(map);
+        KafkaConsumerConfig consumerConfig = KafkaConsumerConfig.fromMap(map);
+
+        return new KafkaConfig(bootstrapServers, producerConfig, consumerConfig);
+    }
+
+    @Override
+    public String toString() {
+        return "KafkaConfig(" +
+                "bootstrapServers=" + this.bootstrapServers +
+                ",producerConfig=" + this.producerConfig +
+                ",consumerConfig=" + this.consumerConfig +
+                ")";
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/KafkaConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/config/KafkaConfigProperties.java
deleted file mode 100644
index 77b4939..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/config/KafkaConfigProperties.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Copyright 2016 Red Hat Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.strimzi.kafka.bridge.config;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
-
-/**
- * Apache Kafka configuration properties
- */
-@Component
-@ConfigurationProperties(prefix = "kafka")
-public class KafkaConfigProperties {
-
-    private static final String DEFAULT_BOOTSTRAP_SERVERS = "localhost:9092";
-
-    private String bootstrapServers = DEFAULT_BOOTSTRAP_SERVERS;
-
-    private KafkaProducerConfigProperties producerConfig = new KafkaProducerConfigProperties();
-    private KafkaConsumerConfigProperties consumerConfig = new KafkaConsumerConfigProperties();
-
-    /**
-     * Get the bootstrap servers
-     *
-     * @return
-     */
-    public String getBootstrapServers() {
-        return this.bootstrapServers;
-    }
-
-    /**
-     * Set the bootstrap servers
-     *
-     * @param bootstrapServers  bootstrap servers list
-     * @return  this instance for setter chaining
-     */
-    public KafkaConfigProperties setBootstrapServers(String bootstrapServers) {
-        this.bootstrapServers = bootstrapServers;
-        return this;
-    }
-
-    /**
-     * Get the Kafka producer configuration
-     *
-     * @return  Kafka producer configuration
-     */
-    public KafkaProducerConfigProperties getProducerConfig() {
-        return this.producerConfig;
-    }
-
-    /**
-     * Set the Kafka producer configuration
-     *
-     * @param producerConfig    Kafka producer configuration
-     * @return  this instance for setter chaining
-     */
-    @Autowired
-    public KafkaConfigProperties setProducerConfig(KafkaProducerConfigProperties producerConfig) {
-        this.producerConfig = producerConfig;
-        return this;
-    }
-
-    /**
-     * Get the Kafka consumer configuration
-     *
-     * @return  Kafka consumer configuration
-     */
-    public KafkaConsumerConfigProperties getConsumerConfig() {
-        return this.consumerConfig;
-    }
-
-    /**
-     * Set the Kafka consumer configuration
-     *
-     * @param consumerConfig    Kafka consumer configuration
-     * @return  this instance for setter chaining
-     */
-    @Autowired
-    public KafkaConfigProperties setConsumerConfig(KafkaConsumerConfigProperties consumerConfig) {
-        this.consumerConfig = consumerConfig;
-        return this;
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/KafkaConsumerConfig.java b/src/main/java/io/strimzi/kafka/bridge/config/KafkaConsumerConfig.java
new file mode 100644
index 0000000..fe3e7dc
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/config/KafkaConsumerConfig.java
@@ -0,0 +1,106 @@
+/*
+ * Copyright 2016 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.config;
+
+import java.util.Map;
+
+/**
+ * Apache Kafka consumer related configuration
+ */
+public class KafkaConsumerConfig {
+
+    private static final String KAFKA_CONSUMER_KEY_DESERIALIZER = "KAFKA_CONSUMER_KEY_DESERIALIZER";
+    private static final String KAFKA_CONSUMER_VALUE_DESERIALIZER = "KAFKA_CONSUMER_VALUE_DESERIALIZER";
+    private static final String KAFKA_CONSUMER_AUTO_OFFSET_RESET = "KAFKA_CONSUMER_AUTO_OFFSET_RESET";
+
+    private static final String DEFAULT_KEY_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";
+    private static final String DEFAULT_VALUE_DESERIALIZER = "org.apache.kafka.common.serialization.ByteArrayDeserializer";
+    private static final String DEFAULT_AUTO_OFFSET_RESET = "earliest";
+    private static final boolean DEFAULT_ENABLE_AUTO_COMMIT = false;
+
+    private String keyDeserializer;
+    private String valueDeserializer;
+    private String autoOffsetReset;
+    private boolean isEnableAutoCommit;
+
+    /**
+     * Constructor
+     *
+     * @param keyDeserializer the Key Deserializer class
+     * @param valueDeserializer the Value deserializer class
+     * @param autoOffsetReset the initial offset behavior
+     */
+    public KafkaConsumerConfig(String keyDeserializer, String valueDeserializer, String autoOffsetReset) {
+        this.keyDeserializer = keyDeserializer;
+        this.valueDeserializer = valueDeserializer;
+        this.autoOffsetReset = autoOffsetReset;
+        this.isEnableAutoCommit = DEFAULT_ENABLE_AUTO_COMMIT;
+    }
+
+    /**
+     * @return the Key Deserializer class
+     */
+    public String getKeyDeserializer() {
+        return this.keyDeserializer;
+    }
+
+    /**
+     * @return the Value deserializer class
+     */
+    public String getValueDeserializer() {
+        return this.valueDeserializer;
+    }
+
+    /**
+     * @return the initial offset behavior
+     */
+    public String getAutoOffsetReset() {
+        return this.autoOffsetReset;
+    }
+
+    /**
+     * @return if auto commit is enabled or not
+     */
+    public boolean isEnableAutoCommit() {
+        // enable.auto.commit isn't configurable
+        return this.isEnableAutoCommit;
+    }
+
+    /**
+     * Loads Kafka consumer related configuration parameters from a related map
+     *
+     * @param map map from which loading configuration parameters
+     * @return Kafka consumer related configuration
+     */
+    public static KafkaConsumerConfig fromMap(Map<String, String> map) {
+        String keyDeserializer = map.getOrDefault(KafkaConsumerConfig.KAFKA_CONSUMER_KEY_DESERIALIZER, KafkaConsumerConfig.DEFAULT_KEY_DESERIALIZER);
+        String valueDeserializer = map.getOrDefault(KafkaConsumerConfig.KAFKA_CONSUMER_VALUE_DESERIALIZER, KafkaConsumerConfig.DEFAULT_VALUE_DESERIALIZER);
+        String autoOffsetReset = map.getOrDefault(KafkaConsumerConfig.KAFKA_CONSUMER_AUTO_OFFSET_RESET, KafkaConsumerConfig.DEFAULT_AUTO_OFFSET_RESET);
+
+        return new KafkaConsumerConfig(keyDeserializer, valueDeserializer, autoOffsetReset);
+    }
+
+    @Override
+    public String toString() {
+        return "KafkaConsumerConfig(" +
+                "keyDeserializer=" + this.keyDeserializer +
+                ",valueDeserializer=" + this.valueDeserializer +
+                ",autoOffsetReset=" + this.autoOffsetReset +
+                ",isEnableAutoCommit=" + this.isEnableAutoCommit +
+                ")";
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/KafkaConsumerConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/config/KafkaConsumerConfigProperties.java
deleted file mode 100644
index b6c7564..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/config/KafkaConsumerConfigProperties.java
+++ /dev/null
@@ -1,108 +0,0 @@
-/*
- * Copyright 2016 Red Hat Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.strimzi.kafka.bridge.config;
-
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
-
-/**
- * Apache Kafka consumer configuration properties
- */
-@Component
-@ConfigurationProperties(prefix = "kafka.consumer")
-public class KafkaConsumerConfigProperties {
-
-    private static final String DEFAULT_KEY_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";
-    private static final String DEFAULT_VALUE_DESERIALIZER = "org.apache.kafka.common.serialization.ByteArrayDeserializer";
-    private static final String DEFAULT_AUTO_OFFSET_RESET = "earliest";
-    private static final boolean DEFAULT_ENABLE_AUTO_COMMIT = false;
-
-    private String keyDeserializer = DEFAULT_KEY_DESERIALIZER;
-    private String valueDeserializer = DEFAULT_VALUE_DESERIALIZER;
-    private String autoOffsetReset = DEFAULT_AUTO_OFFSET_RESET;
-    private boolean isEnableAutoCommit = DEFAULT_ENABLE_AUTO_COMMIT;
-
-    /**
-     * Get the Key Serializer class
-     *
-     * @return
-     */
-    public String getKeyDeserializer() {
-        return this.keyDeserializer;
-    }
-
-    /**
-     * Set the Key Serializer class
-     *
-     * @param keyDeserializer Key Serializer class
-     * @return  this instance for setter chaining
-     */
-    public KafkaConsumerConfigProperties setKeyDeserializer(String keyDeserializer) {
-        this.keyDeserializer = keyDeserializer;
-        return this;
-    }
-
-    /**
-     * Get the Value Serializer class
-     *
-     * @return
-     */
-    public String getValueDeserializer() {
-        return this.valueDeserializer;
-    }
-
-    /**
-     * Set the Value Serializer class
-     *
-     * @param valueDeserializer Value Serializer class
-     * @return  this instance for setter chaining
-     */
-    public KafkaConsumerConfigProperties setValueDeserializer(String valueDeserializer) {
-        this.valueDeserializer = valueDeserializer;
-        return this;
-    }
-
-    /**
-     * Get the initial offset behavior
-     *
-     * @return
-     */
-    public String getAutoOffsetReset() {
-        return this.autoOffsetReset;
-    }
-
-    /**
-     * Set the initial offset behavior
-     *
-     * @param autoOffsetReset initial offset behavior
-     * @return  this instance for setter chaining
-     */
-    public KafkaConsumerConfigProperties setAutoOffsetReset(String autoOffsetReset) {
-        this.autoOffsetReset = autoOffsetReset;
-        return this;
-    }
-
-    /**
-     * Enable auto commit on Kafka Consumer
-     *
-     * @return
-     */
-    public boolean isEnableAutoCommit() {
-        // enable.auto.commit isn't configurable
-        return this.isEnableAutoCommit;
-    }
-}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/KafkaProducerConfig.java b/src/main/java/io/strimzi/kafka/bridge/config/KafkaProducerConfig.java
new file mode 100644
index 0000000..039dc14
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/config/KafkaProducerConfig.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2016 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.config;
+
+import java.util.Map;
+
+/**
+ * Apache Kafka producer related configuration
+ */
+public class KafkaProducerConfig {
+
+    private static final String KAFKA_PRODUCER_KEY_SERIALIZER = "KAFKA_PRODUCER_KEY_SERIALIZER";
+    private static final String KAFKA_PRODUCER_VALUE_SERIALIZER = "KAFKA_PRODUCER_VALUE_SERIALIZER";
+    private static final String KAFKA_PRODUCER_ACKS = "KAFKA_PRODUCER_ACKS";
+
+    private static final String DEFAULT_KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";
+    private static final String DEFAULT_VALUE_SERIALIZER = "org.apache.kafka.common.serialization.ByteArraySerializer";
+    private static final String DEFAULT_ACKS = "1";
+
+    private String keySerializer;
+    private String valueSerializer;
+    private String acks;
+
+    /**
+     * Constructor
+     *
+     * @param keySerializer the Key Serializer class
+     * @param valueSerializer the Value Serializer class
+     * @param acks the acknowledgments behavior
+     */
+    public KafkaProducerConfig(String keySerializer, String valueSerializer, String acks) {
+        this.keySerializer = keySerializer;
+        this.valueSerializer = valueSerializer;
+        this.acks = acks;
+    }
+
+    /**
+     * @return the Key Serializer class
+     */
+    public String getKeySerializer() {
+        return this.keySerializer;
+    }
+
+    /**
+     * @return the Value Serializer class
+     */
+    public String getValueSerializer() {
+        return this.valueSerializer;
+    }
+
+    /**
+     * @return the acknowledgments behavior
+     */
+    public String getAcks() {
+        return this.acks;
+    }
+
+    /**
+     * Loads Kafka producer related configuration parameters from a related map
+     *
+     * @param map map from which loading configuration parameters
+     * @return Kafka producer related configuration
+     */
+    public static KafkaProducerConfig fromMap(Map<String, String> map) {
+        String keySerializer = map.getOrDefault(KafkaProducerConfig.KAFKA_PRODUCER_KEY_SERIALIZER, KafkaProducerConfig.DEFAULT_KEY_SERIALIZER);
+        String valueSerializer = map.getOrDefault(KafkaProducerConfig.KAFKA_PRODUCER_VALUE_SERIALIZER, KafkaProducerConfig.DEFAULT_VALUE_SERIALIZER);
+        String acks = map.getOrDefault(KafkaProducerConfig.KAFKA_PRODUCER_ACKS, KafkaProducerConfig.DEFAULT_ACKS);
+
+        return new KafkaProducerConfig(keySerializer, valueSerializer, acks);
+    }
+
+    @Override
+    public String toString() {
+        return "KafkaProducerConfig(" +
+                "keySerializer=" + this.keySerializer +
+                ",valueSerializer=" + this.valueSerializer +
+                ",acks=" + this.acks +
+                ")";
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/KafkaProducerConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/config/KafkaProducerConfigProperties.java
deleted file mode 100644
index 1078009..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/config/KafkaProducerConfigProperties.java
+++ /dev/null
@@ -1,96 +0,0 @@
-/*
- * Copyright 2016 Red Hat Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.strimzi.kafka.bridge.config;
-
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
-
-/**
- * Apache Kafka producer configuration properties
- */
-@Component
-@ConfigurationProperties(prefix = "kafka.producer")
-public class KafkaProducerConfigProperties {
-
-    private static final String DEFAULT_KEY_SERIALIZER = "org.apache.kafka.common.serialization.StringSerializer";
-    private static final String DEFAULT_VALUE_SERIALIZER = "org.apache.kafka.common.serialization.ByteArraySerializer";
-    private static final String DEFAULT_ACKS = "1";
-
-    private String keySerializer = DEFAULT_KEY_SERIALIZER;
-    private String valueSerializer = DEFAULT_VALUE_SERIALIZER;
-    private String acks = DEFAULT_ACKS;
-
-    /**
-     * Get the Key Serializer class
-     *
-     * @return
-     */
-    public String getKeySerializer() {
-        return this.keySerializer;
-    }
-
-    /**
-     * Set the Key Serializer class
-     *
-     * @param keySerializer Key Serializer class
-     * @return  this instance for setter chaining
-     */
-    public KafkaProducerConfigProperties setKeySerializer(String keySerializer) {
-        this.keySerializer = keySerializer;
-        return this;
-    }
-
-    /**
-     * Get the Value Serializer class
-     *
-     * @return
-     */
-    public String getValueSerializer() {
-        return this.valueSerializer;
-    }
-
-    /**
-     * Set the Value Serializer class
-     *
-     * @param valueSerializer Value Serializer class
-     * @return  this instance for setter chaining
-     */
-    public KafkaProducerConfigProperties setValueSerializer(String valueSerializer) {
-        this.valueSerializer = valueSerializer;
-        return this;
-    }
-
-    /**
-     * Get the acknowledgments behavior
-     *
-     * @return
-     */
-    public String getAcks() {
-        return this.acks;
-    }
-
-    /**
-     * Set the acknowledgments behavior
-     *
-     * @param acks  acknowledgments behavior (0, 1, all)
-     * @return  this instance for setter chaining
-     */
-    public KafkaProducerConfigProperties setAcks(String acks) {
-        this.acks = acks;
-        return this;
-    }
-}
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 d84aa4d..29fe0cf 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java
@@ -23,8 +23,6 @@ import io.vertx.core.Future;
 import io.vertx.core.http.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -33,24 +31,28 @@ import java.util.Map;
  * Main bridge class listening for connections
  * and handling HTTP requests.
  */
-@Component
 public class HttpBridge extends AbstractVerticle {
 
     private static final Logger log = LoggerFactory.getLogger(HttpBridge.class);
 
-    private HttpServer httpServer;
+    private final HttpBridgeConfig httpBridgeConfig;
 
-    private HttpBridgeConfigProperties bridgeConfigProperties;
+    private HttpServer httpServer;
 
     private Map<HttpConnection, SourceBridgeEndpoint> httpSourceEndpoints;
 
     private Map<String, SinkBridgeEndpoint> httpSinkEndpoints;
 
-    private boolean isReady;
+    // if the bridge is ready to handle requests
+    private boolean isReady = false;
 
-    @Autowired
-    public void setBridgeConfigProperties(HttpBridgeConfigProperties httpBridgeConfigProperties) {
-        this.bridgeConfigProperties = httpBridgeConfigProperties;
+    /**
+     * Constructor
+     *
+     * @param httpBridgeConfig bridge configuration for HTTP support
+     */
+    public HttpBridge(HttpBridgeConfig httpBridgeConfig) {
+        this.httpBridgeConfig = httpBridgeConfig;
     }
 
     private void bindHttpServer(Future<Void> startFuture) {
@@ -63,7 +65,7 @@ public class HttpBridge extends AbstractVerticle {
                     if (httpServerAsyncResult.succeeded()) {
                         log.info("HTTP-Kafka Bridge started and listening on port {}", httpServerAsyncResult.result().actualPort());
                         log.info("Kafka bootstrap servers {}",
-                                this.bridgeConfigProperties.getKafkaConfigProperties().getBootstrapServers());
+                                this.httpBridgeConfig.getKafkaConfig().getBootstrapServers());
 
                         this.isReady = true;
                         startFuture.complete();
@@ -126,8 +128,8 @@ public class HttpBridge extends AbstractVerticle {
 
     private HttpServerOptions httpServerOptions() {
         HttpServerOptions httpServerOptions = new HttpServerOptions();
-        httpServerOptions.setHost(this.bridgeConfigProperties.getEndpointConfigProperties().getHost());
-        httpServerOptions.setPort(this.bridgeConfigProperties.getEndpointConfigProperties().getPort());
+        httpServerOptions.setHost(this.httpBridgeConfig.getEndpointConfig().getHost());
+        httpServerOptions.setPort(this.httpBridgeConfig.getEndpointConfig().getPort());
         return httpServerOptions;
     }
 
@@ -141,7 +143,7 @@ public class HttpBridge extends AbstractVerticle {
                 SourceBridgeEndpoint source = this.httpSourceEndpoints.get(httpServerRequest.connection());
 
                 if (source == null) {
-                    source = new HttpSourceBridgeEndpoint(this.vertx, this.bridgeConfigProperties);
+                    source = new HttpSourceBridgeEndpoint(this.vertx, this.httpBridgeConfig);
                     source.closeHandler(s -> {
                         this.httpSourceEndpoints.remove(httpServerRequest.connection());
                     });
@@ -154,7 +156,7 @@ public class HttpBridge extends AbstractVerticle {
 
             //create a sink endpoint and initialize consumer
             case CREATE:
-                SinkBridgeEndpoint<?,?> sink = new HttpSinkBridgeEndpoint<>(this.vertx, this.bridgeConfigProperties);
+                SinkBridgeEndpoint<?,?> sink = new HttpSinkBridgeEndpoint<>(this.vertx, this.httpBridgeConfig);
                 sink.closeHandler(s -> {
                     this.httpSinkEndpoints.remove(sink);
                 });
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfig.java
similarity index 50%
rename from src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfigProperties.java
rename to src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfig.java
index 7cfc36d..7470e0d 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfigProperties.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeConfig.java
@@ -16,17 +16,25 @@
 
 package io.strimzi.kafka.bridge.http;
 
-import io.strimzi.kafka.bridge.config.BridgeConfigProperties;
-import org.springframework.stereotype.Component;
+import io.strimzi.kafka.bridge.config.BridgeConfig;
+import io.strimzi.kafka.bridge.config.KafkaConfig;
+
+import java.util.Map;
 
 /**
- * Bridge configuration properties for HTTP support
+ * Bridge configuration for HTTP support
  */
-@Component
-public class HttpBridgeConfigProperties extends BridgeConfigProperties<HttpConfigProperties> {
+public class HttpBridgeConfig extends BridgeConfig<HttpConfig> {
+
+    public HttpBridgeConfig(KafkaConfig kafkaConfig, HttpConfig httpConfig) {
+        super(kafkaConfig);
+        this.endpointConfig = httpConfig;
+    }
+
+    public static HttpBridgeConfig fromMap(Map<String, String> map) {
+        KafkaConfig kafkaConfig = KafkaConfig.fromMap(map);
+        HttpConfig httpConfig = HttpConfig.fromMap(map);
 
-    public HttpBridgeConfigProperties(){
-        super();
-        this.endpointConfigProperties = new HttpConfigProperties();
+        return new HttpBridgeConfig(kafkaConfig, httpConfig);
     }
 }
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
new file mode 100644
index 0000000..a536b26
--- /dev/null
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpConfig.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright 2018 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.strimzi.kafka.bridge.http;
+
+import java.util.Map;
+
+/**
+ * HTTP related configuration
+ */
+public class HttpConfig {
+
+    private static final String HTTP_HOST = "HTTP_HOST";
+    private static final String HTTP_PORT = "HTTP_PORT";
+
+    private static final String DEFAULT_HOST = "0.0.0.0";
+    private static final int DEFAULT_PORT = 8080;
+
+    private String host;
+    private int port;
+
+    /**
+     * Constructor
+     *
+     * @param host the host for HTTP server (to bind)
+     * @param port the port for HTTP server (to bind)
+     */
+    public HttpConfig(String host, int port) {
+        this.host = host;
+        this.port = port;
+    }
+
+    /**
+     * @return the host for HTTP server (to bind)
+     */
+    public String getHost() {
+        return host;
+    }
+
+    /**
+     * @return the port for HTTP server (to bind)
+     */
+    public int getPort() {
+        return port;
+    }
+
+    /**
+     * Loads HTTP related configuration parameters from a related map
+     *
+     * @param map map from which loading configuration parameters
+     * @return HTTP related configuration
+     */
+    public static HttpConfig fromMap(Map<String, String> map) {
+
+        String host = map.getOrDefault(HttpConfig.HTTP_HOST, HttpConfig.DEFAULT_HOST);
+
+        int port = HttpConfig.DEFAULT_PORT;
+        String portEnvVar = map.get(HttpConfig.HTTP_PORT);
+        if (portEnvVar != null) {
+            port = Integer.parseInt(portEnvVar);
+        }
+
+        return new HttpConfig(host, port);
+    }
+}
diff --git a/src/main/java/io/strimzi/kafka/bridge/http/HttpConfigProperties.java b/src/main/java/io/strimzi/kafka/bridge/http/HttpConfigProperties.java
deleted file mode 100644
index b8dce28..0000000
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpConfigProperties.java
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * Copyright 2018 Red Hat Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.strimzi.kafka.bridge.http;
-
-import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
-
-/**
- * HTTP configuration properties
- */
-@Component
-@ConfigurationProperties(prefix = "http")
-public class HttpConfigProperties {
-
-    private static final String DEFAULT_HOST = "0.0.0.0";
-    private static final int DEFAULT_PORT = 8080;
-
-    private String host = DEFAULT_HOST;
-    private int port = DEFAULT_PORT;
-
-    public String getHost() {
-        return host;
-    }
-
-    public void setHost(String host) {
-        this.host = host;
-    }
-
-    public int getPort() {
-        return port;
-    }
-
-    public void setPort(int port) {
-        this.port = port;
-    }
-
-}
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 a899afb..610c5c9 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java
@@ -49,7 +49,7 @@ public class HttpSinkBridgeEndpoint<V, K> extends SinkBridgeEndpoint<V, K> {
 
     private MessageConverter messageConverter;
 
-    HttpSinkBridgeEndpoint(Vertx vertx, HttpBridgeConfigProperties httpBridgeConfigProperties) {
+    HttpSinkBridgeEndpoint(Vertx vertx, HttpBridgeConfig httpBridgeConfigProperties) {
         super(vertx, httpBridgeConfigProperties);
     }
 
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 9532f38..b6c77b3 100644
--- a/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
+++ b/src/main/java/io/strimzi/kafka/bridge/http/HttpSourceBridgeEndpoint.java
@@ -31,7 +31,7 @@ import io.vertx.kafka.client.producer.RecordMetadata;
 public class HttpSourceBridgeEndpoint extends SourceBridgeEndpoint {
     private MessageConverter messageConverter;
 
-    public HttpSourceBridgeEndpoint(Vertx vertx, HttpBridgeConfigProperties bridgeConfigProperties) {
+    public HttpSourceBridgeEndpoint(Vertx vertx, HttpBridgeConfig bridgeConfigProperties) {
         super(vertx, bridgeConfigProperties);
     }
 


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 6d0d355..a32dcb3 100644
--- a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpBridgeTest.java
@@ -72,6 +72,8 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 	
 	private static final Logger log = LoggerFactory.getLogger(AmqpBridgeTest.class);
 
+	private static Map<String, String> envVars = new HashMap<>();
+
 	private static final String BRIDGE_HOST = "localhost";
 	private static final int BRIDGE_PORT = 5672;
 	
@@ -83,15 +85,15 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 	private Vertx vertx;
 	private AmqpBridge bridge;
 
-	private AmqpBridgeConfigProperties bridgeConfigProperties = new AmqpBridgeConfigProperties();
+	private AmqpBridgeConfig bridgeConfigProperties;
 	
 	@Before
 	public void before(TestContext context) {
 		
 		this.vertx = Vertx.vertx();
 
-		this.bridge = new AmqpBridge();
-		this.bridge.setBridgeConfigProperties(this.bridgeConfigProperties);
+		this.bridgeConfigProperties = AmqpBridgeConfig.fromMap(envVars);
+		this.bridge = new AmqpBridge(this.bridgeConfigProperties);
 
 		this.vertx.deployVerticle(this.bridge, context.asyncAssertSuccess());
 	}
@@ -595,7 +597,7 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 						async.complete();
 					}
 				})
-				.setPrefetch(this.bridgeConfigProperties.getEndpointConfigProperties().getFlowCredit())
+				.setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit())
 				.open();
 
 				ProtonSender sender = connection.createSender(null);
@@ -663,7 +665,7 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 						async.complete();
 					}
 				})
-				.setPrefetch(this.bridgeConfigProperties.getEndpointConfigProperties().getFlowCredit())
+				.setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit())
 				.open();
 			} else {
 				context.fail(ar.cause());
@@ -731,7 +733,7 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 						async.complete();
 					}
 				})
-				.setPrefetch(this.bridgeConfigProperties.getEndpointConfigProperties().getFlowCredit())
+				.setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit())
 				.open();
 			} else {
 				context.fail(ar.cause());
@@ -800,7 +802,7 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 						async.complete();
 					}
 				})
-				.setPrefetch(this.bridgeConfigProperties.getEndpointConfigProperties().getFlowCredit())
+				.setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit())
 				.open();
 			} else {
 				context.fail(ar.cause());
@@ -823,7 +825,7 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 				connection.open();
 
 				ProtonReceiver receiver = connection.createReceiver(topic + "/group.id/my_group");
-				receiver.setPrefetch(this.bridgeConfigProperties.getEndpointConfigProperties().getFlowCredit())
+				receiver.setPrefetch(this.bridgeConfigProperties.getEndpointConfig().getFlowCredit())
 						.open();
 
 				this.vertx.setTimer(2000, t -> {
@@ -839,7 +841,7 @@ public class AmqpBridgeTest extends KafkaClusterTestBase {
 							async.complete();
 						}
 					})
-					.setPrefetch(this.bridgeConfigProperties.getEndpointConfigProperties().getFlowCredit())
+					.setPrefetch(this.bridgeConfigProperties.getEndpointConfig().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 34219ba..1940c74 100644
--- a/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/amqp/AmqpSinkBridgeEndpointMockTest.java
@@ -54,6 +54,8 @@ import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.Mockito.*;
 
 public class AmqpSinkBridgeEndpointMockTest {
+
+	private static Map<String, String> envVars = new HashMap<>();
 	
 	class MockRecordProducer {
 		
@@ -144,7 +146,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
 		MockRecordProducer recordProducer = new MockRecordProducer(topic, 0, 0L);
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		
 		// Create a mock for the sender
@@ -210,7 +212,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
 		MockRecordProducer recordProducer = new MockRecordProducer(topic, 0, 0L);
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		
 		// Create a mock for the sender
@@ -274,7 +276,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	@Test
 	public <K, V> void address_badAddressNoGroupId() throws Exception {
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, "missing group id delimiter");
 		// Call handle()
@@ -289,7 +291,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	@Test
 	public <K, V> void address_badAddressEmptyTopic() throws Exception {
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, "/group.id/blah");
 		// Call handle()
@@ -305,7 +307,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	public <K, V> void address_badAddressEmptyGroup() throws Exception {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic+"/group.id/");
 		// Call handle()
@@ -321,7 +323,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	public <K, V> void filters_nonIntegerPartitionFilter() throws Exception {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic+"/group.id/blah");
 		// Call handle()
@@ -341,7 +343,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	public <K, V> void filters_nonLongOffsetFilter() throws Exception {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic+"/group.id/blah");
 		// Call handle()
@@ -362,7 +364,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	public <K, V> void filters_negativeIntegerPartitionFilter() throws Exception {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic+"/group.id/blah");
 		// Call handle()
@@ -386,7 +388,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	public <K, V> void filters_negativeLongOffsetFilter() throws Exception {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic+"/group.id/blah");
 		// Call handle()
@@ -406,7 +408,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	public <K, V> void filters_offsetFilterButNoPartitionFilter() throws Exception {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, topic+"/group.id/blah");
 		// Call handle()
@@ -424,8 +426,8 @@ public class AmqpSinkBridgeEndpointMockTest {
 	@Test
 	public <K, V> void config_NoSuchConverterClass() throws AmqpErrorConditionException {
 		Vertx vertx = Vertx.vertx();
-		AmqpBridgeConfigProperties config = new AmqpBridgeConfigProperties();
-		config.getEndpointConfigProperties().setMessageConverter("foo.bar.Baz");
+		AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(envVars);
+		config.getEndpointConfig().setMessageConverter("foo.bar.Baz");
 		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, config);
 		
 		endpoint.open();
@@ -441,8 +443,8 @@ public class AmqpSinkBridgeEndpointMockTest {
 	@Test
 	public <K, V> void config_ConverterWrongType() throws AmqpErrorConditionException {
 		Vertx vertx = Vertx.vertx();
-		AmqpBridgeConfigProperties config = new AmqpBridgeConfigProperties();
-		config.getEndpointConfigProperties().setMessageConverter("java.util.HashSet");
+		AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(envVars);
+		config.getEndpointConfig().setMessageConverter("java.util.HashSet");
 		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, config);
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, "");
@@ -478,8 +480,8 @@ public class AmqpSinkBridgeEndpointMockTest {
 	@Test
 	public <K, V> void config_ConverterNoDefaultConstructor() throws AmqpErrorConditionException {
 		Vertx vertx = Vertx.vertx();
-		AmqpBridgeConfigProperties config = new AmqpBridgeConfigProperties();
-		config.getEndpointConfigProperties().setMessageConverter(NoNullaryCtor.class.getName());
+		AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(envVars);
+		config.getEndpointConfig().setMessageConverter(NoNullaryCtor.class.getName());
 		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, config);
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, "");
@@ -516,8 +518,8 @@ public class AmqpSinkBridgeEndpointMockTest {
 	@Test
 	public <K, V> void config_ConverterDefaultConstructorThrows() throws AmqpErrorConditionException {
 		Vertx vertx = Vertx.vertx();
-		AmqpBridgeConfigProperties config = new AmqpBridgeConfigProperties();
-		config.getEndpointConfigProperties().setMessageConverter(CtorThrows.class.getName());
+		AmqpBridgeConfig config = AmqpBridgeConfig.fromMap(envVars);
+		config.getEndpointConfig().setMessageConverter(CtorThrows.class.getName());
 		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, config);
 		endpoint.open();
 		ProtonSender mockSender = mockSender(ProtonQoS.AT_MOST_ONCE, "");
@@ -539,7 +541,7 @@ public class AmqpSinkBridgeEndpointMockTest {
 	public <K, V> void partitionsForFails() throws Exception {
 		String topic = "my_topic";
 		Vertx vertx = Vertx.vertx();
-		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, new AmqpBridgeConfigProperties());
+		AmqpSinkBridgeEndpoint<K,V> endpoint = new AmqpSinkBridgeEndpoint<K,V>(vertx, AmqpBridgeConfig.fromMap(envVars));
 		endpoint.open();
 		
 		// Create a mock for the sender
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 7d2360c..7771b03 100644
--- a/src/test/java/io/strimzi/kafka/bridge/example/AmqpBridgeServer.java
+++ b/src/test/java/io/strimzi/kafka/bridge/example/AmqpBridgeServer.java
@@ -17,7 +17,7 @@
 package io.strimzi.kafka.bridge.example;
 
 import io.strimzi.kafka.bridge.amqp.AmqpBridge;
-import io.strimzi.kafka.bridge.amqp.AmqpBridgeConfigProperties;
+import io.strimzi.kafka.bridge.amqp.AmqpBridgeConfig;
 import io.vertx.core.Vertx;
 
 import java.io.IOException;
@@ -31,10 +31,9 @@ public class AmqpBridgeServer {
 		
 		Vertx vertx = Vertx.vertx();
 
-		AmqpBridgeConfigProperties bridgeConfigProperties = new AmqpBridgeConfigProperties();
+		AmqpBridgeConfig bridgeConfigProperties = AmqpBridgeConfig.fromMap(System.getenv());
 		
-		AmqpBridge bridge = new AmqpBridge();
-		bridge.setBridgeConfigProperties(bridgeConfigProperties);
+		AmqpBridge bridge = new AmqpBridge(bridgeConfigProperties);
 
 		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 abfd69e..02b20bb 100644
--- a/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/http/HttpBridgeTest.java
@@ -39,6 +39,8 @@ import org.junit.runner.RunWith;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -47,6 +49,8 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
 
     private static final Logger log = LoggerFactory.getLogger(HttpBridgeTest.class);
 
+    private static Map<String, String> envVars = new HashMap<>();
+
     private static final String BRIDGE_HOST = "0.0.0.0";
     private static final int BRIDGE_PORT = 8080;
 
@@ -58,15 +62,15 @@ public class HttpBridgeTest extends KafkaClusterTestBase {
     private Vertx vertx;
     private HttpBridge httpBridge;
 
-    private HttpBridgeConfigProperties bridgeConfigProperties = new HttpBridgeConfigProperties();
+    private HttpBridgeConfig bridgeConfigProperties;
 
     @Before
     public void before(TestContext context) {
 
         vertx = Vertx.vertx();
 
-        this.httpBridge = new HttpBridge();
-        this.httpBridge.setBridgeConfigProperties(this.bridgeConfigProperties);
+        this.bridgeConfigProperties = HttpBridgeConfig.fromMap(envVars);
+        this.httpBridge = new HttpBridge(this.bridgeConfigProperties);
 
         this.vertx.deployVerticle(this.httpBridge, context.asyncAssertSuccess());
     }