AWS: add retry logic to S3InputStream #4912

jackye1995 · 2022-05-31T06:43:31Z

@danielcweeks @rajarshisarkar @amogh-jahagirdar @xiaoxuandev @singhpk234

Add retry for the S3InputStream so that when we encounter network failures (mostly SSLException for server side connection reset and SocketTimoutException for client side connection reset), we can quickly retry the read without failing the entire operation.

jackye1995 · 2022-05-31T06:45:17Z

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3InputStream.java

  }

  @Test
-  public void testRead() throws Exception {
+  public void testReadWithNormalClient() throws Exception {


cannot really figure out a way to do parameterized test here due to the use of S3MockRule, parameterization always gives me initialization failure... I will continue to see if I can remove the redundant tests, meanwhile I will also update integration test to run against both normal client and a fuzzy client.

amogh-jahagirdar · 2022-05-31T23:17:57Z

aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java

+   * Number of times to retry S3 read operation.
+   */
+  public static final String S3_READ_RETRY_NUM_RETRIES = "s3.read.retry.num-retries";
+  public static final int S3_READ_RETRY_NUM_RETRIES_DEFAULT = 6;


Just want to confirm how were these default values selected. From Hadoop S3A configurations?

+1, would be really nice to know

I think they are diff S3A consts : Code Pointer

retry limit - 7

retry interval - 500ms

thanks for the pointer, updated

amogh-jahagirdar · 2022-06-01T00:14:49Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

-    positionStream();
+    AtomicInteger byteRef = new AtomicInteger(0);
+    try {
+      Tasks.foreach(0)


Any reason we don't pass in the stream in the foreach? I guess it's a private field so it's a bit awkward to pass it in as an argument but seems more readable than Tasks.foreach(0) imo

I think we cannot, because the input stream needs to be closed and re-opened, but if we put it here then the retry will always retry against the same stream that is already closed after the first failure.

amogh-jahagirdar · 2022-06-01T01:17:19Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

+    if (exception instanceof UncheckedIOException) {
+      if (exception.getCause() instanceof EOFException) {
+        return false;
+      }
+    }
+
+    if (exception instanceof AwsServiceException) {
+      switch (((AwsServiceException) exception).statusCode()) {
+        case HttpURLConnection.HTTP_FORBIDDEN:
+        case HttpURLConnection.HTTP_BAD_REQUEST:
+          return false;
+      }
+    }
+
+    if (exception instanceof S3Exception) {
+      switch (((S3Exception) exception).statusCode()) {
+        case HttpURLConnection.HTTP_NOT_FOUND:
+        case 416: // range not satisfied
+          return false;
+      }
+    }
+
+    return true;


Curious is this the same retry policy that Hadoop S3A has?

It's not. It's closer to the ones in Presto and Trino. Basically it retires almost all IO exceptions except for EOF, because they are most likely network issues. For AWS side exceptions, this logic seems sufficient to me if they are proven sufficient in Presto and Trino. I am not sure if we need to list every single possible exception class like S3A did.

singhpk234 · 2022-05-31T16:44:31Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

@@ -88,23 +97,68 @@ public void seek(long newPos) {

  @Override
  public int read() throws IOException {
-    Preconditions.checkState(!closed, "Cannot read: already closed");
-    positionStream();
+    AtomicInteger byteRef = new AtomicInteger(0);


[question] do we need AtomicInt here, I think the requirement for this arises, here since we are using it in lambda provided in Task below, we can also use final int byteRef[] = {0} , does it seems reasonable ? Considering it's single threaded access.

Yes good point, it's only used to hold the value, not really trying to prevent any multi-thread access.

singhpk234 · 2022-05-31T16:49:32Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

+  private static boolean shouldRetry(Exception exception) {
+    if (exception instanceof UncheckedIOException) {
+      if (exception.getCause() instanceof EOFException) {
+        return false;
+      }
+    }
+
+    if (exception instanceof AwsServiceException) {
+      switch (((AwsServiceException) exception).statusCode()) {
+        case HttpURLConnection.HTTP_FORBIDDEN:
+        case HttpURLConnection.HTTP_BAD_REQUEST:
+          return false;
+      }
+    }
+
+    if (exception instanceof S3Exception) {
+      switch (((S3Exception) exception).statusCode()) {
+        case HttpURLConnection.HTTP_NOT_FOUND:
+        case 416: // range not satisfied
+          return false;
+      }
+    }
+
+    return true;


Should we add this in a seperate util class ? Considering it can be extended to all s3 interactions

Also any pointers if we know this is the complete list considering API's we use to connect to S3, For ex : (sample list : S3A)

Let's discuss this in the thread above #4912 (comment)

singhpk234 · 2022-06-01T04:30:17Z

aws/src/main/java/org/apache/iceberg/aws/AwsProperties.java

+   * Number of times to retry S3 read operation.
+   */
+  public static final String S3_READ_RETRY_NUM_RETRIES = "s3.read.retry.num-retries";
+  public static final int S3_READ_RETRY_NUM_RETRIES_DEFAULT = 6;


+1, would be really nice to know

I think they are diff S3A consts : Code Pointer

retry limit - 7

retry interval - 500ms

singhpk234 · 2022-06-01T04:54:51Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

    closeStream();
-    stream = s3.getObject(requestBuilder.build(), ResponseTransformer.toInputStream());
+    stream = readRange(String.format("bytes=%s-", pos));
  }

  private void closeStream() throws IOException {


[minor] Now since we are catching the IOException in closeServerSideStream closeStream actually doesn't throw IOException we can remove it from signature of closeStream / openStream then

singhpk234 · 2022-06-01T14:57:03Z

[doubt] SDK V2 also has a possibility to define retry policy / condition by override ClientConfiguration while building the s3 client itself, why are we opting do it here via Tasks here ?

danielcweeks · 2022-06-02T18:01:24Z

I'd like to take a closer look at this. So please hold off on merging.

danielcweeks

Ok, I understand what you're trying to achieve here, but I think this is best handled by the S3Client via configuration for Retry Policy and not custom handling in Iceberg.

Introducing our own retries can ultimately conflict with other retries and can have some bad side-effects (like failures resulting in retries on top of retries).

Unless there's some reason we can't handle this via the SDK, then don't think we should introduce retries here. @jackye1995 did you look into the retry policy? It seems like we could use many of these configs and just apply them in the AwsClientFactory.

jackye1995 · 2022-06-05T19:02:16Z

Introducing our own retries can ultimately conflict with other retries and can have some bad side-effects (like failures resulting in retries on top of retries).

Yes you are right, I actually plan to add one more PR for configuring retry policy. The issue here is a bit different. I think there are 2 issues that cannot be handled at SDK level:

AWS SDK only handles exceptions up to the AWS client calls, in this case s3.getObject. Once you get the input stream, further exceptions are not handled during operations like read and seek, and that's typically when network issues like connection timeout or socket timeout might happen. I haven't found logic in the SDK that could close a stream, re-seek to the right position and retry a read operation, but maybe I missed some code places.
When closing the input stream, the closing should call abort instead of close to terminate the HTTP connection. This should be done both at retry and stream close time. See this javadoc for more details.

danielcweeks · 2022-06-15T23:04:18Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

    }
  }

-  public void setSkipSize(int skipSize) {
-    this.skipSize = skipSize;
+  private static boolean shouldRetry(Exception exception) {


I think we need to be more explicit about the cases where we do want to retry rather than just defaulting to retry. I think there were specific cases that were mentined as known issues (e.g. socket timeout). However, the problem we've had in other retry scenarios is that the retry cases are overly broad and retry when they really shouldn't.

I think the default here should return false and only return true for the known Exceptions.

reasonably confident its a lot more complicated than this, especially as SDK-level failures often create deep chains with the underlying cause at the bottom

danielcweeks · 2022-06-15T23:10:18Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

-    positionStream();
+    int[] byteRef = new int[1];
+    try {
+      Tasks.foreach(0)


This is a lot of duplicate code for the retry logic. Could we consolidate to a method that wraps just the logic we want to execute?

For example:

retry(() -> { Preconditions.checkState(!closed, "Cannot read: already closed"); positionStream(); byteRef[0] = stream.read(); })

(you many have to use consumer/function for the range versions, but seems like there might be an more concise way to write this)

danielcweeks · 2022-06-15T23:12:19Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

-  private void closeStream() throws IOException {
-    if (stream != null) {
-      stream.close();
+  private static void closeServerSideStream(InputStream streamToClose) {


I'd just put this in the close stream (not have a separate method for it).

danielcweeks · 2022-06-15T23:30:54Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

@@ -88,23 +96,69 @@ public void seek(long newPos) {

  @Override
  public int read() throws IOException {
-    Preconditions.checkState(!closed, "Cannot read: already closed");
-    positionStream();
+    int[] byteRef = new int[1];


This is less than ideal but Tasks is a little limited in this area because it's really based on threaded execution and we're reusing it here for retry. It might be worth exploring whether we can tweak Tasks to support this usecase:

Ideally we'd have something like:

int read = Tasks.single() .exponentialBackoff(...) ... run(<funciton>);

We could also consider bringing in Failsafe, which is more closely aligned with what we want for these cases. Though it is a new dependency (but zero deps and apache licensed).

javrasya · 2024-01-31T07:38:00Z

Why was this PR closed? We are experiencing exactly the same problem and S3FileIO is literally unusable for us at the moment. no matter what we did to tune the apache http client such as TcpKeepAlive or socket timeout, it still hits this often and it breaks the whole pipeline.

We created our own S3FileIO and all the classes till S3InputStream where a SocketException arises, got the inspiration from this PR to retry and our problem is solved. But it is annoying to maintain our own version of those classes.

Is it not possible to revive this PR?

puchengy · 2024-05-16T05:56:40Z

@jackye1995 Hi, I also encountered the same issue as described in #10340 can we continue the PR to address the issue here? ALSo I wonder have you seen similar AWS clients having same issue? How do they end up solving the issue? Thanks.

@xiaoxuandev Hi, you added this PR, and I wonder are you interested in pushing it forward #8221?

@danielcweeks @rdblue @amogh-jahagirdar @nastra I wonder if you can comment/ suggest the path forward?

puchengy · 2024-05-20T15:06:47Z

@jackye1995 @xiaoxuandev @danielcweeks @rdblue @amogh-jahagirdar @nastra appreciate your feedback on this, thank you.

danielcweeks · 2024-05-21T15:30:32Z

@puchengy I spoke with @amogh-jahagirdar the other day and he's taking a look at this. There have been a few different attempts to address this issue and we're trying to find a solution that fits cleanly into the implementation.

puchengy · 2024-05-21T16:29:42Z

thank you @danielcweeks and @amogh-jahagirdar !

puchengy · 2024-05-28T20:21:48Z

@amogh-jahagirdar Hello, i wonder is there any update on this? Our clients from time to time hit the issue and cause some interruptions to their workload. Thanks!

amogh-jahagirdar · 2024-05-28T20:23:49Z

Hey @puchengy I'm wokring on it! I am aiming to have a PR out sometime this week to address this.

puchengy · 2024-06-12T22:16:00Z

@amogh-jahagirdar Hi, I am checking in to see if there is any update? thank you!

puchengy · 2024-07-01T17:17:58Z

@amogh-jahagirdar Hi, a gentle ping, may I know if you are still working on this? thanks

SandeepSinghGahir · 2024-08-13T00:02:36Z

Hi,

Is there any update on this PR? We have been facing the same issue while reading data from iceberg tables in glue job blocking the entire pipeline. Would really appreciate if we can get update and a probable fix for this.

steveloughran

commented. It's a start

steveloughran · 2024-08-14T14:25:55Z

aws/src/test/java/org/apache/iceberg/aws/s3/TestS3InputStream.java

@@ -178,12 +192,16 @@ private byte[] randomData(int size) {
  }

  private void writeS3Data(S3URI uri, byte[] data) throws IOException {
-    s3.putObject(
+    s3Client.putObject(
        PutObjectRequest.builder()
            .bucket(uri.bucket())
            .key(uri.key())
            .contentLength((long) data.length)
            .build(),
        RequestBody.fromBytes(data));


FWIW, this triggers a full array copy of the data.

steveloughran · 2024-08-14T14:33:23Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

+              byteRef[0] =  stream.read();
+            } catch (IOException e) {
+              closeStream();
+              throw new UncheckedIOException(e);


on an unrecoverable network error, abortStream() is required to avoid the failed TLS connection being recycled. This adds a new challenge: identify an unrecoverable network error from the deep nested stacks which can come from the SDK, and distinguish from other failures

steveloughran · 2024-08-14T14:35:24Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

+          // Stated in the ResponseInputStream javadoc:
+          // If it is not desired to read remaining data from the stream,
+          // you can explicitly abort the connection via abort().
+          ((Abortable) streamToClose).abort();


abort() removes it from the HTTP pool.

this is good if you are closing with an unrecoverable network error

this is absolutely what you do not want to do on a normal read.

steveloughran · 2024-08-14T14:40:00Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3InputStream.java

    }
  }

-  public void setSkipSize(int skipSize) {
-    this.skipSize = skipSize;
+  private static boolean shouldRetry(Exception exception) {


reasonably confident its a lot more complicated than this, especially as SDK-level failures often create deep chains with the underlying cause at the bottom

jackye1995 requested a review from danielcweeks May 31, 2022 06:43

github-actions bot added the AWS label May 31, 2022

jackye1995 commented May 31, 2022

View reviewed changes

amogh-jahagirdar reviewed Jun 1, 2022

View reviewed changes

singhpk234 reviewed Jun 1, 2022

View reviewed changes

danielcweeks requested changes Jun 2, 2022

View reviewed changes

jackye1995 force-pushed the s3-error branch from ff3e77f to be15ffa Compare June 5, 2022 18:43

AWS: add retry logic to S3InputStream

963bd8c

jackye1995 force-pushed the s3-error branch from be15ffa to 963bd8c Compare June 5, 2022 19:26

jackye1995 requested a review from danielcweeks June 15, 2022 03:34

danielcweeks reviewed Jun 15, 2022

View reviewed changes

jackye1995 closed this Mar 10, 2023

amogh-jahagirdar mentioned this pull request Jan 19, 2024

Caused by: java.net.SocketException: Connection reset #9444

Closed

SandeepSinghGahir mentioned this pull request Aug 13, 2024

javax.net.ssl.SSLException: Connection reset on S3 w/ S3FileIO and Apache HTTP client #10340

Closed

steveloughran reviewed Aug 14, 2024

View reviewed changes

AWS: add retry logic to S3InputStream #4912

AWS: add retry logic to S3InputStream #4912

Conversation

jackye1995 commented May 31, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

singhpk234 May 31, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

singhpk234 commented Jun 1, 2022

danielcweeks commented Jun 2, 2022

danielcweeks left a comment

Choose a reason for hiding this comment

jackye1995 commented Jun 5, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

javrasya commented Jan 31, 2024 • edited Loading

puchengy commented May 16, 2024 • edited Loading

puchengy commented May 20, 2024

danielcweeks commented May 21, 2024

puchengy commented May 21, 2024

puchengy commented May 28, 2024

amogh-jahagirdar commented May 28, 2024

puchengy commented Jun 12, 2024

puchengy commented Jul 1, 2024

SandeepSinghGahir commented Aug 13, 2024

steveloughran left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

singhpk234 May 31, 2022 •

edited

Loading

jackye1995 commented Jun 5, 2022 •

edited

Loading

javrasya commented Jan 31, 2024 •

edited

Loading

puchengy commented May 16, 2024 •

edited

Loading