Update KafkaClients.java

Author: MohamedEL59

core/src/main/java/com/softwaremill/kmq/KafkaClients.java

@@ -21,7 +21,7 @@ public <K, V> KafkaProducer<K, V> createProducer(Class<? extends Serializer<K>>
         return createProducer(keySerializer, valueSerializer, Collections.emptyMap());
     }
 
-    public <K, V> KafkaProducer<K, V> createProducer(Class<? extends Serializer<K>> keySerializer, Class<? extends Serializer<V>> valueSerializer,
+    public <K, V> KafkaProducer<K, V>  createProducer(Class<? extends Serializer<K>> keySerializer, Class<? extends Serializer<V>> valueSerializer,
                                                             Map<String, Object> extraConfig) {
         Properties props = new Properties();
         props.put("bootstrap.servers", bootstrapServers);
@@ -39,9 +39,13 @@ public <K, V> KafkaProducer<K, V> createProducer(Class<? extends Serializer<K>>
         return new KafkaProducer<>(props);
     }
 
+  public <K, V> KafkaConsumer<K, V> createConsumer(String groupId, Class<? extends Deserializer<K>> keyDeserializer, Class<? extends Deserializer<V>> valueDeserializer) {
+      return createConsumer(groupId, keyDeserializer, valueDeserializer, Collections.emptyMap());
+  }
+
     public <K, V> KafkaConsumer<K, V> createConsumer(String groupId,
                                                      Class<? extends Deserializer<K>> keyDeserializer,
-                                                     Class<? extends Deserializer<V>> valueDeserializer) {
+                                                     Class<? extends Deserializer<V>> valueDeserializer, Map<String, Object> extraConfig) {
         Properties props = new Properties();
         props.put("bootstrap.servers", bootstrapServers);
         props.put("enable.auto.commit", "false");
@@ -51,6 +55,10 @@ public <K, V> KafkaConsumer<K, V> createConsumer(String groupId,
         if (groupId != null) {
             props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
         }
+        // extraConfig : configure the kafka parameters (ex: ssl, ...)
+        for (Map.Entry<String, Object> extraCfgEntry : extraConfig.entrySet()) {
+          props.put(extraCfgEntry.getKey(), extraCfgEntry.getValue());
+        }
 
         return new KafkaConsumer<>(props);
     }