Added BufferedInputStream to allow mark and reset ops during IO errors

[vikasvb90]

Description

In S3 SDK v2, if stream supplied by the caller doesn't support mark and reset then in case of IO errors, entire upload fails instead of just retrying the buffer. This change wraps the provided stream with BufferedInputStream to always allow mark and reset for efficient uploads.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[Bukhtawar]

Can you share the thread dumps or profiling samples to understand the thread pool size changes

[github-actions]

Compatibility status:

Checks if related components are compatible with change 9bc21d7

Incompatible components

Incompatible components: [https://github.com/opensearch-project/asynchronous-search.git, https://github.com/opensearch-project/ml-commons.git]

Skipped components

Compatible components

Compatible components: [https://github.com/opensearch-project/security-analytics.git, https://github.com/opensearch-project/custom-codecs.git, https://github.com/opensearch-project/security.git, https://github.com/opensearch-project/opensearch-oci-object-storage.git, https://github.com/opensearch-project/index-management.git, https://github.com/opensearch-project/geospatial.git, https://github.com/opensearch-project/job-scheduler.git, https://github.com/opensearch-project/sql.git, https://github.com/opensearch-project/notifications.git, https://github.com/opensearch-project/observability.git, https://github.com/opensearch-project/k-nn.git, https://github.com/opensearch-project/neural-search.git, https://github.com/opensearch-project/cross-cluster-replication.git, https://github.com/opensearch-project/alerting.git, https://github.com/opensearch-project/anomaly-detection.git, https://github.com/opensearch-project/performance-analyzer.git, https://github.com/opensearch-project/performance-analyzer-rca.git, https://github.com/opensearch-project/common-utils.git, https://github.com/opensearch-project/reporting.git]

[github-actions]

Gradle Check (Jenkins) Run Completed with:

• RESULT: FAILURE ❌
• URL: https://build.ci.opensearch.org/job/gradle-check/28231/
• CommitID: e8eb3a6
  Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green.
  Is the failure a flaky test unrelated to your change?

[vikasvb90]

@Bukhtawar
We don't need future completion pool resizing. I have reverted that. For stream reader pools, it is a decision between performance and refresh lag. For so workload, active thread count with 1 host generating the load is as shown in the graph

Stream read pool graph

Priority stream read pool graph

This will naturally impact CPU and run with scaling threadpool was seen consuming more CPU as well but it had lesser refresh lag.

[github-actions]

Gradle Check (Jenkins) Run Completed with:

• RESULT: UNSTABLE ❕
• TEST FAILURES:

      1 org.opensearch.repositories.azure.AzureBlobStoreRepositoryTests.testRequestStats
      1 org.opensearch.remotestore.SegmentReplicationUsingRemoteStoreIT.testDeleteOperations

• URL: https://build.ci.opensearch.org/job/gradle-check/28228/
• CommitID: 5d09eb8
  Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.

[codecov]

Codecov Report

Merging #10690 (9bc21d7) into main (9cc0e7d) will increase coverage by 0.05%.
Report is 2 commits behind head on main.
The diff coverage is 33.33%.

@@             Coverage Diff              @@
##               main   #10690      +/-   ##
============================================
+ Coverage     71.23%   71.28%   +0.05%     
+ Complexity    58571    58549      -22     
============================================
  Files          4853     4853              
  Lines        275796   275797       +1     
  Branches      40139    40139              
============================================
+ Hits         196458   196605     +147     
+ Misses        62925    62715     -210     
- Partials      16413    16477      +64     

Files	Coverage Δ
...earch/repositories/s3/async/AsyncPartsHandler.java	90.38% <100.00%> (ø)
...ch/repositories/s3/async/AsyncTransferManager.java	83.44% <100.00%> (ø)
...opensearch/repositories/s3/S3RepositoryPlugin.java	0.00% <0.00%> (ø)

... and 479 files with indirect coverage changes

[github-actions]

Gradle Check (Jenkins) Run Completed with:

• RESULT: UNSTABLE ❕
• TEST FAILURES:

      1 org.opensearch.remotestore.SegmentReplicationUsingRemoteStoreIT.testPrimaryStopped_ReplicaPromoted

• URL: https://build.ci.opensearch.org/job/gradle-check/28233/
• CommitID: 9bc21d7
  Please review all flaky tests that succeeded after retry and create an issue if one does not already exist to track the flaky failure.

[reta]

@vikasvb90 curious who is closing the BufferedInputStream? Does not look like AsyncRequestBody will do so

[vikasvb90]

Thanks @reta for pointing this!
We are wrapping the input stream with BufferedInputStream. Caller of async S3 api should close the provided input stream which means that we won't have any IO resource left open and all subsequent IO ops on the inner stream should result in stream closed errors. And instance of BufferedInputStream should get gc collected.

In remote store cases, we are closing the streams here

[reta]

Thanks @vikasvb90 , I understand regarding the wrapped input stream, but not BufferedInputStream - this is clearly a leak of the resource since BufferedInputStream allocated buffers that should be cleaned up as well. Please add hook to close the BufferedInputStream instance (or make inputStreamContainer to take care of it).

[vikasvb90]

If reference to BufferedInputStream is not present then internal buffer should also be eligible for gc. I can add hook for additional cleanup but doesn't look like a leak. I also don't see any rise in JVM in the benchmark if this was the case.

[reta]

This is resource management discipline: once created, the InputStream should be closed.