Update CommitMarkerOffsetsActor.scala

Author: MohamedEL59

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

@@ -10,9 +10,14 @@ import org.apache.kafka.common.serialization.ByteArrayDeserializer
 import scala.collection.JavaConverters._
 import scala.concurrent.duration._
 
-class CommitMarkerOffsetsActor(markerTopic: String, clients: KafkaClients) extends Actor with StrictLogging {
+class CommitMarkerOffsetsActor(markerTopic: String, clients: KafkaClients, extraConfig: Option[java.util.Map[String, Object]] = None) extends Actor with StrictLogging {
 
-  private val consumer = clients.createConsumer(null, classOf[ByteArrayDeserializer], classOf[ByteArrayDeserializer])
+  private val consumer = 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])
+  }
 
   private var toCommit: Map[Partition, Offset] = Map()