Re: [PR] KAFKA-21234: FileRecords#readInto should honor slice boundaries [kafka]

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

junrao commented on code in PR #23737:
URL: https://github.com/apache/kafka/pull/23737#discussion_r4252711604


##########
clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java:
##########
@@ -128,15 +126,19 @@ public FileChannel channel() {
     }
 
     /**
-     * Read log batches into the given buffer until there are no bytes remaining in the buffer or the end of the file
-     * is reached.
+     * Read log batches into the given buffer until there are no bytes remaining in the buffer or the end of
+     * this FileRecords object is reached.
      *
      * @param buffer The buffer to write the batches to
-     * @param position Position in the buffer to read from
+     * @param position Position to start reading from, relative to the start of this FileRecords object
      * @throws IOException If an I/O error occurs, see {@link FileChannel#read(ByteBuffer, long)} for details on the
      * possible exceptions
      */
     public void readInto(ByteBuffer buffer, int position) throws IOException {
+        int available = availableBytes(position, buffer.remaining());
+        if (buffer.remaining() > available) {
+            buffer.limit(buffer.position() + available);
+        }

Review Comment:
   Claude found the following issue with RecordsIterator.createMemoryRecords(). The following comment is no longer true can throw a NullPointerException.
   
   >         // firstBatchSize() is always non-null because the minimum buffer is HEADER_SIZE_UP_TO_MAGIC.
   
   ```
   Example: slice(0, B1 + 10) with batchSize = B1.
   The first read gets B1, which is consumed.
   compact() leaves an empty buffer.
   The capped read returns only 10 bytes.
   firstBatchSize() returns null because fewer than HEADER_SIZE_UP_TO_MAGIC (17) bytes remain, and unboxing it into int firstBatchSize throws.
   The old code read past the slice instead, which was also wrong but didn't crash. The comment at :139 ("always non-null because the minimum buffer is HEADER_SIZE_UP_TO_MAGIC") is no longer true.
   This can't happen in practice today. Every raft caller I checked ends its slice on a batch boundary: the high-watermark or log-end positions for fireHandleCommit (KafkaRaftClient:442) and maybeLoadLog (KRaftControlRecordStateMachine:228), and whole files for snapshots. The iterator should still handle a partial tail.
   Fix: once bytesRead == records.sizeInBytes(), the slice is used up, so return memoryRecords as is. Its batch iterator yields nothing and iteration ends. This also skips the "grow buffer" path, which would otherwise allocate a buffer of firstBatchSize for a tail that can't be completed.
   
   
   Integer firstBatchSize = memoryRecords.firstBatchSize();
   if (firstBatchSize == null || firstBatchSize <= buffer.remaining() || bytesRead >= fileRecords.sizeInBytes())
       return memoryRecords;
   ```
   



##########
clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java:
##########
@@ -128,15 +126,19 @@ public FileChannel channel() {
     }
 
     /**
-     * Read log batches into the given buffer until there are no bytes remaining in the buffer or the end of the file
-     * is reached.
+     * Read log batches into the given buffer until there are no bytes remaining in the buffer or the end of
+     * this FileRecords object is reached.
      *
      * @param buffer The buffer to write the batches to
-     * @param position Position in the buffer to read from
+     * @param position Position to start reading from, relative to the start of this FileRecords object
      * @throws IOException If an I/O error occurs, see {@link FileChannel#read(ByteBuffer, long)} for details on the
      * possible exceptions
      */
     public void readInto(ByteBuffer buffer, int position) throws IOException {
+        int available = availableBytes(position, buffer.remaining());
+        if (buffer.remaining() > available) {

Review Comment:
   This check is unnecessary since it's always true. Also, the error message in availableBytes() seems very specific to the slice() call. Claude suggested the following change.
   
   ```
   if (position < 0 || position > size)
       throw new IllegalArgumentException("Read position " + position + " is outside " + this);
   buffer.limit(buffer.position() + Math.min(buffer.remaining(), size - position));
   ```



##########
clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java:
##########
@@ -770,8 +832,19 @@ private static List<RecordBatch> batches(Records buffer) {
         return TestUtils.toList(buffer.batches());
     }
 
+    private static List<Long> baseOffsets(ByteBuffer buffer) {

Review Comment:
   No need for the new helper. The caller can just do `batches(MemoryRecords.readableRecords(buffer))`.



##########
clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java:
##########
@@ -128,15 +126,19 @@ public FileChannel channel() {
     }
 
     /**
-     * Read log batches into the given buffer until there are no bytes remaining in the buffer or the end of the file
-     * is reached.
+     * Read log batches into the given buffer until there are no bytes remaining in the buffer or the end of
+     * this FileRecords object is reached.
      *
      * @param buffer The buffer to write the batches to
-     * @param position Position in the buffer to read from
+     * @param position Position to start reading from, relative to the start of this FileRecords object

Review Comment:
   Could we add below that this method now throws IllegalArgumentException ?



-- 
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