How can i reduce lag in kafka consumer/producer

Question

I am looking for improvement in scala kafka code. For reduce lag, what should i do in consumer & 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.

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()
  }
}

and this is consumer 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 => 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 <- 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 => c.toByte)
          handleMessage(outMessage)
        }
      }
    }
  }
}

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.

Answer 1

Increase the number of consumer(KafkaListener) instances with same group id. It will increase the consumption rate. Eventually your lag between producer write & consumer will get minimized.