Improving Retries based on Iceberg Integration Feedback

common/build.gradle.kts

@@ -10,6 +10,7 @@ plugins {
 
 dependencies {
     implementation(libs.slf4j.api)
+    implementation(libs.failsafe)
 
     testImplementation(libs.junit.jupiter)
     testImplementation(libs.mockito.core)

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/OpenStreamInformation.java

@@ -21,6 +21,8 @@
 import software.amazon.s3.analyticsaccelerator.request.EncryptionSecrets;
 import software.amazon.s3.analyticsaccelerator.request.ObjectMetadata;
 import software.amazon.s3.analyticsaccelerator.request.StreamAuditContext;
+import software.amazon.s3.analyticsaccelerator.util.retry.DefaultRetryStrategyImpl;
+import software.amazon.s3.analyticsaccelerator.util.retry.RetryStrategy;
 
 /**
  * Open stream information, useful for allowing the stream opening application to pass down known
@@ -45,6 +47,8 @@ public class OpenStreamInformation {
   @Builder.Default private RequestCallback requestCallback = new DefaultRequestCallbackImpl();
   private final EncryptionSecrets encryptionSecrets;
 
+  @Builder.Default private final RetryStrategy retryStrategy = new DefaultRetryStrategyImpl();
+
   /** Default set of settings for {@link OpenStreamInformation} */
   public static final OpenStreamInformation DEFAULT = OpenStreamInformation.builder().build();
 }

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/DefaultRetryStrategyImpl.java

@@ -0,0 +1,158 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package software.amazon.s3.analyticsaccelerator.util.retry;
+
+import dev.failsafe.Failsafe;
+import dev.failsafe.FailsafeException;
+import dev.failsafe.FailsafeExecutor;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import software.amazon.s3.analyticsaccelerator.common.Preconditions;
+
+/**
+ * Retry strategy implementation for seekable input stream operations. Uses Failsafe library to
+ * execute operations with configurable retry policies.
+ *
+ * <p>This strategy will be additive to readTimeout and readRetryCount set on PhysicalIO
+ * configuration.
+ */
+public class DefaultRetryStrategyImpl implements RetryStrategy {
+  private final List<RetryPolicy> retryPolicies;
+  FailsafeExecutor<Object> failsafeExecutor;
+
+  /** Creates a retry strategy with no retry policies (no retries). */
+  public DefaultRetryStrategyImpl() {
+    this.retryPolicies = new ArrayList<>();
+    this.failsafeExecutor = Failsafe.none();
+  }
+
+  /**
+   * Creates a retry strategy with one or more retry policies.
+   *
+   * @param outerPolicy the primary retry policy (required)
+   * @param policies additional retry policies (optional)
+   */
+  @SuppressWarnings("varargs")
+  public DefaultRetryStrategyImpl(RetryPolicy outerPolicy, RetryPolicy... policies) {
+    Preconditions.checkNotNull(outerPolicy);
+    this.retryPolicies = new ArrayList<>();
+    this.retryPolicies.add(outerPolicy);
+    if (policies != null && policies.length > 0) {
+      this.retryPolicies.addAll(Arrays.asList(policies));
+    }
+    this.failsafeExecutor = Failsafe.with(getDelegates());
+  }
+
+  /**
+   * Creates a retry strategy with a list of retry policies.
+   *
+   * @param policies the list of retry policies to apply
+   */
+  public DefaultRetryStrategyImpl(List<RetryPolicy> policies) {
+    Preconditions.checkNotNull(policies);
+    this.retryPolicies = new ArrayList<>();
+    this.retryPolicies.addAll(policies);
+    this.failsafeExecutor = Failsafe.with(getDelegates());
+  }
+
+  /**
+   * Executes a runnable operation with retry logic.
+   *
+   * @param runnable the operation to execute
+   * @throws IOException if the operation fails after all retries
+   */
+  @Override
+  public void execute(IORunnable runnable) throws IOException {
+    try {
+      this.failsafeExecutor.run(runnable::apply);
+    } catch (Exception ex) {
+      throw handleExceptionAfterRetry(ex);
+    }
+  }
+
+  /**
+   * Executes a supplier operation with retry logic.
+   *
+   * @param <T> return type of the supplier
+   * @param supplier the operation that returns a byte array
+   * @return the result of the supplier operation
+   * @throws IOException if the operation fails after all retries
+   */
+  @Override
+  public <T> T get(IOSupplier<T> supplier) throws IOException {
+    try {
+      return this.failsafeExecutor.get(supplier::apply);
+    } catch (Exception ex) {
+      throw handleExceptionAfterRetry(ex);
+    }
+  }
+
+  @Override
+  public RetryStrategy amend(RetryPolicy policy) {
+    Preconditions.checkNotNull(policy);
+    this.failsafeExecutor = this.failsafeExecutor.compose(policy.getDelegate());
+    return this;
+  }
+
+  @Override
+  public RetryStrategy merge(RetryStrategy strategy) {
+    Preconditions.checkNotNull(strategy);
+    for (RetryPolicy policy : strategy.getRetryPolicies()) {
+      this.failsafeExecutor = this.failsafeExecutor.compose(policy.getDelegate());
+    }
+    return this;
+  }
+
+  @Override
+  public List<RetryPolicy> getRetryPolicies() {
+    return this.retryPolicies;
+  }
+
+  /**
+   * Converts retry policies to their Failsafe delegate policies.
+   *
+   * @return list of Failsafe policies
+   */
+  private List<dev.failsafe.Policy<Object>> getDelegates() {
+    return this.retryPolicies.stream().map(RetryPolicy::getDelegate).collect(Collectors.toList());
+  }
+
+  /**
+   * Handles exceptions after retry attempts are exhausted.
+   *
+   * @param e the exception that occurred
+   * @return an IOException to throw
+   */
+  private IOException handleExceptionAfterRetry(Exception e) {
+    IOException toThrow = new IOException("Failed to execute operation with retries", e);
+
+    if (e instanceof FailsafeException) {
+      Optional<Throwable> cause = Optional.ofNullable(e.getCause());
+      if (cause.isPresent()) {
+        if (cause.get() instanceof IOException) {
+          return (IOException) cause.get();
+        } else {
+          toThrow = new IOException("Failed to execute operation with retries", cause.get());
+        }
+      }
+    }
+    return toThrow;
+  }
+}

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IORunnable.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package software.amazon.s3.analyticsaccelerator.util.retry;
+
+import java.io.IOException;
+
+/** A functional interface that mimics {@link Runnable}, but allows IOException to be thrown. */
+public interface IORunnable {
+  /**
+   * Functional representation of the code that takes no parameters and returns no value. The code
+   * is allowed to throw any exception.
+   *
+   * @throws IOException on error condition.
+   */
+  void apply() throws IOException;
+}

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/IOSupplier.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package software.amazon.s3.analyticsaccelerator.util.retry;
+
+import java.io.IOException;
+
+/**
+ * A function that mimics {@link java.util.function.Supplier}, but allows IOException to be thrown
+ * and returns T.
+ */
+@FunctionalInterface
+public interface IOSupplier<T> {
+
+  /**
+   * Functional representation of the code that takes no parameters and returns a value of type
+   * {@link T}. The code is allowed to throw any exception.
+   *
+   * @return a value of type {@link T}.
+   * @throws IOException on error condition.
+   */
+  T apply() throws IOException;
+}

common/src/main/java/software/amazon/s3/analyticsaccelerator/util/retry/RetryPolicy.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package software.amazon.s3.analyticsaccelerator.util.retry;
+
+/**
+ * A retry policy interface that wraps the Failsafe retry policy for byte array operations. Provides
+ * factory methods to create retry policies with default or custom configurations.
+ */
+public interface RetryPolicy {
+
+  /**
+   * Creates a new retry policy builder.
+   *
+   * @return a new RetryPolicyBuilder instance
+   */
+  static RetryPolicyBuilder builder() {
+    return new RetryPolicyBuilder();
+  }
+
+  /**
+   * Creates a retry policy with default settings.
+   *
+   * @return a RetryPolicy with default configuration
+   */
+  static RetryPolicy ofDefaults() {
+    return RetryPolicy.builder().build();
+  }
+
+  /**
+   * Gets the underlying Failsafe retry policy delegate.
+   *
+   * @return the Failsafe RetryPolicy for byte arrays
+   */
+  dev.failsafe.RetryPolicy<Object> getDelegate();
+}