Как прочитать значения из темы кафки, используя потребителя через akka-streams / alpakka-kafka? - PullRequest
Новая
       23

Как прочитать значения из темы кафки, используя потребителя через akka-streams / alpakka-kafka?

0 голосов
/ 28 октября 2019

При работе с Consumer.plainSource ничего не происходит. Разве это не единственный способ читать тему кафки через потоки. 

val consumerSettings2 = ConsumerSettings(system,new StringDeserializer,new StringDeserializer)
        .withBootstrapServers("localhost:3333")
        .withGroupId("ssss")
        .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest")

val source: Source[ConsumerRecord[String, String], Consumer.Control] =
Consumer.plainSource(consumerSettings2, Subscriptions.topics("candy"))

val sink = 
Sink.foreach[ConsumerRecord[String,String]](x=>println("consumed "+x))

source.runWith(sink)
...