Update Redeliverer.scala

Author: MohamedEL59

core/src/main/scala/com.softwaremill.kmq/redelivery/Redeliverer.scala

@@ -21,15 +21,20 @@ trait Redeliverer {
 
 class DefaultRedeliverer(
   partition: Partition, producer: KafkaProducer[Array[Byte], Array[Byte]],
-  config: KmqConfig, clients: KafkaClients)
+  config: KmqConfig, clients: KafkaClients, extraConfig: Option[java.util.Map[String, Object]] = None)
   extends Redeliverer with StrictLogging {
 
   private val SendTimeoutSeconds = 60L
 
   private val tp = new TopicPartition(config.getMsgTopic, partition)
 
   private val reader = {
-    val c = clients.createConsumer(null, classOf[ByteArrayDeserializer], classOf[ByteArrayDeserializer])
+    val c = extraConfig match {
+      // extraConfig is not empty
+      case Some(cfg) => clients.createConsumer(null, classOf[ByteArrayDeserializer], classOf[ByteArrayDeserializer], cfg)
+      // extraConfig is empty
+      case None => clients.createConsumer(null, classOf[ByteArrayDeserializer], classOf[ByteArrayDeserializer])
+    }
     c.assign(Collections.singleton(tp))
     new SingleOffsetReader(tp, c)
   }
@@ -133,4 +138,4 @@ private class SingleOffsetReader(tp: TopicPartition, consumer: KafkaConsumer[Arr
   def close(): Unit = {
     consumer.close()
   }
-}
+}