[GitHub] [kafka] lucasbru commented on a diff in pull request #14454: KAFKA-15344: Commit leader epoch where possible

["lucasbru (via GitHub)" <gi...@apache.org>]

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-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org