[FLINK-39484][filesystem/s3] NativeS3InputStream: graceful abort on early close; close() first for connection reuse

[macdoor]

What is the purpose of the change

Fix ConnectionClosedException: Premature end of Content-Length delimited message body
when reading large objects (e.g. Parquet) through flink-s3-fs-native against
S3-compatible endpoints (e.g. MinIO), especially after seek(), skip(), or when the
stream is closed before all bytes are consumed.

Brief change log

• Seek/skip: When discarding a partially-read GetObject body after seek() or skip(),
  call ResponseInputStream.abort() immediately so the SDK does not attempt to read and
  drain the remainder of the response.
• Close (normal path): Attempt close() first to preserve HTTP connection reuse for
  well-behaved S3 servers. If the server closes the connection early (MinIO pattern),
  ConnectionClosedException is caught, treated as non-fatal, escalated to WARN, and
  abort() is called as fallback -- because the connection is already broken and cannot be
  reused anyway, so aborting carries no additional performance penalty.
• isPrematureEndOfMessage(IOException): helper that identifies MinIO/S3-compatible
  early-connection-close by checking for Premature end of Content-Length,
  Connection closed, or ConnectionClosed in the exception message.

This is the optimal approach:

• No connection-pool penalty for correct S3/AWS behavior -- close() reuses connections
• No task failures when MinIO closes early -- exception caught, WARN logged, task survives
• Seek/skip still aborts immediately -- correct semantics regardless of server behavior

Verifying this change

Verified against MinIO with large Parquet reads and repeated seek() operations: the
ConnectionClosedException is replaced by a clean WARN log and the job completes
successfully. Azure Pipelines for flink-s3-fs-native (see CI on this PR).

Does this pull request potentially affect one of the following parts?

• Dependencies: no
• Public API: no
• Serializers: no
• Runtime per-record code paths: no
• Deployment or recovery: no
• The S3 file system connector: yes

Documentation

• New feature: no

JIRA

https://issues.apache.org/jira/browse/FLINK-39484

[flinkbot]

CI report:

• a1e4c2b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[spuru9]

@macdoor spotless is failing for the build, also there are some changes in the license header which I believe in unintended.

[macdoor]

@spuru9 Thanks — Spotless is fixed (spotless:apply + spotless:check), and the license header in NativeS3InputStream.java is aligned with the standard ASF template.

[macdoor]

@flinkbot run azure

[Samrat002]

Thank you @macdoor for raising the patch.

i have a fundamental query associate with this change.

is the problem true for AWS S3?

As per my analysis , Premature end of Content-Length delimited message body can only occur when the server closes the TCP connection before delivering all Content-Length bytes. This is an HTTP/1.1 protocol violation.
AWS S3 does not do this. S3 always delivers exactly the number of bytes promised in Content-Length. The drain loop in Apache HttpClient's ContentLengthInputStream.close() will always succeed against real S3. It may be slow (reading GBs to discard), but it will never hit a premature EOF.

The only valid concern for real S3 is performance. I see probable improvement is draining large amounts of unread data on close() just to reuse a connection.

The SDK JavaDoc says it plainly:

"If it is not desired to read remaining data from the stream, you can explicitly abort the connection via abort() instead. This will close the underlying connection and require establishing a new HTTP connection on subsequent requests which may outweigh the cost of reading the additional data."

reference : https://github.com/aws/aws-sdk-java-v2/blob/master/core/sdk-core/src/main/java/software/amazon/awssdk/core/ResponseInputStream.java#L52