KAFKA-20841: Bound the coordinator load buffer to the FileRecords slice - #22963
Open
asdf2014 wants to merge 1 commit into
Open
KAFKA-20841: Bound the coordinator load buffer to the FileRecords slice#22963asdf2014 wants to merge 1 commit into
asdf2014 wants to merge 1 commit into
Conversation
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.
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.
Apply the same fix to TransactionStateManager, which loads
__transaction_state through a structurally identical block and fails the
same way. There 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.
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 file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
CoordinatorLoaderImpl#toReadableMemoryRecordsallocates (or reuses) abuffer of at least
loadBufferSizeand hands it toFileRecords#readIntowith its limit at capacity. That method delegatesto
Utils#readFully, which keeps reading from the channel until thedestination buffer is full or the physical end of the file is reached:
the end of the bounded slice returned by
LogSegment#readis neverenforced.
When the slice is smaller than the buffer, the copy continues into
whatever follows the slice in the segment file. With
log.preallocate=truethe active segment is physically longer than itslogical end and that region reads back as zeros, so iterating the
resulting
MemoryRecordsparses them as a batch header: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.preallocateisoff 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#appendChunkFromFilealready uses to keepreadIntowithin the range it means to copy. The limit is reset by theclear()or the reallocation that precedes the next read, so the bufferis still reused across iterations.
The same fix is applied to
TransactionStateManager, which loads__transaction_statethrough a structurally identical block and failsthe 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#readIntostop at theslice 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
readIntocalls are not exposed, becauseit only reads non-active segments, which are trimmed when they roll.
One behavior change worth noting:
LoadSummary#numBytes, reported inthe "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
loadBufferSizeand a laterread pulled in batches beyond its slice.
Testing: adds a regression test for each loader that backs the log with
a real preallocated segment (
FileRecords.openwithpreallocate=true), so the actualreadIntoandUtils#readFullyrunrather than a mock of them. Without this change the coordinator-common
test fails with the
CorruptRecordExceptionabove; the transaction testfails on a missing transaction, because that loader catches the
exception and keeps the partial state. Verified with these tests plus
the full
CoordinatorLoaderImplTestandTransactionStateManagerTestclasses, checkstyle, and
spotbugsMain, on JDK 17 and JDK 25.