Added some finals

src/main/java/io/strimzi/kafka/bridge/config/AbstractConfig.java

@@ -12,7 +12,7 @@
  */
 public abstract class AbstractConfig {
 
-    protected Map<String, Object> config;
+    protected final Map<String, Object> config;
 
     /**
      * Constructor

src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java

@@ -24,8 +24,8 @@ public class BridgeConfig extends AbstractConfig {
     /** Tracing system to be used in the bridge */
     public static final String TRACING_TYPE = BRIDGE_CONFIG_PREFIX + "tracing";
 
-    private KafkaConfig kafkaConfig;
-    private HttpConfig httpConfig;
+    private final KafkaConfig kafkaConfig;
+    private final HttpConfig httpConfig;
 
     /**
      * Constructor

src/main/java/io/strimzi/kafka/bridge/config/KafkaConfig.java

@@ -17,9 +17,9 @@ public class KafkaConfig extends AbstractConfig {
     /** Prefix for Kafka related configuration parameters */
     public static final String KAFKA_CONFIG_PREFIX = "kafka.";
 
-    private KafkaProducerConfig producerConfig;
-    private KafkaConsumerConfig consumerConfig;
-    private KafkaAdminConfig adminConfig;
+    private final KafkaProducerConfig producerConfig;
+    private final KafkaConsumerConfig consumerConfig;
+    private final KafkaAdminConfig adminConfig;
 
     /**
      * Constructor

src/main/java/io/strimzi/kafka/bridge/http/HttpBridge.java

@@ -76,9 +76,9 @@ public class HttpBridge extends AbstractVerticle {
 
     private Router router;
 
-    private Map<ConsumerInstanceId, Long> timestampMap = new HashMap<>();
+    private final Map<ConsumerInstanceId, Long> timestampMap = new HashMap<>();
 
-    private MetricsReporter metricsReporter;
+    private final MetricsReporter metricsReporter;
 
     /**
      * Constructor
@@ -653,183 +653,183 @@ private boolean isReady() {
         return this.isReady;
     }
 
-    HttpOpenApiOperation SEND = new HttpOpenApiOperation(HttpOpenApiOperations.SEND) {
+    final HttpOpenApiOperation SEND = new HttpOpenApiOperation(HttpOpenApiOperations.SEND) {
 
         @Override
         public void process(RoutingContext routingContext) {
             send(routingContext);
         }
     };
 
-    HttpOpenApiOperation SEND_TO_PARTITION = new HttpOpenApiOperation(HttpOpenApiOperations.SEND_TO_PARTITION) {
+    final HttpOpenApiOperation SEND_TO_PARTITION = new HttpOpenApiOperation(HttpOpenApiOperations.SEND_TO_PARTITION) {
 
         @Override
         public void process(RoutingContext routingContext) {
             sendToPartition(routingContext);
         }
     };
 
-    HttpOpenApiOperation CREATE_CONSUMER = new HttpOpenApiOperation(HttpOpenApiOperations.CREATE_CONSUMER) {
+    final HttpOpenApiOperation CREATE_CONSUMER = new HttpOpenApiOperation(HttpOpenApiOperations.CREATE_CONSUMER) {
 
         @Override
         public void process(RoutingContext routingContext) {
             createConsumer(routingContext);
         }
     };
 
-    HttpOpenApiOperation DELETE_CONSUMER = new HttpOpenApiOperation(HttpOpenApiOperations.DELETE_CONSUMER) {
+    final HttpOpenApiOperation DELETE_CONSUMER = new HttpOpenApiOperation(HttpOpenApiOperations.DELETE_CONSUMER) {
 
         @Override
         public void process(RoutingContext routingContext) {
             deleteConsumer(routingContext);
         }
     };
 
-    HttpOpenApiOperation SUBSCRIBE = new HttpOpenApiOperation(HttpOpenApiOperations.SUBSCRIBE) {
+    final HttpOpenApiOperation SUBSCRIBE = new HttpOpenApiOperation(HttpOpenApiOperations.SUBSCRIBE) {
 
         @Override
         public void process(RoutingContext routingContext) {
             subscribe(routingContext);
         }
     };
 
-    HttpOpenApiOperation UNSUBSCRIBE = new HttpOpenApiOperation(HttpOpenApiOperations.UNSUBSCRIBE) {
+    final HttpOpenApiOperation UNSUBSCRIBE = new HttpOpenApiOperation(HttpOpenApiOperations.UNSUBSCRIBE) {
 
         @Override
         public void process(RoutingContext routingContext) {
             unsubscribe(routingContext);
         }
     };
 
-    HttpOpenApiOperation LIST_SUBSCRIPTIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_SUBSCRIPTIONS) {
+    final HttpOpenApiOperation LIST_SUBSCRIPTIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_SUBSCRIPTIONS) {
 
         @Override
         public void process(RoutingContext routingContext) {
             listSubscriptions(routingContext);
         }
     };
 
-    HttpOpenApiOperation LIST_TOPICS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_TOPICS) {
+    final HttpOpenApiOperation LIST_TOPICS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_TOPICS) {
 
         @Override
         public void process(RoutingContext routingContext) {
             listTopics(routingContext);
         }
     };
 
-    HttpOpenApiOperation GET_TOPIC = new HttpOpenApiOperation(HttpOpenApiOperations.GET_TOPIC) {
+    final HttpOpenApiOperation GET_TOPIC = new HttpOpenApiOperation(HttpOpenApiOperations.GET_TOPIC) {
 
         @Override
         public void process(RoutingContext routingContext) {
             getTopic(routingContext);
         }
     };
 
-    HttpOpenApiOperation LIST_PARTITIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_PARTITIONS) {
+    final HttpOpenApiOperation LIST_PARTITIONS = new HttpOpenApiOperation(HttpOpenApiOperations.LIST_PARTITIONS) {
 
         @Override
         public void process(RoutingContext routingContext) {
             listPartitions(routingContext);
         }
     };
 
-    HttpOpenApiOperation GET_PARTITION = new HttpOpenApiOperation(HttpOpenApiOperations.GET_PARTITION) {
+    final HttpOpenApiOperation GET_PARTITION = new HttpOpenApiOperation(HttpOpenApiOperations.GET_PARTITION) {
 
         @Override
         public void process(RoutingContext routingContext) {
             getPartition(routingContext);
         }
     };
 
-    HttpOpenApiOperation GET_OFFSETS = new HttpOpenApiOperation(HttpOpenApiOperations.GET_OFFSETS) {
+    final HttpOpenApiOperation GET_OFFSETS = new HttpOpenApiOperation(HttpOpenApiOperations.GET_OFFSETS) {
 
         @Override
         public void process(RoutingContext routingContext) {
             getOffsets(routingContext);
         }
     };
 
-    HttpOpenApiOperation ASSIGN = new HttpOpenApiOperation(HttpOpenApiOperations.ASSIGN) {
+    final HttpOpenApiOperation ASSIGN = new HttpOpenApiOperation(HttpOpenApiOperations.ASSIGN) {
 
         @Override
         public void process(RoutingContext routingContext) {
             assign(routingContext);
         }
     };
 
-    HttpOpenApiOperation POLL = new HttpOpenApiOperation(HttpOpenApiOperations.POLL) {
+    final HttpOpenApiOperation POLL = new HttpOpenApiOperation(HttpOpenApiOperations.POLL) {
 
         @Override
         public void process(RoutingContext routingContext) {
             poll(routingContext);
         }
     };
 
-    HttpOpenApiOperation COMMIT = new HttpOpenApiOperation(HttpOpenApiOperations.COMMIT) {
+    final HttpOpenApiOperation COMMIT = new HttpOpenApiOperation(HttpOpenApiOperations.COMMIT) {
 
         @Override
         public void process(RoutingContext routingContext) {
             commit(routingContext);
         }
     };
 
-    HttpOpenApiOperation SEEK = new HttpOpenApiOperation(HttpOpenApiOperations.SEEK) {
+    final HttpOpenApiOperation SEEK = new HttpOpenApiOperation(HttpOpenApiOperations.SEEK) {
 
         @Override
         public void process(RoutingContext routingContext) {
             seek(routingContext);
         }
     };
 
-    HttpOpenApiOperation SEEK_TO_BEGINNING = new HttpOpenApiOperation(HttpOpenApiOperations.SEEK_TO_BEGINNING) {
+    final HttpOpenApiOperation SEEK_TO_BEGINNING = new HttpOpenApiOperation(HttpOpenApiOperations.SEEK_TO_BEGINNING) {
 
         @Override
         public void process(RoutingContext routingContext) {
             seekToBeginning(routingContext);
         }
     };
 
-    HttpOpenApiOperation SEEK_TO_END = new HttpOpenApiOperation(HttpOpenApiOperations.SEEK_TO_END) {
+    final HttpOpenApiOperation SEEK_TO_END = new HttpOpenApiOperation(HttpOpenApiOperations.SEEK_TO_END) {
 
         @Override
         public void process(RoutingContext routingContext) {
             seekToEnd(routingContext);
         }
     };
 
-    HttpOpenApiOperation HEALTHY = new HttpOpenApiOperation(HttpOpenApiOperations.HEALTHY) {
+    final HttpOpenApiOperation HEALTHY = new HttpOpenApiOperation(HttpOpenApiOperations.HEALTHY) {
 
         @Override
         public void process(RoutingContext routingContext) {
             healthy(routingContext);
         }
     };
 
-    HttpOpenApiOperation READY = new HttpOpenApiOperation(HttpOpenApiOperations.READY) {
+    final HttpOpenApiOperation READY = new HttpOpenApiOperation(HttpOpenApiOperations.READY) {
 
         @Override
         public void process(RoutingContext routingContext) {
             ready(routingContext);
         }
     };
 
-    HttpOpenApiOperation OPENAPI = new HttpOpenApiOperation(HttpOpenApiOperations.OPENAPI) {
+    final HttpOpenApiOperation OPENAPI = new HttpOpenApiOperation(HttpOpenApiOperations.OPENAPI) {
 
         @Override
         public void process(RoutingContext routingContext) {
             openapi(routingContext);
         }
     };
 
-    HttpOpenApiOperation METRICS = new HttpOpenApiOperation(HttpOpenApiOperations.METRICS) {
+    final HttpOpenApiOperation METRICS = new HttpOpenApiOperation(HttpOpenApiOperations.METRICS) {
 
         @Override
         public void process(RoutingContext routingContext) {
             metrics(routingContext);
         }
     };
 
-    HttpOpenApiOperation INFO = new HttpOpenApiOperation(HttpOpenApiOperations.INFO) {
+    final HttpOpenApiOperation INFO = new HttpOpenApiOperation(HttpOpenApiOperations.INFO) {
 
         @Override
         public void process(RoutingContext routingContext) {

src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeContext.java

@@ -20,8 +20,8 @@
  */
 public class HttpBridgeContext<K, V> {
 
-    private Map<ConsumerInstanceId, HttpSinkBridgeEndpoint<K, V>> httpSinkEndpoints = new HashMap<>();
-    private Map<HttpConnection, HttpSourceBridgeEndpoint<K, V>> httpSourceEndpoints = new HashMap<>();
+    private final Map<ConsumerInstanceId, HttpSinkBridgeEndpoint<K, V>> httpSinkEndpoints = new HashMap<>();
+    private final Map<HttpConnection, HttpSourceBridgeEndpoint<K, V>> httpSourceEndpoints = new HashMap<>();
     private HttpAdminBridgeEndpoint httpAdminBridgeEndpoint;
     private HttpOpenApiOperations openApiOperation;
 

src/main/java/io/strimzi/kafka/bridge/http/HttpBridgeEndpoint.java

@@ -20,8 +20,8 @@ public abstract class HttpBridgeEndpoint {
     protected final Logger log = LoggerFactory.getLogger(getClass());
 
     protected String name;
-    protected EmbeddedFormat format;
-    protected BridgeConfig bridgeConfig;
+    protected final EmbeddedFormat format;
+    protected final BridgeConfig bridgeConfig;
     private Handler<HttpBridgeEndpoint> closeHandler;
 
     /**

src/main/java/io/strimzi/kafka/bridge/http/HttpOpenApiOperation.java

@@ -18,7 +18,7 @@ public abstract class HttpOpenApiOperation implements Handler<RoutingContext> {
 
     protected final static String LOGGER_NAME_PREFIX = "http.openapi.operation.";
 
-    protected Logger log;
+    protected final Logger log;
     protected final HttpOpenApiOperations operationId;
 
     /**

src/main/java/io/strimzi/kafka/bridge/http/HttpSinkBridgeEndpoint.java

@@ -60,9 +60,9 @@ public class HttpSinkBridgeEndpoint<K, V> extends HttpBridgeEndpoint {
     private long pollTimeOut = 100;
     private long maxBytes = Long.MAX_VALUE;
 
-    Pattern forwardedHostPattern = Pattern.compile("host=([^;]+)", Pattern.CASE_INSENSITIVE);
-    Pattern forwardedProtoPattern = Pattern.compile("proto=([^;]+)", Pattern.CASE_INSENSITIVE);
-    Pattern hostPortPattern = Pattern.compile("^.*:[0-9]+$");
+    final Pattern forwardedHostPattern = Pattern.compile("host=([^;]+)", Pattern.CASE_INSENSITIVE);
+    final Pattern forwardedProtoPattern = Pattern.compile("proto=([^;]+)", Pattern.CASE_INSENSITIVE);
+    final Pattern hostPortPattern = Pattern.compile("^.*:[0-9]+$");
 
     private MessageConverter<K, V, byte[], byte[]> messageConverter;
     private final HttpBridgeContext<K, V> httpBridgeContext;

src/main/java/io/strimzi/kafka/bridge/tracing/BridgeContextStorageProvider.java

@@ -45,7 +45,7 @@ public ContextStorage get() {
     enum BridgeContextStorage implements ContextStorage {
         INSTANCE;
 
-        private ConcurrentMap<Object, Object> data = new ConcurrentHashMap();
+        private final ConcurrentMap<Object, Object> data = new ConcurrentHashMap();
 
         @Override
         public Scope attach(Context toAttach) {