lucasbru commented on code in PR #14454:
URL: https://github.com/apache/kafka/pull/14454#discussion_r1337139205
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -431,11 +433,20 @@ public Map<TopicPartition, OffsetAndMetadata>
prepareCommit() {
}
}
- private Long findOffset(final TopicPartition partition) {
+ private OffsetAndMetadata findOffsetAndMetadata(final TopicPartition
partition) {
Long offset = partitionGroup.headRecordOffset(partition);
+ Optional<Integer> leaderEpoch =
partitionGroup.headRecordLeaderEpoch(partition);
+ final long partitionTime =
partitionGroup.partitionTimestamp(partition);
if (offset == null) {
try {
offset = mainConsumer.position(partition);
+ // If we happen to commit the next offset after the last
consumed record, use it's
+ // leader epoch. Otherwise, we do not know the leader epoch.
+ if (consumedOffsets.containsKey(partition) && offset ==
consumedOffsets.get(partition) + 1) {
+ leaderEpoch = consumedLeaderEpochs.get(partition);
+ } else {
+ leaderEpoch = Optional.empty();
Review Comment:
Here, we'd insert the code for fetching the leader epoch for the current
consumer position, if we make it possible in the consumer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
