Ytukyg

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.

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.