Samrat002 commented on code in PR #28012:
URL: https://github.com/apache/flink/pull/28012#discussion_r3138027587
##########
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.
+ */
Review Comment:
added as part of this change
##########
flink-filesystems/flink-s3-fs-native/src/main/java/org/apache/flink/fs/s3native/NativeS3InputStream.java:
##########
@@ -108,24 +108,7 @@ private void lazyInitialize() throws IOException {
private void openStreamAtCurrentPosition() throws IOException {
lock.lock();
try {
- if (bufferedStream != null) {
Review Comment:
added as part of this pr
--
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]
