# Java examples for Kafka consumer

**URL:** <https://discuss.akka.io/t/java-examples-for-kafka-consumer/3556>\
**Category:** Akka Streams & Alpakka\
**Tags:** java, kafka\
**Created:** [February 26, 2019, 8:32pm UTC](https://discuss.akka.io/t/java-examples-for-kafka-consumer/3556 "2019-02-26T20:32:01Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![shashankbrgowda](https://avatars.discourse-cdn.com/v4/letter/s/90db22/32.png) [@shashankbrgowda](https://discuss.akka.io/u/shashankbrgowda)\
**Post date:** [February 26, 2019, 8:32pm UTC](https://discuss.akka.io/t/java-examples-for-kafka-consumer/3556/1 "2019-02-26T20:32:02Z")

</div>

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.

---

<div class="post-metadata">

**Author:** ![TimMoore](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/timmoore/32/7_2.png) [@TimMoore](https://discuss.akka.io/u/TimMoore)\
**Post date:** [February 27, 2019, 1:43am UTC](https://discuss.akka.io/t/java-examples-for-kafka-consumer/3556/2 "2019-02-27T01:43:11Z")

</div>

What problems are you seeing?

---

<div class="post-metadata">

**Author:** ![shashankbrgowda](https://avatars.discourse-cdn.com/v4/letter/s/90db22/32.png) [@shashankbrgowda](https://discuss.akka.io/u/shashankbrgowda)\
**Post date:** [February 27, 2019, 4:20am UTC](https://discuss.akka.io/t/java-examples-for-kafka-consumer/3556/3 "2019-02-27T04:20:10Z")

</div>

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’.

---

<div class="post-metadata">

**Author:** ![shashankbrgowda](https://avatars.discourse-cdn.com/v4/letter/s/90db22/32.png) [@shashankbrgowda](https://discuss.akka.io/u/shashankbrgowda)\
**Post date:** [February 27, 2019, 4:43am UTC](https://discuss.akka.io/t/java-examples-for-kafka-consumer/3556/4 "2019-02-27T04:43:51Z")

</div>

Its working. There was mismatch with JSON attributes.