Skip to content

Instantly share code, notes, and snippets.

@mrwillis mrwillis/main.scala
Created Jun 2, 2018

Embed
What would you like to do?
Kafka BroadcastHub (reactive-kafka)
import akka.NotUsed
import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, KafkaConsumerActor, Subscriptions}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{BroadcastHub, Keep, RunnableGraph, Sink, Source}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
object Main {
implicit val system = ActorSystem("System")
implicit val mat = ActorMaterializer()
def main(args: Array[String]): Unit = {
val consumerSettings =
ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
.withGroupId(scala.util.Random.alphanumeric take 10 mkString)
val kafkaConsumerActor =
system.actorOf(KafkaConsumerActor.props(consumerSettings))
val producer
: Source[ConsumerRecord[String, String], Consumer.Control] =
Consumer.plainExternalSource[String, String](
kafkaConsumerActor,
Subscriptions.assignment(
new TopicPartition("topic", 0))
)
val hub
: RunnableGraph[Source[ConsumerRecord[String, String], NotUsed]] =
producer.toMat(BroadcastHub.sink)(Keep.right)
val source: Source[ConsumerRecord[String, String], NotUsed] =
hub.run()
source
.runWith(Sink.foreach(data => println("Consumer 1" + data.value)))
source
.runWith(Sink.foreach(data => println("Consumer 2" + data.value)))
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.