asdf2014 opened a new pull request, #22963:
URL: https://github.com/apache/kafka/pull/22963
`CoordinatorLoaderImpl#toReadableMemoryRecords` allocates (or reuses) a
buffer of at least `loadBufferSize` and hands it to `FileRecords#readInto` with
its limit at capacity. That method delegates to `Utils#readFully`, which keeps
reading from the channel until the destination buffer is full or the physical
end of the file is reached: the end of the bounded slice returned by
`LogSegment#read` is never enforced.
When the slice is smaller than the buffer, the copy continues into whatever
follows the slice in the segment file. With `log.preallocate=true` the active
segment is physically longer than its logical end and that region reads back as
zeros, so iterating the resulting `MemoryRecords` parses them as a batch header:
```
CorruptRecordException: Record size 0 is less than the minimum record
overhead (14)
```
even though every record in the slice is valid, and the coordinator shard
transitions to FAILED. This affects the group and share coordinators, which
both load through this class. `log.preallocate` is off by default, but it is
the documented recommendation for Kafka on Windows.
Bound the buffer to the size of the slice before reading into it, the
technique `LogSegment#appendChunkFromFile` already uses to keep `readInto`
within the range it means to copy. The limit is reset by the `clear()` or the
reallocation that precedes the next read, so the buffer is still reused across
iterations.
The same fix is applied to `TransactionStateManager`, which loads
`__transaction_state` through a structurally identical block and fails the same
way. That path is worse: the exception is caught and logged while the partially
loaded state is kept, so the partition is advertised as loaded with the tail of
the log missing. The duplicated `buffer.clear()` at that call site is removed.
The alternative would be to make `FileRecords#readInto` stop at the slice
boundary. That is a wider change: it contradicts the method's documented
contract ("until there are no bytes remaining in the buffer or the end of the
file is reached") and affects its other callers, so this patch bounds the
buffer at the call sites that need it instead. For reference, the log cleaner's
`readInto` calls are not exposed, because it only reads non-active segments,
which are trimmed when they roll.
One behavior change worth noting: `LoadSummary#numBytes`, reported in the
"Loaded N records which total to M bytes" log line, now counts exactly the
bytes of the slice. It could previously be inflated when an oversized batch had
grown the buffer past `loadBufferSize` and a later read pulled in batches
beyond its slice.
Testing: adds a regression test for each loader that backs the log with a
real preallocated segment (`FileRecords.open` with `preallocate=true`), so the
actual `readInto` and `Utils#readFully` run rather than a mock of them. Without
this change the coordinator-common test fails with the `CorruptRecordException`
above; the transaction test fails on a missing transaction, because that loader
catches the exception and keeps the partial state. Verified with these tests
plus the full `CoordinatorLoaderImplTest` and `TransactionStateManagerTest`
classes, checkstyle, and `spotbugsMain`, on JDK 17 and JDK 25.
--
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]
