title: How can i reduce lag in kafka consumer/producer
body: <p>I am looking for improvement in scala kafka code. For reduce lag, what should i do in consumer &amp; producer. 
  This is the code I got from someone.
  I know this code is not a difficult code. But I have never seen scala code before, and I am just beginning to learn about kafka. So I have a hard time finding the problem.</p>
  
  <pre><code>import java.util.Properties
  import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
  
  import scala.util.Try
  
  class KafkaMessenger(val servers: String, val sender: String) {
    val props = new Properties()
    props.put("bootstrap.servers", servers)
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("producer.type", "async")
  
    val producer = new KafkaProducer[String, String](props)
  
    def send(topic: String, message: Any): Try[Unit] = Try {
      producer.send(new ProducerRecord(topic, message.toString))
    }
  
    def close(): Unit = producer.close()
  }
  
  object KafkaMessenger {
    def apply(host: String, topic: String, sender: String, message: String): Unit = {
      val messenger = new KafkaMessenger(host, sender)
      messenger.send(topic, message)
      messenger.close()
    }
  }
  </code></pre>
  
  <p>and this is consumer code.</p>
  
  <pre><code>import java.util.Properties
  import java.util.concurrent.Executors
  
  import com.satreci.g2gs.common.impl.utils.KafkaMessageTypes._
  import kafka.admin.AdminUtils
  import kafka.consumer._
  import kafka.utils.ZkUtils
  import org.I0Itec.zkclient.{ZkClient, ZkConnection}
  import org.slf4j.LoggerFactory
  
  import scala.language.postfixOps
  
  class KafkaListener(val zookeeper: String,
                      val groupId: String,
                      val topic: String,
                      val handleMessage: ByteArrayMessage =&gt; Unit,
                      val workJson: String = ""
                     ) extends AutoCloseable {
    private lazy val logger = LoggerFactory.getLogger(this.getClass)
    val config: ConsumerConfig = createConsumerConfig(zookeeper, groupId)
    val consumer: ConsumerConnector = Consumer.create(config)
    val sessionTimeoutMs: Int = 10 * 1000
    val connectionTimeoutMs: Int = 8 * 1000
    val zkClient: ZkClient = ZkUtils.createZkClient(zookeeper, sessionTimeoutMs, connectionTimeoutMs)
    val zkUtils = new ZkUtils(zkClient, new ZkConnection(zookeeper), false)
  
    def createConsumerConfig(zookeeper: String, groupId: String): ConsumerConfig = {
      val props = new Properties()
      props.put("zookeeper.connect", zookeeper)
      props.put("group.id", groupId)
      props.put("auto.offset.reset", "smallest")
      props.put("zookeeper.session.timeout.ms", "5000") 
      props.put("zookeeper.sync.time.ms", "200")
      props.put("auto.commit.interval.ms", "1000")
      props.put("partition.assignment.strategy", "roundrobin")
      new ConsumerConfig(props)
    }
  
    def run(threadCount: Int = 1): Unit = {
      val streams = consumer.createMessageStreamsByFilter(Whitelist(topic), threadCount)
  
      if (!AdminUtils.topicExists(zkUtils, topic)) {
        AdminUtils.createTopic(zkUtils, topic, 1, 1)
      }
  
      val executor = Executors.newFixedThreadPool(threadCount)
      for (stream &lt;- streams) {
        executor.submit(new MessageConsumer(stream))
      }
      logger.debug(s"KafkaListener start with ${threadCount}thread (topic=$topic)")
    }
  
    override def close(): Unit = {
      consumer.shutdown()
      logger.debug(s"$topic Listener close")
    }
  
    class MessageConsumer(val stream: MessageStream) extends Runnable {
      override def run(): Unit = {
        val it = stream.iterator()
        while (it.hasNext()) {
          val message = it.next().message()
          if (workJson == "") {
            handleMessage(message)
          }
          else {
            val strMessage = new String(message)
            val newMessage = s"$strMessage/#/$workJson"
            val outMessage = newMessage.toCharArray.map(c =&gt; c.toByte)
            handleMessage(outMessage)
          }
        }
      }
    }
  }
  </code></pre>
  
  <p>Specifically, I want to modify the structure that creates KafkaProduce objects whenever I send a message. There seems to be many other improvements to reduce lag.</p>