cs-kafka-reader

Kafka CloudStack Events Reader and Evaluator Framework

The aim of the library is the convenient handling of Kafka messages. It provides the mechanisms to:

Buffer the messages to vary a count of processing messages without changing a consumer properties
Implement the logic of storing consumer offsets to any place

Install with SBT

Add the following to your build.sbt

libraryDependencies += "com.bwsw" %% "kafka-reader" % "0.10.1"

Getting Started

The diagram below is a simple illustration of how the library's classes should be used.
Implement your own EventHandler and EventManager in the way as it is displayed on the diagram.
where:
* K - type of ConsumerRecord key
* V - type of ConsumerRecord value
* T - type of data after handle a ConsumerRecord by the instance of EventHandler implementation

Example Usage

The example below shows how to print messages from Kafka to the console. The call to the method is performed in Future:

class SimpleEventHandler(messageQueue: MessageQueue[String,String], messageCount: Int)
  extends EventHandler[String,String,Future[Unit]](messageQueue, messageCount) {

  override def handle(flag: AtomicBoolean): List[OutputEnvelope[Future[Unit]]] = {
    val inputEnvelopes = messageQueue.take(messageCount)
    inputEnvelopes.map { x =>
      OutputEnvelope[Future[Unit]](x.topic, x.partition, x.offset, Future(println(x.data)))
    }
  }

}

According to the diagram above the main class looks like this:

object EventManager {
    
    def main(args: Array[String]): Unit = {
          val dummyFlag = new AtomicBoolean(true)
          val consumer = new Consumer[String,String](Consumer.Settings("localhost:9092", "group01", 3000))
      
          val checkpointInfoProcessor = new CheckpointInfoProcessor[String,String,Future[Unit]](
            TopicInfoList(List(TopicInfo(topic = "topic1"))),
            consumer
          )
      
          val messageQueue = new MessageQueue[String,String](consumer)
      
          val eventHandler = new SimpleEventHandler(messageQueue, countOfMessages = 1)
      
          checkpointInfoProcessor.load()
      
          val outputEnvelopes = eventHandler.handle(dummyFlag)
          
          outputEnvelopes.data.foreach {
            case Success(x) => 
            case Failure(e) =>
              prinln(s"something went wrong, exception was thrown: $e")
              throw e
          }
          
          checkpointInfoProcessor.save(outputEnvelopes)
          consumer.close()
      }
}

Testing

Unit tests

Run tests: sbt test

Integration tests

Run tests: sbt it:test

Versioning

Library has the same version as Apache Kafka library

License

This project is licensed under the Apache License - see the LICENSE file for details

bwsw / kafka-reader 0.10.2