KAFKA-16310 ListOffsets doesn't report the offset with maxTimestamp a…

clients/src/main/java/org/apache/kafka/common/record/FileRecords.java

@@ -355,18 +355,18 @@ public TimestampAndOffset searchForTimestamp(long targetTimestamp, int startingP
      */
     public TimestampAndOffset largestTimestampAfter(int startingPosition) {
         long maxTimestamp = RecordBatch.NO_TIMESTAMP;
-        long offsetOfMaxTimestamp = -1L;
+        long shallowOffsetOfMaxTimestamp = -1L;
         int leaderEpochOfMaxTimestamp = RecordBatch.NO_PARTITION_LEADER_EPOCH;
 
         for (RecordBatch batch : batchesFrom(startingPosition)) {
             long timestamp = batch.maxTimestamp();
             if (timestamp > maxTimestamp) {
                 maxTimestamp = timestamp;
-                offsetOfMaxTimestamp = batch.lastOffset();
+                shallowOffsetOfMaxTimestamp = batch.lastOffset();
                 leaderEpochOfMaxTimestamp = batch.partitionLeaderEpoch();
             }
         }
-        return new TimestampAndOffset(maxTimestamp, offsetOfMaxTimestamp,
+        return new TimestampAndOffset(maxTimestamp, shallowOffsetOfMaxTimestamp,
                 maybeLeaderEpoch(leaderEpochOfMaxTimestamp));
     }
 

clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java

@@ -209,7 +209,7 @@ private static FilterResult filterTo(TopicPartition partition, Iterable<MutableR
                                 partition, batch.lastOffset(), maxRecordBatchSize, filteredBatchSize);
 
                         MemoryRecordsBuilder.RecordsInfo info = builder.info();
-                        filterResult.updateRetainedBatchMetadata(info.maxTimestamp, info.offsetOfMaxTimestamp,
+                        filterResult.updateRetainedBatchMetadata(info.maxTimestamp, info.shallowOffsetOfMaxTimestamp,
                             maxOffset, retainedRecords.size(), filteredBatchSize);
                     }
                 }
@@ -399,7 +399,7 @@ public static class FilterResult {
         private int bytesRetained = 0;
         private long maxOffset = -1L;
         private long maxTimestamp = RecordBatch.NO_TIMESTAMP;
-        private long offsetOfMaxTimestamp = -1L;
+        private long shallowOffsetOfMaxTimestamp = -1L;
 
         private FilterResult(ByteBuffer outputBuffer) {
             this.outputBuffer = outputBuffer;
@@ -411,21 +411,21 @@ private void updateRetainedBatchMetadata(MutableRecordBatch retainedBatch, int n
                     retainedBatch.lastOffset(), numMessagesInBatch, bytesRetained);
         }
 
-        private void updateRetainedBatchMetadata(long maxTimestamp, long offsetOfMaxTimestamp, long maxOffset,
+        private void updateRetainedBatchMetadata(long maxTimestamp, long shallowOffsetOfMaxTimestamp, long maxOffset,
                                                  int messagesRetained, int bytesRetained) {
-            validateBatchMetadata(maxTimestamp, offsetOfMaxTimestamp, maxOffset);
+            validateBatchMetadata(maxTimestamp, shallowOffsetOfMaxTimestamp, maxOffset);
             if (maxTimestamp > this.maxTimestamp) {
                 this.maxTimestamp = maxTimestamp;
-                this.offsetOfMaxTimestamp = offsetOfMaxTimestamp;
+                this.shallowOffsetOfMaxTimestamp = shallowOffsetOfMaxTimestamp;
             }
             this.maxOffset = Math.max(maxOffset, this.maxOffset);
             this.messagesRetained += messagesRetained;
             this.bytesRetained += bytesRetained;
         }
 
-        private void validateBatchMetadata(long maxTimestamp, long offsetOfMaxTimestamp, long maxOffset) {
-            if (maxTimestamp != RecordBatch.NO_TIMESTAMP && offsetOfMaxTimestamp < 0)
-                throw new IllegalArgumentException("offset undefined for maximum timestamp " + maxTimestamp);
+        private void validateBatchMetadata(long maxTimestamp, long shallowOffsetOfMaxTimestamp, long maxOffset) {
+            if (maxTimestamp != RecordBatch.NO_TIMESTAMP && shallowOffsetOfMaxTimestamp < 0)
+                throw new IllegalArgumentException("shallowOffset undefined for maximum timestamp " + maxTimestamp);
             if (maxOffset < 0)
                 throw new IllegalArgumentException("maxOffset undefined");
         }
@@ -458,8 +458,8 @@ public long maxTimestamp() {
             return maxTimestamp;
         }
 
-        public long offsetOfMaxTimestamp() {
-            return offsetOfMaxTimestamp;
+        public long shallowOffsetOfMaxTimestamp() {
+            return shallowOffsetOfMaxTimestamp;
         }
     }
 

clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java

@@ -240,25 +240,54 @@ public MemoryRecords build() {
         return builtRecords;
     }
 
+
     /**
-     * Get the max timestamp and its offset. The details of the offset returned are a bit subtle.
-     * Note: The semantic for the offset of max timestamp is the first offset with the max timestamp if there are multi-records having same timestamp.
-     *
-     * If the log append time is used, the offset will be the first offset of the record.
-     *
-     * If create time is used, the offset will always be the offset of the record with the max timestamp.
-     *
-     * If it's NO_TIMESTAMP (i.e. MAGIC_VALUE_V0), we'll return offset -1 since no timestamp info in records.
-     *
-     * @return The max timestamp and its offset
+     * There are three cases of finding max timestamp to return:
+     * 1) version 0: The max timestamp is NO_TIMESTAMP (-1)
+     * 2) LogAppendTime: All records have same timestamp, and so the max timestamp is equal to logAppendTime
+     * 3) CreateTime: The max timestamp of record
+     * <p>
+     * Let's talk about OffsetOfMaxTimestamp. There are some paths that we don't try to find the OffsetOfMaxTimestamp
+     * to avoid expensive records iteration. Those paths include follower append and index recovery. In order to
+     * avoid inconsistent time index, we let all paths find shallowOffsetOfMaxTimestamp instead of OffsetOfMaxTimestamp.
+     * <p>
+     * Let's define the shallowOffsetOfMaxTimestamp: It is last offset of the batch having max timestamp. If there are
+     * many batches having same max timestamp, we pick up the earliest batch.
+     * <p>
+     * There are five cases of finding shallowOffsetOfMaxTimestamp to return:
+     * 1) version 0: It is always the -1
+     * 2) LogAppendTime with single batch: It is the offset of last record
+     * 3) LogAppendTime with many single-record batches: Those single-record batches have same max timestamp, so we return
+     *                                                   the base offset, which is equal to the last offset of earliest batch
+     * 4) CreateTime with single batch: We return offset of last record to follow the spec we mentioned above. Of course,
+     *                                  we do have the OffsetOfMaxTimestamp for this case, but we want to make all paths
+     *                                  find the shallowOffsetOfMaxTimestamp rather than offsetOfMaxTimestamp
+     * 5) CreateTime with many single-record batches: Each batch is composed of single record, and hence offsetOfMaxTimestamp
+     *                                                is equal to the last offset of earliest batch with max timestamp
      */
     public RecordsInfo info() {
         if (timestampType == TimestampType.LOG_APPEND_TIME) {
-            return new RecordsInfo(logAppendTime, baseOffset);
+            if (compressionType != CompressionType.NONE || magic >= RecordBatch.MAGIC_VALUE_V2)
+                // maxTimestamp => case 2
+                // shallowOffsetOfMaxTimestamp => case 2
+                return new RecordsInfo(logAppendTime, lastOffset);
+            else
+                // maxTimestamp => case 2
+                // shallowOffsetOfMaxTimestamp => case 3
+                return new RecordsInfo(logAppendTime, baseOffset);
+        } else if (maxTimestamp == RecordBatch.NO_TIMESTAMP) {
+            // maxTimestamp => case 1
+            // shallowOffsetOfMaxTimestamp => case 1
+            return new RecordsInfo(RecordBatch.NO_TIMESTAMP, -1);
         } else {
-            // For create time, we always use offsetOfMaxTimestamp for the correct time -> offset mapping
-            // If it's MAGIC_VALUE_V0, the value will be the default value: [-1, -1]
-            return new RecordsInfo(maxTimestamp, offsetOfMaxTimestamp);
+            if (compressionType != CompressionType.NONE || magic >= RecordBatch.MAGIC_VALUE_V2)
+                // maxTimestamp => case 3
+                // shallowOffsetOfMaxTimestamp => case 4
+                return new RecordsInfo(maxTimestamp, lastOffset);
+            else
+                // maxTimestamp => case 3
+                // shallowOffsetOfMaxTimestamp => case 5
+                return new RecordsInfo(maxTimestamp, offsetOfMaxTimestamp);
         }
     }
 
@@ -849,12 +878,12 @@ private long nextSequentialOffset() {
 
     public static class RecordsInfo {
         public final long maxTimestamp;
-        public final long offsetOfMaxTimestamp;
+        public final long shallowOffsetOfMaxTimestamp;
 
         public RecordsInfo(long maxTimestamp,
-                           long offsetOfMaxTimestamp) {
+                           long shallowOffsetOfMaxTimestamp) {
             this.maxTimestamp = maxTimestamp;
-            this.offsetOfMaxTimestamp = offsetOfMaxTimestamp;
+            this.shallowOffsetOfMaxTimestamp = shallowOffsetOfMaxTimestamp;
         }
     }
 

clients/src/main/java/org/apache/kafka/common/record/RecordBatch.java

@@ -21,6 +21,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Iterator;
+import java.util.Optional;
 import java.util.OptionalLong;
 
 /**
@@ -245,4 +246,23 @@ public interface RecordBatch extends Iterable<Record> {
      * @return Whether this is a batch containing control records
      */
     boolean isControlBatch();
+
+    /**
+     * iterate all records to find the offset of max timestamp.
+     * noted:
+     * 1) that the earliest offset will return if there are multi records having same (max) timestamp
+     * 2) it always returns None if the {@link RecordBatch#magic()} is equal to {@link RecordBatch#MAGIC_VALUE_V0}
+     * @return offset of max timestamp
+     */
+    default Optional<Long> offsetOfMaxTimestamp() {
+        if (magic() == RecordBatch.MAGIC_VALUE_V0) return Optional.empty();
+        long maxTimestamp = maxTimestamp();
+        try (CloseableIterator<Record> iter = streamingIterator(BufferSupplier.create())) {
+            while (iter.hasNext()) {
+                Record record = iter.next();
+                if (maxTimestamp == record.timestamp()) return Optional.of(record.offset());
+            }
+        }
+        return Optional.empty();
+    }
 }

clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsBuilderTest.java

@@ -377,8 +377,11 @@ public void buildUsingLogAppendTime(Args args) {
 
         MemoryRecordsBuilder.RecordsInfo info = builder.info();
         assertEquals(logAppendTime, info.maxTimestamp);
-        // When logAppendTime is used, the first offset of the batch will be the offset of maxTimestamp
-        assertEquals(0L, info.offsetOfMaxTimestamp);
+
+        if (args.compressionType == CompressionType.NONE && magic <= MAGIC_VALUE_V1)
+            assertEquals(0L, info.shallowOffsetOfMaxTimestamp);
+        else
+            assertEquals(2L, info.shallowOffsetOfMaxTimestamp);
 
         for (RecordBatch batch : records.batches()) {
             if (magic == MAGIC_VALUE_V0) {
@@ -413,10 +416,11 @@ public void buildUsingCreateTime(Args args) {
         }
 
         if (magic == MAGIC_VALUE_V0)
-            // in MAGIC_VALUE_V0's case, we don't have timestamp info in records, so always return -1.
-            assertEquals(-1L, info.offsetOfMaxTimestamp);
+            assertEquals(-1, info.shallowOffsetOfMaxTimestamp);
+        else if (args.compressionType == CompressionType.NONE && magic == MAGIC_VALUE_V1)
+            assertEquals(1L, info.shallowOffsetOfMaxTimestamp);
         else
-            assertEquals(1L, info.offsetOfMaxTimestamp);
+            assertEquals(2L, info.shallowOffsetOfMaxTimestamp);
 
         int i = 0;
         long[] expectedTimestamps = new long[] {0L, 2L, 1L};
@@ -494,11 +498,11 @@ public void writePastLimit(Args args) {
 
         MemoryRecordsBuilder.RecordsInfo info = builder.info();
         if (magic == MAGIC_VALUE_V0) {
+            assertEquals(-1, info.shallowOffsetOfMaxTimestamp);
             assertEquals(-1, info.maxTimestamp);
-            assertEquals(-1L, info.offsetOfMaxTimestamp);
         } else {
+            assertEquals(2L, info.shallowOffsetOfMaxTimestamp);
             assertEquals(2L, info.maxTimestamp);
-            assertEquals(2L, info.offsetOfMaxTimestamp);
         }
 
         long i = 0L;

clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java

@@ -352,7 +352,7 @@ protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
                     assertEquals(0, filterResult.messagesRetained());
                     assertEquals(DefaultRecordBatch.RECORD_BATCH_OVERHEAD, filterResult.bytesRetained());
                     assertEquals(12, filterResult.maxTimestamp());
-                    assertEquals(baseOffset + 1, filterResult.offsetOfMaxTimestamp());
+                    assertEquals(baseOffset + 1, filterResult.shallowOffsetOfMaxTimestamp());
 
                     // Verify filtered records
                     filtered.flip();
@@ -413,7 +413,7 @@ protected boolean shouldRetainRecord(RecordBatch recordBatch, Record record) {
         assertEquals(0, filterResult.messagesRetained());
         assertEquals(DefaultRecordBatch.RECORD_BATCH_OVERHEAD, filterResult.bytesRetained());
         assertEquals(timestamp, filterResult.maxTimestamp());
-        assertEquals(baseOffset, filterResult.offsetOfMaxTimestamp());
+        assertEquals(baseOffset, filterResult.shallowOffsetOfMaxTimestamp());
         assertTrue(filterResult.outputBuffer().position() > 0);
 
         // Verify filtered records
@@ -893,7 +893,10 @@ public void testFilterTo(Args args) {
         assertEquals(filtered.limit(), result.bytesRetained());
         if (magic > RecordBatch.MAGIC_VALUE_V0) {
             assertEquals(20L, result.maxTimestamp());
-            assertEquals(4L, result.offsetOfMaxTimestamp());
+            if (compression == CompressionType.NONE && magic < RecordBatch.MAGIC_VALUE_V2)
+                assertEquals(4L, result.shallowOffsetOfMaxTimestamp());
+            else
+                assertEquals(5L, result.shallowOffsetOfMaxTimestamp());
         }
 
         MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered);

core/src/main/scala/kafka/log/LocalLog.scala

@@ -406,8 +406,8 @@ class LocalLog(@volatile private var _dir: File,
     }
   }
 
-  private[log] def append(lastOffset: Long, largestTimestamp: Long, offsetOfMaxTimestamp: Long, records: MemoryRecords): Unit = {
-    segments.activeSegment.append(lastOffset, largestTimestamp, offsetOfMaxTimestamp, records)
+  private[log] def append(lastOffset: Long, largestTimestamp: Long, shallowOffsetOfMaxTimestamp: Long, records: MemoryRecords): Unit = {
+    segments.activeSegment.append(lastOffset, largestTimestamp, shallowOffsetOfMaxTimestamp, records)
     updateLogEndOffset(lastOffset + 1)
   }
 

core/src/main/scala/kafka/log/LogCleaner.scala

@@ -812,7 +812,7 @@ private[log] class Cleaner(val id: Int,
         val retained = MemoryRecords.readableRecords(outputBuffer)
         // it's OK not to hold the Log's lock in this case, because this segment is only accessed by other threads
         // after `Log.replaceSegments` (which acquires the lock) is called
-        dest.append(result.maxOffset, result.maxTimestamp, result.offsetOfMaxTimestamp, retained)
+        dest.append(result.maxOffset, result.maxTimestamp, result.shallowOffsetOfMaxTimestamp(), retained)
         throttler.maybeThrottle(outputBuffer.limit())
       }