[jira] [Commented] (KAFKA-18989) Optimize FileRecord#searchForOffsetWithSize

[Chia-Ping Tsai (Jira)]

    [ 
https://issues.apache.org/jira/browse/KAFKA-18989?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17935771#comment-17935771
 ] 

Chia-Ping Tsai commented on KAFKA-18989:
----------------------------------------


{code:java}
diff --git 
a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java 
b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
index c905a4678a..ecf6abd104 100644
--- a/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
+++ b/clients/src/main/java/org/apache/kafka/common/record/FileRecords.java
@@ -298,11 +298,15 @@ public class FileRecords extends AbstractRecords 
implements Closeable {
      * @return the batch's base offset, its physical position, and its size 
(including log overhead)
      */
     public LogOffsetPosition searchForOffsetWithSize(long targetOffset, int 
startingPosition) {
+        FileChannelRecordBatch previous = null;
         for (FileChannelRecordBatch batch : batchesFrom(startingPosition)) {
-            long offset = batch.lastOffset();
-            if (offset >= targetOffset)
-                return new LogOffsetPosition(batch.baseOffset(), 
batch.position(), batch.sizeInBytes());
+            if (targetOffset <= batch.baseOffset()) {
+                if (previous == null || previous.lastOffset() < targetOffset) 
return new LogOffsetPosition(batch);
+                return new LogOffsetPosition(previous);
+            }
+            previous = batch;
         }
+        if (previous != null && previous.lastOffset() >= targetOffset) return 
new LogOffsetPosition(previous);
         return null;
     }
 
@@ -464,6 +468,10 @@ public class FileRecords extends AbstractRecords 
implements Closeable {
         public final int position;
         public final int size;
 
+        private LogOffsetPosition(FileChannelRecordBatch batch) {
+            this(batch.baseOffset(), batch.position(), batch.sizeInBytes());
+        }
+
         public LogOffsetPosition(long offset, int position, int size) {
             this.offset = offset;
             this.position = position;
{code}


> Optimize FileRecord#searchForOffsetWithSize
> -------------------------------------------
>
>                 Key: KAFKA-18989
>                 URL: https://issues.apache.org/jira/browse/KAFKA-18989
>             Project: Kafka
>          Issue Type: Improvement
>            Reporter: Chia-Ping Tsai
>            Assignee: Chia-Ping Tsai
>            Priority: Major
>
> the `lastOffset` needs to load whole header of batch, so we should check 
> `baseOffset` instead. this approach is similar to 
> https://github.com/apache/kafka/pull/17322#discussion_r1807869552



--
This message was sent by Atlassian Jira
(v8.20.10#820010)