Message timestamp and commit_timestamp in Kafka's __consumer_offsets

user_11597774 0
Viewed 1946

I'm currently working on an application that partially relies on Apache Kafka (version 2.2.0). One thing I have to do is to keep track of what (and more importantly when) other consumers commit their current offsets. As far as I can tell, just using the Java client, there is no way to get the associated timestamps for the committed offsets because AdminClient's listConsumerGroupOffsets method eventually leads to an OffsetAndMetadata object, which does not include a timestamp. So instead, I simply started reading the messages from the __consumer_offsets topic. If there is a better way to do this, please let me know.

Now, if one reads the messages in __consumer_offsets directly, then one all of a sudden has two timestamps. One is the timestamp attached to the actual commit message and the other one is the commit_timestamp, which is part of the message's content. My first thought was that one of these is probably set by the broker and the other one might be set by the client that committed it (also, if you look at /config/topics/__consumer_offsets in ZooKeeper, it doesn't specify a LogAppendTime message timestamp, so one could assume that it just uses the default). Alas, a quick experiment with manually shifted system times shows that both are actually set by the broker. What's more, they don't always agree (the message's timestamp is sometimes slightly ahead of the commit_timestamp). I've tried to dive into the Kafka code to understand exactly what's going on but it is rather convoluted and I'm not familiar enough with it to quickly get a good grasp. So here are my questions:

  1. Why is the message timestamp in __consumer_offsets automatically LogAppendTime even though that's not explicitly specified? Is it just that the producer that is used to send the commit message leaves the timestamp empty?
  2. Why do the message timestamp and the commit_timestamp included in the message not agree? I seem to recall having read somewhere that it used to be possible to explicitly set the commit_timestamp and thereby manually control the retention of the committed offset.
  3. More importantly: Is there any reason to use one over the other? For example, if it's still possible to set the commit_timestamp manually, it would make much more sense to use the timestamp attached to the message.

I understand that this is a highly specific question and probably of little importance to most. But until now, I was always able to understand what's going on in the background by using Google and looking at Kafka's source code; yet, this one has me a bit stumped. So any insights are greatly appreciated.

kafka-consumer-api apache-kafka
1 Answers

I think latter time stamp is expiration time. Can you please try following to make sure?

Set internal topics accessible by setting parameter "exclude.internal.topics=false" in consumer.config.

bin/kafka-console-consumer.sh  --consumer.config /tmp/consumer.config \
     --bootstrap-server localhost:9092 \
     --topic __consumer_offsets

I could see result as follows:

[mygroup1,mytopic1,11]::[OffsetMetadata[55166421,NO_METADATA],CommitTime 1502060076305,ExpirationTime 1502146476305]
[mygroup1,mytopic1,13]::[OffsetMetadata[55037927,NO_METADATA],CommitTime 1502060076305,ExpirationTime 1502146476305]
[mygroup2,mytopic2,0]::[OffsetMetadata[126,NO_METADATA],CommitTime 1502060076343,ExpirationTime 1502146476343]

I don't have specific version as you mentioned in your question installed in my machine, so please check it in your end.

user_115297010
Related