If you have questions or are working on a pull request or just curious, please feel welcome to join the chat room:
Reactive Streams wrapper for Apache Kafka.
Initiated by SoftwareMill
Supports Kafka 0.8.2.1
Available at Maven Central for Scala 2.10 and 2.11:
libraryDependencies += "com.softwaremill" %% "reactive-kafka" % "0.6.0"
Tests require Apache Kafka and Zookeeper to be available on localhost:9092 and localhost:2181
import akka.actor.ActorSystem
import akka.stream.ActorFlowMaterializer
import akka.stream.scaladsl.{Sink, Source}
import com.softwaremill.react.kafka.ReactiveKafka
import kafka.serializer.{StringDecoder, StringEncoder}
implicit val actorSystem = ActorSystem("ReactiveKafka")
implicit val materializer = ActorFlowMaterializer()
val kafka = new ReactiveKafka(host = "localhost:9092", zooKeeperHost = "localhost:2181")
val publisher = kafka.consume("lowercaseStrings", "groupName", new StringDecoder())
val subscriber = kafka.publish("uppercaseStrings", "groupName", new StringEncoder())
Source(publisher).map(_.toUpperCase).to(Sink(subscriber)).run()
By default a new consumer will start reading from the beginning of a topic. If you want to start reading from the end, you can use alternative way to create a consumer:
val publisher = kafka.consumeFromEnd(topic, groupId, new StringDecoder())
KafkaActorSubscriber and KafkaActorPublisher have their own thread pools, configured in reference.conf
.
You can tune them by overriding kafka-publisher-dispatcher.thread-pool-executor
and
kafka-subscriber-dispatcher.thread-pool-executor
in your application.conf
file.