SEPURI-SAI-KRISHNA opened a new pull request, #23211:
URL: https://github.com/apache/kafka/pull/23211
`RemoteLogManager.read()` opens an `InputStream` from the pluggable
`RemoteStorageManager` and closes it in a `finally` block gated on a condition
that is not satisfied when an exception is thrown:
```java
EnrichedRecordBatch enrichedRecordBatch = new EnrichedRecordBatch(null, 0);
InputStream remoteSegInputStream = null;
try {
while (enrichedRecordBatch.batch == null &&
rlsMetadataOptional.isPresent()) {
...
remoteSegInputStream =
remoteStorageManagerPlugin.get().fetchLogSegment(remoteLogSegmentMetadata,
startPos);
RemoteLogInputStream remoteLogInputStream =
getRemoteLogInputStream(remoteSegInputStream);
enrichedRecordBatch = findFirstBatch(remoteLogInputStream, offset);
// can throw
...
}
...
} finally {
if (enrichedRecordBatch.batch != null) {
Utils.closeQuietly(remoteSegInputStream,
"RemoteLogSegmentInputStream");
}
}
```
`findFirstBatch()` calls `RemoteLogInputStream.nextBatch()` in a loop. That
method declares `throws IOException` and can also throw the unchecked
`CorruptRecordException`. When either is thrown, the assignment to
`enrichedRecordBatch` never completes, so it keeps its stale null-batch value —
the initial `new EnrichedRecordBatch(null, 0)` on the first iteration, or the
previous iteration's value. The `finally` block then evaluates that same stale
`enrichedRecordBatch.batch != null` as false and skips the close, even though
the stream was opened moments earlier in that iteration.
For real tiered-storage plugins the leaked resource is typically a socket or
file handle. `read()` is on the consumer fetch path — invoked for every fetch
that misses the local log and falls through to tiered storage — so sustained
remote I/O flakiness, or a corrupted segment being repeatedly requested, leaks
one handle per failed read and risks file-descriptor or connection-pool
exhaustion.
The sibling method `lookupTimestamp()` in the same class closes its stream
unconditionally and does not have this problem.
### Changes
The conditional guard appears to have been intended to avoid a double close
on the path where the loop advances to the next segment, since that path
already closes the stream before looking up the next segment's metadata. Rather
than widening the condition, this sets `remoteSegInputStream = null` after that
in-loop close and makes the `finally` close unconditional. `Utils.closeQuietly`
ignores a null argument, so each stream is now closed exactly once on every
path:
- a batch is found — the stream stays open for the subsequent
`Utils.readFully`, and `finally` closes it
- the loop exhausts all segments without finding a batch — each stream is
closed in the loop, and `finally` sees null
- `findFirstBatch` throws — the stream is still open and non-null, so
`finally` now closes it (this is the leak being fixed)
- `lookupPositionForOffset` or `fetchLogSegment` throws on a later iteration
— the previous stream was already closed and nulled, so `finally` sees null
- anything after the loop throws (`readFully`, `addAbortedTransactions`) —
the batch is non-null and the stream is open, so `finally` closes it
### Testing
Added
`RemoteLogManagerTest#testReadClosesRemoteSegmentInputStreamWhenReadingBatchFails`,
which stubs `nextBatch()` to throw `CorruptRecordException` — standing in for
a corrupted remote segment or a plugin I/O error part-way through a segment —
and asserts the segment stream is closed even though no batch was read.
The test was confirmed to fail against the unfixed code, with Mockito
reporting `Wanted but not invoked: fileInputStream.close(); Actually, there
were zero interactions with this mock`, and to pass with the fix applied. The
full `RemoteLogManagerTest` suite passes (96 tests, 0 failures), covering the
existing success and log-compaction paths through `read()` that exercise the
other close paths described above.
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]
