sepuri sai krishna created KAFKA-20967:
------------------------------------------
Summary: RemoteLogManager.read() leaks the remote segment
InputStream when reading a batch fails
Key: KAFKA-20967
URL: https://issues.apache.org/jira/browse/KAFKA-20967
Project: Kafka
Issue Type: Bug
Components: Tiered-Storage
Reporter: sepuri sai krishna
Assignee: sepuri sai krishna
RemoteLogManager.read()
(storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManager.java,
lines 1857-1949) opens an InputStream from the pluggable RemoteStorageManager
and closes it in a finally block that is gated on a condition which is not
satisfied when an exception is thrown:
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
is declared "throws IOException" and can also throw the unchecked
CorruptRecordException. If either is thrown, the assignment to
enrichedRecordBatch never completes, so it retains its stale value with a null
batch -- either the initial "new EnrichedRecordBatch(null, 0)" on the first
iteration, or the previous iteration's null-batch value.
The finally block then evaluates that same stale "enrichedRecordBatch.batch !=
null" as false and skips closing remoteSegInputStream, even though the stream
was successfully opened moments earlier in that iteration. The stream is leaked.
For real tiered-storage plugins (S3/GCS/HDFS-backed and similar) the underlying
resource is typically a socket or file handle, so each failed read leaks one.
This is on the consumer fetch path -- RemoteLogManager.read() is invoked for
every fetch that misses the local log and falls through to tiered storage -- so
under sustained remote I/O flakiness or a corrupted segment being repeatedly
requested, the broker leaks a handle per failed read, risking file-descriptor
or connection-pool exhaustion.
The sibling method lookupTimestamp() in the same class closes its stream
unconditionally in its finally block and does not have this problem. The
conditional guard in read() appears to have been intended to avoid a double
close on the success path, where the stream is consumed later via
Utils.readFully(), but it does not account for the exception path.
I'm happy to submit a PR for this fix.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
