[FLINK-39533][s3] Use abort() instead of drain on close/seek when remaining bytes exceed threshold in NativeS3InputStream

[Samrat002]

What is the purpose of the change

NativeS3InputStream calls ResponseInputStream.close() when releasing streams during seek(), skip(), and close() operations. Apache HttpClient's close() implementation
drains all remaining bytes from the response body to enable HTTP connection reuse. For large S3 objects where only a small portion was read (e.g., checkpoint metadata from a
multi-GB state file), this drains potentially gigabytes of data over the network — causing severe latency during checkpoint restore and seek-heavy read patterns.

The AWS SDK v2 ResponseInputStream JavaDoc explicitly recommends
calling abort() when remaining data is not needed. This PR replaces close() with abort() in the stream release path.

Brief change log

• Added releaseStream() method to NativeS3InputStream that calls abort() instead of close() on the underlying ResponseInputStream, and drops the BufferedInputStream
  wrapper without closing it (closing would delegate to the drain path)

• openStreamAtCurrentPosition() and close() now use releaseStream() for stream cleanup

• Added NativeS3InputStreamTest with 8 tests covering abort lifecycle, data correctness, position tracking, and error paths

  Verifying this change

  This change added tests and can be verified as follows:

• Unit Test

• Manually validated end-to-end on a local Flink 2.3-SNAPSHOT cluster with a stateful job writing checkpoints (up to 199MB) to S3, triggering a savepoint, restoring from it, and confirming checkpoints completed successfully after restore with zero S3/stream errors

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

  • Dependencies (does it add or upgrade a dependency): no
  • The public API, i.e., is any changed class annotated with @Public(Evolving): no
  • The serializers: no
  • The runtime per-record code paths (performance sensitive): no
  • Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
  • The S3 file system connector: yes

  Documentation

  • Does this pull request introduce a new feature? no
  • If yes, how is the feature documented? not applicable

  ---

  Was generative AI tooling used to co-author this PR?

  • Yes (please specify the tool below)

  took help from claude Sonet for 2nd commit

[Samrat002]

cc: @gaborgsomogyi

[flinkbot]

CI report:

• 26d217c Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[gaborgsomogyi]

I think the following 2 blocks can be flipped (lazyInitialize can come after the if confition) to cover an edge case. We can say 100% sure we've not read anything when position >= contentLength. We don't need any stream to say it, right? If you agree then there are multiple occasions

            lazyInitialize();
            if (position >= contentLength) {
                return -1;
            }

[gaborgsomogyi]

I think seek(contentLength) triggers S3 416 when a stream is open.

When seeking to exactly contentLength (a valid EOF-boundary seek) with an open stream, the current code calls openStreamAtCurrentPosition() unconditionally. This issues a range request bytes=contentLength-, which starts one byte past the last byte. Per RFC 7233, this range is unsatisfiable and S3 returns 416.

The fix: when seeking to contentLength, release the existing stream but do not reopen it. Any subsequent read() will return -1 immediately by the position check, so an open stream at EOF serves no purpose.

  if (desired != position) {                                                                                                                                                                                                                        
      position = desired;                                                                                                                                                                                                                           
      if (currentStream != null) {                                                                                                                                                                                                                  
          if (desired >= contentLength) {                                                                                                                                                                                                           
              releaseStreams();                                                                                                                                                                                                                     
          } else {                                                                                                                                                                                                                                  
              openStreamAtCurrentPosition();                                                                                                                                                                                                        
          }                                                                                                                                                                                                                                         
      }                                                                                                                                                                                                                                             
  }                                                                                                                                                                                                                                                 

releaseStreams() is called directly (not via openStreamAtCurrentPosition()) because seek() already holds the lock. The >= guard instead of == is defensive — desired > contentLength is already rejected earlier, but this makes the branch self-contained against future refactoring.

The same root cause exists in skip() and in read()/read(byte[],int,int) where lazyInitialize() is called before the EOF position check - but those are separate fixes.