Samrat002 commented on code in PR #28012:
URL: https://github.com/apache/flink/pull/28012#discussion_r3139724931
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -143,27 +126,35 @@ private void openStreamAtCurrentPosition() throws
IOException {
currentStream = s3Client.getObject(requestBuilder.build());
bufferedStream = new BufferedInputStream(currentStream,
readBufferSize);
} catch (Exception e) {
- if (bufferedStream != null) {
- try {
- bufferedStream.close();
- } catch (IOException ignored) {
- }
- bufferedStream = null;
- }
- if (currentStream != null) {
- try {
- currentStream.close();
- } catch (IOException ignored) {
- }
- currentStream = null;
- }
+ releaseStream();
throw new IOException("Failed to open S3 stream for " +
bucketName + "/" + key, e);
}
} finally {
lock.unlock();
}
}
+ /**
+ * Aborts the in-flight S3 response and releases both stream references.
+ *
+ * <p>We deliberately skip {@link BufferedInputStream#close()} because it
delegates to the
+ * underlying {@link ResponseInputStream#close()}, which drains remaining
bytes.
+ */
+ private void releaseStream() {
+ // Drop the wrapper without closing it; closing would trigger the
drain path.
+ bufferedStream = null;
Review Comment:
Fair point, tried adding a test for it.
TrackingInputStream now tracks three things: wasAborted, wasClosed, and
wasAbortedBeforeClose (set inside close() based on whether abort already ran).
New test `closeAbortsAndThenClosesUnderlyingStream` asserts all three after
the stream is closed, and I've applied the same three assertions on the seek()
and skip() paths too, since those also swap out the stream mid-life.
The three together catch both failure modes:
1. drop close() → wasClosed fails (connection pool leak)
2. reverse the order → wasAbortedBeforeClose fails (drain regression)
3. drop abort() → wasAborted fails
So any future refactor that reintroduces the leak will break exactly one of
them.
--
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]
