SEPURI-SAI-KRISHNA opened a new pull request, #23206:
URL: https://github.com/apache/kafka/pull/23206
`RemoteLogInputStream.nextBatch()` reads a 4-byte batch-size field off the
`InputStream` returned by the pluggable `RemoteStorageManager` and uses it to
size an allocation, checking only a lower bound:
```java
int size = logHeaderBuffer.getInt(SIZE_OFFSET);
// V0 has the smallest overhead, stricter checking is done later
if (size < LegacyRecord.RECORD_OVERHEAD_V0)
throw new CorruptRecordException(...);
int bufferSize = LOG_OVERHEAD + size;
ByteBuffer buffer = ByteBuffer.allocate(bufferSize); // no upper bound on
size
```
Nothing verifies that `size` is not absurdly large before allocating. Since
`size` is a signed int read directly from the remote segment's bytes, a
corrupted or bit-rotted segment — or a buggy `RemoteStorageManager`
implementation — can drive a single allocation of up to ~2GB, per batch read.
The sibling class `ByteBufferLogInputStream` parses the identical
length-prefixed header format and already guards against exactly this:
```java
if (recordSize > maxMessageSize)
throw new CorruptRecordException(String.format(
"Record size %d exceeds the largest allowable message size (%d).",
recordSize, maxMessageSize));
```
`RemoteLogInputStream` has no equivalent check, and is the more exposed of
the two: `ByteBufferLogInputStream` only slices an already-in-memory,
already-bounded `ByteBuffer`, whereas `RemoteLogInputStream` calls
`ByteBuffer.allocate()` on the untrusted value before confirming the stream
even holds that many bytes.
Both reachable call sites are hot server-side paths, not test-only code:
- `RemoteLogManager.read()` — consumer fetches that fall through to tiered
storage
- `RemoteLogManager.lookupTimestamp()` — offset-by-timestamp lookups against
tiered segments
### Changes
- `RemoteLogInputStream` now takes a `maxMessageSize` in its constructor and
throws `CorruptRecordException` when the declared batch size exceeds it,
mirroring the existing `ByteBufferLogInputStream` check so the two paths
validate consistently.
- `RemoteLogManager` threads `UnifiedLog.config().maxMessageSize()` into
both call sites. That value is already available at each one, so no new
configuration or plumbing is introduced.
The bound is the same one the local-log read path already enforces, so a
segment that could legitimately be written locally still reads back from remote
storage unchanged; only sizes that could never correspond to a valid batch are
rejected.
### Testing
- Added
`RemoteLogInputStreamTest#testNextBatchThrowsCorruptRecordExceptionWhenSizeExceedsMaxMessageSize`,
which writes a valid record, patches the on-disk size field to exceed a
deliberately small `maxMessageSize`, and asserts `CorruptRecordException` is
raised instead of the oversized allocation being attempted. This mirrors
`ByteBufferLogInputStreamTest#iteratorRaisesOnTooLargeRecords`, keeping the two
classes' coverage symmetric.
- Existing `RemoteLogInputStreamTest` cases were updated for the new
constructor and pass unchanged (61 tests, 0 failures), confirming that valid
segments across all magic values and compression types still read back
correctly.
- `RemoteLogManagerTest` passes (95 tests, 0 failures), covering both
updated call sites including the log-compaction retry path in `read()`.
Co-Authored-By: Claude Sonnet 5
--
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]
