junrao commented on code in PR #22963:
URL: https://github.com/apache/kafka/pull/22963#discussion_r4169492899
##########
clients/src/test/java/org/apache/kafka/common/record/internal/FileRecordsTest.java:
##########
@@ -84,6 +84,69 @@ public void cleanup() throws IOException {
this.fileRecords.close();
}
+ /**
+ * A buffer with more room than the records left from the given position
is filled only up to
+ * their end, so that the bytes which follow them in the file are not
read. Those bytes exist
+ * whenever the records are a slice of a larger file, or the segment was
preallocated.
+ */
+ @Test
+ public void testReadIntoStopsAtTheEndOfTheRecords() throws Exception {
+ // The bytes which follow a slice are usually more records rather than
zeros, in which case
+ // reading past it stays silent instead of failing, so cover both.
+ int firstBatchSize =
fileRecords.batches().iterator().next().sizeInBytes();
+ ByteBuffer sliced = ByteBuffer.allocate(1024);
+ fileRecords.slice(0, firstBatchSize).readInto(sliced, 0);
+ assertEquals(firstBatchSize, sliced.remaining());
+
+ // A slice which does not start at the beginning of the file reads
from its own start.
+ FileRecords tail = fileRecords.slice(firstBatchSize,
fileRecords.sizeInBytes() - firstBatchSize);
+ sliced.clear();
+ tail.readInto(sliced, 0);
+ assertEquals(fileRecords.sizeInBytes() - firstBatchSize,
sliced.remaining());
+ assertEquals(values.length - 1,
recordCount(MemoryRecords.readableRecords(sliced)));
+
+ try (FileRecords preallocated = FileRecords.open(tempFile(), false,
1024, true)) {
+ append(preallocated, values);
+ int sizeInBytes = preallocated.sizeInBytes();
+ // The file is longer than the records, otherwise this proves
nothing.
+ assertTrue(preallocated.file().length() > sizeInBytes);
+
+ ByteBuffer buffer = ByteBuffer.allocate(1024);
+ preallocated.readInto(buffer, 0);
+ assertEquals(sizeInBytes, buffer.remaining());
+ // The bytes past the records would be read as a corrupt batch
header, so check that
+ // what was read is exactly the records that were appended.
+ assertEquals(values.length,
recordCount(MemoryRecords.readableRecords(buffer)));
+
+ FileRecords slice = preallocated.slice(0, sizeInBytes - 1);
+ buffer.clear();
+ slice.readInto(buffer, 0);
+ assertEquals(sizeInBytes - 1, buffer.remaining());
Review Comment:
Should we check the buffer now has one fewer record?
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -1351,6 +1400,33 @@ class TransactionStateManagerTest {
assertEquals(1,
transactionManager.transactionMetadataCache(partitionId).coordinatorEpoch)
}
+ @Test
+ def testLoadTransactionMetadataFromPreallocatedSegment(): Unit = {
+ txnMetadata1.state(TransactionState.PREPARE_COMMIT)
+ txnMetadata1.addPartitions(util.Set.of(new TopicPartition("topic1", 0)))
+ txnMetadata2.state(TransactionState.ONGOING)
+ txnMetadata2.addPartitions(util.Set.of(new TopicPartition("topic1", 1)))
+
+ // The two transactions are in different segments, so they are loaded by
successive reads. A
+ // read that is not bounded to its slice runs into the preallocated tail
of the first segment
+ // and aborts the load, which would leave the second transaction missing
from the cache.
+ val firstSegmentRecords = MemoryRecords.withRecords(0L, Compression.NONE,
+ new SimpleRecord(txnMessageKeyBytes1,
TransactionLog.valueToBytes(txnMetadata1.prepareNoTransit(), TV_2)))
+ val secondSegmentRecords = MemoryRecords.withRecords(1L, Compression.NONE,
+ new SimpleRecord(txnMessageKeyBytes2,
TransactionLog.valueToBytes(txnMetadata2.prepareNoTransit(), TV_2)))
+
+ prepareTxnLogFromPreallocatedSegment(topicPartition, firstSegmentRecords,
secondSegmentRecords)
+
+ transactionManager.loadTransactionsForTxnTopicPartition(partitionId,
coordinatorEpoch = 1, (_, _, _, _) => ())
+
+ assertEquals(0, transactionManager.loadingPartitions.size)
+
assertTrue(transactionManager.transactionMetadataCache.contains(partitionId))
+ val txnMetadataPool =
transactionManager.transactionMetadataCache(partitionId).metadataPerTransactionalId
+ assertTrue(txnMetadataPool.containsKey(transactionalId1))
+ assertTrue(txnMetadataPool.containsKey(transactionalId2))
+ assertEquals(txnMetadata2.state,
txnMetadataPool.get(transactionalId2).state)
Review Comment:
should we check the state for transactionalId1 too?
##########
core/src/test/scala/unit/kafka/coordinator/transaction/TransactionStateManagerTest.scala:
##########
@@ -1258,6 +1264,49 @@ class TransactionStateManagerTest {
})
}
+ /**
+ * Backs the log with two segments, the first of them preallocated: it is
physically longer than
+ * the records appended to it, and the region past its logical end reads
back as zeros. Each
Review Comment:
This doesn't match the real scenario. When we roll a segment, the segment is
trimmed to its actual size. So, a typical scenario is that a preallocated
segment follows a non-preallocated one.
##########
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:
Even with this fix, this api is a bit dangerous. If called on a non-sliced
FileRecords, it could expose uncommitted records. So, it would be useful to
document that client-facing callers should not use this api only on non-sliced
active FileRecords.
##########
raft/src/main/java/org/apache/kafka/raft/internals/RecordsIterator.java:
##########
@@ -145,7 +145,8 @@ private MemoryRecords createMemoryRecords(FileRecords
fileRecords) {
MemoryRecords memoryRecords = readFileRecords(fileRecords, buffer);
- // firstBatchSize() is always non-null because the minimum buffer is
HEADER_SIZE_UP_TO_MAGIC.
+ // firstBatchSize() is always non-null because the records end on a
batch boundary, so the
+ // read returns either at least a batch header or nothing at all once
they are exhausted.
Review Comment:
How about the following?
```
// firstBatchSize() is non-null here: this method is only reached
when there are still
// unread bytes (see nextBatches), and the records end on a batch
boundary, so the read
// above always returns at least one complete batch header.
```
--
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]
