ardlema ardlema - 2 years ago 94
Scala Question

Kafka streams word count application

I'm playing around with the kafka streaming API (Kakfa version: 0.10.2.0) trying to make a simple wordcount example work: Wordcount App gist. I'm running both producer and console consumer:

./kafka-console-producer.sh -topic input-topic --broker-list localhost:9092


./kafka-console-consumer.sh --topic output-topic --bootstrap-server localhost:9092 --from-beginning


start the application and everything seems to be working fine but when I type in some strings within the console producer, the consumer receives nothing at all. If I change the app to do a simple toUppercase on the input the consumer receives the stream (modified to upper case) fine:

//The following code works fine:
val uppercasedWithMapValues: KStream[String, String] = textLines.mapValues(_.toUpperCase())
uppercasedWithMapValues.to("output-topic")


Does anyone know why I'm receiving nothing on the word-count example? Should I specify any serializer on the consumer? In my last test the console consumer processed the messages that I sent through the console but didn't show them, see below the output:

➜ bin ./kafka-console-consumer.sh \
--topic output-topic \
--bootstrap-server localhost:9092 \
--from-beginning
[2017-08-02 07:48:20,187]WARN Error while fetching metadata with correlation id 2 :
{output-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-08-02 07:48:20,197] WARN The following subscribed topics are not assigned
to any members in the group console-consumer-91651 : [output-topic]
(org.apache.kafka.clients.consumer.internals.ConsumerCoordinator)


^CProcessed a total of 7 messages

Answer Source

KStream works because it doesn't use caching. For KTable you have to wait a bit, or set cache.max.bytes.buffering to 0 (but not in a production code!)

Recommended from our users: Dynamic Network Monitoring from WhatsUp Gold from IPSwitch. Free Download