Is it possible to access message headers with Kafka Streams?

by

Records headers are accessible since versions 2.0.0 (cf. KIP-244 for details).

You can access record metadata via the Processor API (ie, via transform(), transformValues(), or process()), by the given “context” object (cf. https://docs.confluent.io/current/streams/developer-guide/processor-api.html#accessing-processor-context).

Update

As of 2.7.0 release, the Processor API was improved (cf. KIP-478), adding a new type-safe api.Processor class with process(Record) instead of process(K, V) method. For this case, headers (and record metadata) are accessible via the Record class).

This new feature is not yet available in “PAPI method of the DSL though (eg. KStream#process(), KStream#transform() and siblings).

+++++

Prior to 2.0, the context only exposes topic, partition, offset, and timestamp—but not headers that are in fact dropped by Streams on read in those older versions.

Metadata is not available at DSL level though. However, there is also work in progress to extend the DSL via KIP-159.

More Related Contents:

  1. Handling exceptions in Kafka streams
  2. Dynamically connecting a Kafka input stream to multiple output streams
  3. Connect to Kafka running in Docker
  4. How can I send large messages with Kafka (over 15MB)?
  5. How to create a Topic in Kafka through Java
  6. How Can we create a topic in Kafka from the IDE using API
  7. What determines Kafka consumer offset?
  8. Writing Custom Kafka Serializer
  9. Java, How to get number of messages in a topic in apache kafka
  10. How to Delete a topic in apache kafka [duplicate]
  11. When does the Apache Kafka client throw a “Batch Expired” exception?
  12. What is a simple, effective way to debug custom Kafka connectors?
  13. Is it possible to add partitions to an existing topic in Kafka 0.8.2
  14. How to get kafka consume lag in java program
  15. How to create Custom serializer in kafka?
  16. Kafka Consumer CommitFailedException
  17. Can a Kafka consumer(0.8.2.2) read messages in batch
  18. How can I delete a topic in Apache Kafka? [duplicate]
  19. Iterate through a possibly infinite amount of nested ArrayLists
  20. Normalization in DOM parsing with java – how does it work?
  21. How can I get a list of all the implementations of an interface programmatically in Java?
  22. Convert String to Code
  23. Why does autoReconnect=true not seem to work?
  24. Java user.home is being set to %userprofile% and not being resolved
  25. Try-with-resources equivalent in Java 1.6
  26. How perform task on JavaFX TextField at onfocus and outfocus?
  27. Set field value with reflection
  28. What’s the difference between a System property and environment variable
  29. How to sort an attribute of an object using Collections
  30. Add ArrayList to another ArrayList in java

Leave a Comment