KAFKA-15499: Fix the flaky DeleteSegmentsDueToLogStartOffsetBreach test (#14439)

DeleteSegmentsDueToLogStartOffsetBreach configures the segment such that it can hold at-most 2 record-batches. And, it asserts that the local-log-start-offset based on the assumption that each segment will contain exactly two messages.

During leader switch, the segment can get rotated and may not always contain two records. Previously, we were checking whether the expected local-log-start-offset is equal to the base-offset-of-the-first-local-log-segment. With this patch, we will scan the first local-log-segment for the expected offset.

Reviewers: Divij Vaidya <diviv@amazon.com>

storage/src/test/java/org/apache/kafka/server/log/remote/storage/LocalTieredStorageTest.java

@@ -43,7 +43,6 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.nio.file.StandardOpenOption;
-import java.text.NumberFormat;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
@@ -595,14 +594,6 @@ public final class LocalTieredStorageTest {
         private static final byte[] PRODUCER_SNAPSHOT_FILE_BYTES = "pid".getBytes();
         private static final byte[] LEADER_EPOCH_CHECKPOINT_FILE_BYTES = "0\n2\n0 0\n2 12".getBytes();
 
-        private static final NumberFormat OFFSET_FORMAT = NumberFormat.getInstance();
-
-        static {
-            OFFSET_FORMAT.setMaximumIntegerDigits(20);
-            OFFSET_FORMAT.setMaximumFractionDigits(0);
-            OFFSET_FORMAT.setGroupingUsed(false);
-        }
-
         private final Path segmentPath = Paths.get("local-segments");
         private long baseOffset = 0;
 
@@ -621,7 +612,7 @@ public final class LocalTieredStorageTest {
         }
 
         LogSegmentData nextSegment(final byte[]... data) {
-            final String offset = OFFSET_FORMAT.format(baseOffset);
+            final String offset = LogFileUtils.filenamePrefixFromOffset(baseOffset);
 
             try {
                 final FileChannel channel = FileChannel.open(

storage/src/test/java/org/apache/kafka/tiered/storage/utils/BrokerLocalStorage.java

@@ -17,6 +17,8 @@
 package org.apache.kafka.tiered.storage.utils;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.record.FileLogInputStream;
+import org.apache.kafka.common.record.FileRecords;
 import org.apache.kafka.common.utils.Time;
 import org.apache.kafka.common.utils.Timer;
 import org.apache.kafka.common.utils.Utils;
@@ -67,12 +69,14 @@ public final class BrokerLocalStorage {
     public void waitForEarliestLocalOffset(TopicPartition topicPartition,
                                            Long offset) {
         Function<OffsetHolder, Optional<String>> relativePosFunc = offsetHolder -> {
-            if (offsetHolder.firstLogFileBaseOffset < offset) {
-                return Optional.of("smaller than");
+            Optional<String> result = Optional.empty();
+            if (offsetHolder.firstLogFileBaseOffset < offset &&
+                    !isOffsetPresentInFirstLocalSegment(topicPartition, offsetHolder.firstLogFileBaseOffset, offset)) {
+                result = Optional.of("smaller than");
             } else if (offsetHolder.firstLogFileBaseOffset > offset) {
-                return Optional.of("ahead of");
+                result = Optional.of("ahead of");
             }
-            return Optional.empty();
+            return result;
         };
         waitForOffset(topicPartition, offset, relativePosFunc);
     }
@@ -90,10 +94,12 @@ public final class BrokerLocalStorage {
     public void waitForAtLeastEarliestLocalOffset(TopicPartition topicPartition,
                                                   Long offset) {
         Function<OffsetHolder, Optional<String>> relativePosFunc = offsetHolder -> {
-            if (offsetHolder.firstLogFileBaseOffset < offset) {
-                return Optional.of("smaller than");
+            Optional<String> result = Optional.empty();
+            if (offsetHolder.firstLogFileBaseOffset < offset &&
+                    !isOffsetPresentInFirstLocalSegment(topicPartition, offsetHolder.firstLogFileBaseOffset, offset)) {
+                result = Optional.of("smaller than");
             }
-            return Optional.empty();
+            return result;
         };
         waitForOffset(topicPartition, offset, relativePosFunc);
     }
@@ -119,6 +125,37 @@ public final class BrokerLocalStorage {
         }
     }
 
+    /**
+     * Check if the given offset is present in the first local segment of the given topic-partition.
+     * @param topicPartition The topic-partition to check.
+     * @param firstLogFileBaseOffset The base offset of the first local segment.
+     * @param offsetToSearch The offset to search.
+     * @return true if the offset is present in the first local segment, false otherwise.
+     */
+    private boolean isOffsetPresentInFirstLocalSegment(TopicPartition topicPartition,
+                                                       Long firstLogFileBaseOffset,
+                                                       Long offsetToSearch)  {
+        if (offsetToSearch < firstLogFileBaseOffset) {
+            return false;
+        }
+        if (offsetToSearch.equals(firstLogFileBaseOffset)) {
+            return true;
+        }
+        File partitionDir = new File(brokerStorageDirectory.getAbsolutePath(), topicPartition.toString());
+        File firstSegmentFile = new File(partitionDir.getAbsolutePath(),
+                LogFileUtils.filenamePrefixFromOffset(firstLogFileBaseOffset) + LogFileUtils.LOG_FILE_SUFFIX);
+        try (FileRecords fileRecords = FileRecords.open(firstSegmentFile, false)) {
+            for (FileLogInputStream.FileChannelRecordBatch batch : fileRecords.batches()) {
+                if (batch.baseOffset() <= offsetToSearch && batch.lastOffset() >= offsetToSearch) {
+                    return true;
+                }
+            }
+        } catch (final IOException ex) {
+            return false;
+        }
+        return false;
+    }
+
     public void eraseStorage() throws IOException {
         for (File file : Objects.requireNonNull(brokerStorageDirectory.listFiles())) {
             Utils.delete(file);