gaborgsomogyi commented on code in PR #28012:
URL: https://github.com/apache/flink/pull/28012#discussion_r3138108824
##########
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:
Can we test that somehow? I mean missing this can cause quite some leaks
--
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]
