Java examples for Kafka consumer

Akka Akka Streams & Alpakka
,
1

I’m new to Akka streams, Are there any Java examples for streaming data from Kafka using Akka Stream and persisting Into Cassandra?
I’m having trouble deserializing JSON object. Anything wrong with below code?

final ObjectMapper mapper = new ObjectMapper();
final ObjectReader reader = mapper.readerFor(InventoryEvent.class);

Consumer.plainSource(consumerSettings, Subscriptions.topics(“inventory”))
.map(ConsumerRecord::value)
.map(reader::readValue)
.runForeach(event -> System.out.println(event.toString()), materializer);

As next step, I need to aggregate complex events and persist them into Cassandra.

2

What problems are you seeing?

3

I’m able to get the ConsumerRecord
ConsumerRecord(topic = inventory, partition = 0, leaderEpoch = 0, offset = 51, CreateTime = 1551194094648, serialized key size = -1, serialized value size = 131, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = [B@b7526f9) ConsumerRecord(topic = inventory, partition = 0, leaderEpoch = 0, offset = 52, CreateTime = 1551197760500, serialized key size = -1, serialized value size = 2, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = [B@354acdfd)

But, When I try to deserialize the value, Its not displaying anything. It logs
[INFO] [02/27/2019 09:48:29.026] [inventory-akka-consumer-akka.actor.default-dispatcher-2] [akka://inventory-akka-consumer/system/kafka-consumer-1] Message [akka.kafka.KafkaConsumerActor$Stop$] from Actor[akka://inventory-akka-consumer/system/StreamSupervisor-0/$$a#1716334169] to Actor[akka://inventory-akka-consumer/system/kafka-consumer-1#345710968] was not delivered. [1] dead letters encountered. If this is not an expected behavior, then [Actor[akka://inventory-akka-consumer/system/kafka-consumer-1#345710968]] may have terminated unexpectedly, This logging can be turned off or adjusted with configuration settings ‘akka.log-dead-letters’ and ‘akka.log-dead-letters-during-shutdown’.

4

Its working. There was mismatch with JSON attributes.