junrao commented on code in PR #22963:
URL: https://github.com/apache/kafka/pull/22963#discussion_r4209887070
##########
clients/src/main/java/org/apache/kafka/common/record/internal/FileRecords.java:
##########
@@ -126,15 +126,21 @@ 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 these
+ * records is reached. A buffer with more room than the records left from
{@code position} has its limit lowered
+ * so that only those records are read and the bytes which follow them in
the file are left alone. The buffer is
+ * flipped before returning, so it is positioned at the first byte read
and limited to the number of bytes read.
*
* @param buffer The buffer to write the batches to
- * @param position Position in the buffer to read from
+ * @param position Position in these records to read from
+ * @throws IllegalArgumentException If the position is negative or past
the end of these records
* @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());
Review Comment:
Yes, the issue with potentially exposing uncommitted records is not limited
to readInto(). batchesFrom() has a similar issue
(https://issues.apache.org/jira/browse/KAFKA-21204). Claud suggested the
following approach to separate the classes.
```
LogFileRecords (internal physical segment) — owns the channel, start, end =
size = LEO. Keeps the full/mutating/position API: append, truncateTo,
readInto(buf,pos), batchesFrom(pos), slice. Visible only to storage internals,
recovery, compaction, tiered upload, raft. Does not implement the client-facing
Records interface, so it can never be passed where a Records is expected.
BoundedRecords (immutable, implements Records) — the window returned by
slice(...). Exposes only whole-extent iteration/transfer (batches(),
batchIterator(), writeTo, searchForTimestamp, sizeInBytes), with no
arbitrary-position re-entry, no re-widening, no mutation. This is the only
FileRecords-backed Records that flows into FetchDataInfo and responses
(consumer, replication, share-fetch — all of which already go through read).
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
