Awesome
Use case
Read the famous internal topic __consumer_offsets
with any projects that can only read JSON topics, such as timeseries databases, to enable some monitoring and alerting.
Details on the blog post: https://www.sderosiaux.com/articles/2017/08/07/looking-at-kafka-s-consumers-offsets
How to run it
Clone the project, and sbt run
, or via Intellij IDEA, run the Main
class.
You can also build it using sbt universal:packageBin
and unzip target\universal\kafka-consumer-offsets-to-json-topic-1.0.zip
.
Dependencies
It only depends on Kafka Streams 0.11 (it can be setup to 0.10 if needed).
It is using some classes from Kafka server to know how to parse the content of __consumer_offsets
which is binary-based.
Notes
Kafka is regularly commiting messages to __consumer_offsets
, even if no-one is consuming.
By running this Kafka Streams project, and listening to the output topic, you'll get something like :
$ kafka-console-consumer.bat --topic __consumer_offsets_json --new-consumer --bootstrap-server localhost:9092
{"topic":"WordsWithCountsTopic","partition":0,"group":"console-consumer-26549","version":1,"offset":95,"metadata":"","commitTimestamp":1501542796444,"expireTimestamp":1501629196444}
{"topic":"WordsWithCountsTopic","partition":0,"group":"console-consumer-26549","version":1,"offset":95,"metadata":"","commitTimestamp":1501542801444,"expireTimestamp":1501629201444}
{"topic":"WordsWithCountsTopic","partition":0,"group":"console-consumer-26549","version":1,"offset":95,"metadata":"","commitTimestamp":1501542806444,"expireTimestamp":1501629206444}
{"topic":"WordsWithCountsTopic","partition":0,"group":"console-consumer-26549","version":1,"offset":95,"metadata":"","commitTimestamp":1501542811445,"expireTimestamp":1501629211445}
{"topic":"WordsWithCountsTopic","partition":0,"group":"console-consumer-26549","version":1,"offset":95,"metadata":"","commitTimestamp":1501542816447,"expireTimestamp":1501629216447}
{"topic":"__consumer_offsets","partition":5,"group":"consumer-offsets-consumer-app-2","version":1,"offset":106,"metadata":"","commitTimestamp":1501542868586,"expireTimestamp":1501629268586}
{"topic":"__consumer_offsets","partition":17,"group":"consumer-offsets-consumer-app-2","version":1,"offset":1025,"metadata":"","commitTimestamp":1501542868736,"expireTimestamp":1501629268736}
The output topic offsets are not published to not cause an infinite loop (listening to it would cause messages published to this same topic, that would move on the offsets, that would cause new messages to pop etc.).
Warning
It's probably not a good idea to rely on this in production, because it's an implementation detail that can changed across versions.
Options
The broker address and the output topic can be customized.
Defaults are:
localhost:9092
__consumer_offsets_json
-b, --broker <arg>
-o, --output-topic <arg>
--help Show help message