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 c097f558a7567..46ac481e77388 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 @@ -135,17 +135,20 @@ public void readInto(ByteBuffer buffer, int position) throws IOException { * @return A sliced wrapper on this message set limited based on the given position and size */ public FileRecords slice(int position, int size) throws IOException { + // Cache current size in case concurrent write changes it + int currentSizeInBytes = sizeInBytes(); + if (position < 0) throw new IllegalArgumentException("Invalid position: " + position + " in read from " + this); - if (position > sizeInBytes() - start) + if (position > currentSizeInBytes - start) throw new IllegalArgumentException("Slice from position " + position + " exceeds end position of " + this); if (size < 0) throw new IllegalArgumentException("Invalid size: " + size + " in read from " + this); int end = this.start + position + size; // handle integer overflow or if end is beyond the end of the file - if (end < 0 || end >= start + sizeInBytes()) - end = start + sizeInBytes(); + if (end < 0 || end > start + currentSizeInBytes) + end = start + currentSizeInBytes; return new FileRecords(file, channel, this.start + position, end, true); } diff --git a/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java index bf799879e61a9..08dbc57d7f29d 100644 --- a/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java +++ b/clients/src/test/java/org/apache/kafka/common/record/FileRecordsTest.java @@ -36,6 +36,9 @@ import java.util.List; import java.util.Optional; import java.util.Random; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import static java.util.Arrays.asList; import static org.apache.kafka.common.utils.Utils.utf8; @@ -118,6 +121,36 @@ public void testIterationOverPartialAndTruncation() throws IOException { testPartialWrite(6, fileRecords); } + @Test + public void testSliceSizeLimitWithConcurrentWrite() throws Exception { + FileRecords log = FileRecords.open(tempFile()); + ExecutorService executor = Executors.newFixedThreadPool(2); + int maxSizeInBytes = 16384; + + try { + Future readerCompletion = executor.submit(() -> { + while (log.sizeInBytes() < maxSizeInBytes) { + int currentSize = log.sizeInBytes(); + FileRecords slice = log.slice(0, currentSize); + assertEquals(currentSize, slice.sizeInBytes()); + } + return null; + }); + + Future writerCompletion = executor.submit(() -> { + while (log.sizeInBytes() < maxSizeInBytes) { + append(log, values); + } + return null; + }); + + writerCompletion.get(); + readerCompletion.get(); + } finally { + executor.shutdownNow(); + } + } + private void testPartialWrite(int size, FileRecords fileRecords) throws IOException { ByteBuffer buffer = ByteBuffer.allocate(size); for (int i = 0; i < size; i++)