Feature Suggestion: Kafka Consumer consume from timestamp

[a1shadows]

Hi @authorjapps
I think it would be a great feature if we could configure the kafka consumer to consume from a specific timestamp. If we had this, we would not have to cover backlog on the kafka topic if there is significant gap between consecutive test runs.

If you think this is a feature worth pursuing, I would love to contribute to the feature. We can have something like:

      "consumerLocalConfigs": {
          "seekTImestamp": "1706244382000"
      }

}

This would make the consumer seek to this timestamp and start consumption from there.

[authorjapps]

Hi @authorjapps I think it would be a great feature if we could configure the kafka consumer to consume from a specific timestamp. If we had this, we would not have to cover backlog on the kafka topic if there is significant gap between consecutive test runs.

If you think this is a feature worth pursuing, I would love to contribute to the feature. We can have something like:

      "consumerLocalConfigs": {
          "seekTImestamp": "1706244382000"
      }

}

This would make the consumer seek to this timestamp and start consumption from there.

This is a great suggestion.
Also, we need to make sure messages are consumed from all partitions of the topic(s).

Happy for you to pick this issue and suggest a solution if you're fine with this.

[a1shadows]

Happy for you to pick this issue and suggest a solution if you're fine with

I think we can add Long seekToTimestamp to ConsumerLocalConfigs class and this can be used in handleSeekOffset method in KafkaReceiver to fetch the offset-timestamp map by

1. unsubscribe from current subscription
2. first fetching partitions on topic using KafkaConsumer#partitionsFor(String) and then
3. calling KafkaConsumer#offsetsForTimes(Map) on that.
4. calling KafkaConsumer#assign(Collection) on fetched topicPartitions
5. calling KafkaConsumer#seek(TopicPartition, long) for each topicPartition

We need not add the parameter to ConsumerCommonConfigs since it doesn't make sense outside of the local context.

In addition, I think only one of 'seek' and 'seekToTimestamp' should be provided to avoid confusion. Maybe we can add an extra validation where only one of the two should be present in the ConsumerLocalConfigs.

There is also a TODO which mentions removing 'seek' from the ConsumerCommonConfigs. I can take this up as well.

Let me know your thoughts @authorjapps

[a1shadows]

#619 This is a draft PR I have raised with what the changes might look like

[authorjapps]

@a1shadows , I had a quick look at the PR. Its heading in right direction. Thanks you 🤝

---

There is also a TODO which mentions removing 'seek' from the ConsumerCommonConfigs. I can take this up as well.

Yes, that will great to get this fixed as it may not be useful or used at all.

I think only one of 'seek' and 'seekToTimestamp' should be provided to avoid confusion.

Yes, a validation would be necessary. If you could provide that, that will be great.

Sample error message:
Only one of 'seek' and 'seekToTimestamp' should be provided, but not both. Please fix and rerun

Please go ahead and implement in your current PR 🟢

[a1shadows]

@authorjapps I'll take up these changes. I had some doubts about how to write tests for the repo. Looks like we need to run kafka locally and create topics there. Is there some documentation around this? "mvn clean install" seems to be failing locally looking for missing topics

[nirmalchandra]

@authorjapps I'll take up these changes. I had some doubts about how to write tests for the repo. Looks like we need to run kafka locally and create topics there. Is there some documentation around this? "mvn clean install" seems to be failing locally looking for missing topics

Hello @a1shadows , BUILDING might help you run tests locally.

Section -> "With tests executed(kafka)" might be what you're after.

Looks like we need to run kafka locally

Yes

and create topics there.

No.
Topics are created automatically as and when you run the tests.

[nirmalchandra]

This feature will be available in version 1.3.37 and onwards.

Only pending bits for this ticket is:

• Docs need update: steps are here