Core: Report commit retry exhaustion reason

core/src/main/java/org/apache/iceberg/BaseReplaceSortOrder.java

@@ -29,6 +29,7 @@
 
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.expressions.Term;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.Tasks;
 
 public class BaseReplaceSortOrder implements ReplaceSortOrder {
@@ -57,6 +58,7 @@ public void commit() {
             base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(
             taskOps -> {
               this.base = ops.refresh();

core/src/main/java/org/apache/iceberg/PropertiesUpdate.java

@@ -33,6 +33,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.Tasks;
 
 class PropertiesUpdate implements UpdateProperties {
@@ -107,6 +108,7 @@ public void commit() {
             base.propertyTryAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(
             taskOps -> {
               Map<String, String> newProperties = apply();

core/src/main/java/org/apache/iceberg/RemoveSnapshots.java

@@ -49,6 +49,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.relocated.com.google.common.util.concurrent.MoreExecutors;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.DateTimeUtil;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.SnapshotUtil;
@@ -366,6 +367,7 @@ public void commit() {
             base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(
             item -> {
               TableMetadata updated = internalApply();

core/src/main/java/org/apache/iceberg/SetLocation.java

@@ -28,6 +28,7 @@
 import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT;
 
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.Tasks;
 
 public class SetLocation implements UpdateLocation {
@@ -61,6 +62,7 @@ public void commit() {
             base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(taskOps -> taskOps.commit(base, base.updateLocation(newLocation)));
   }
 }

core/src/main/java/org/apache/iceberg/SetSnapshotOperation.java

@@ -31,6 +31,7 @@
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.SnapshotUtil;
 import org.apache.iceberg.util.Tasks;
 
@@ -115,6 +116,7 @@ public void commit() {
             base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(
             taskOps -> {
               Snapshot snapshot = apply();

core/src/main/java/org/apache/iceberg/SetStatistics.java

@@ -32,6 +32,7 @@
 import java.util.Optional;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.Tasks;
 
 public class SetStatistics implements UpdateStatistics {
@@ -71,6 +72,7 @@ public void commit() {
                 .propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(
             taskOps -> {
               TableMetadata base = taskOps.refresh();

core/src/main/java/org/apache/iceberg/SnapshotProducer.java

@@ -72,6 +72,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.relocated.com.google.common.math.IntMath;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.Exceptions;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.PropertyUtil;
@@ -494,6 +495,7 @@ public void commit() {
                 base.propertyAsInt(COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
                 2.0 /* exponential */)
             .onlyRetryOn(CommitFailedException.class)
+            .onRetryExhausted(CommitRetry::retryExhaustedException)
             .countAttempts(commitMetrics().attempts())
             .run(
                 taskOps -> {

core/src/main/java/org/apache/iceberg/rest/ErrorHandlers.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.rest;
 
 import java.util.function.Consumer;
+import org.apache.iceberg.RetryableValidationException;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.BadRequestException;
 import org.apache.iceberg.exceptions.CommitFailedException;
@@ -125,6 +126,13 @@ public void accept(ErrorResponse error) {
         case 404:
           throw new NoSuchTableException("%s", error.message());
         case 409:
+          if (RetryableValidationException.class.getSimpleName().equals(error.type())) {
+            throw new CommitFailedException(
+                new RetryableValidationException("%s", error.message()),
+                "Commit failed: %s",
+                error.message());
+          }
+
           throw new CommitFailedException("Commit failed: %s", error.message());
         case 500:
         case 502:

core/src/main/java/org/apache/iceberg/util/CommitRetry.java

@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.util;
+
+import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES;
+import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS;
+
+import org.apache.iceberg.RetryableValidationException;
+import org.apache.iceberg.exceptions.CommitFailedException;
+
+public class CommitRetry {
+  private CommitRetry() {}
+
+  public static CommitFailedException retryExhaustedException(
+      Exception cause, Tasks.RetryExhaustionReason reason) {
+    if (isRetryableValidationCommitFailure(cause)) {
+      return (CommitFailedException) cause;
+    }
+
+    return new CommitFailedException(
+        cause, "Commit failed after exhausting retries. %s", retryExhaustionMessage(reason));
+  }
+
+  public static boolean isRetryableValidationCommitFailure(Exception cause) {
+    return cause instanceof CommitFailedException
+        && cause.getCause() instanceof RetryableValidationException;
+  }
+
+  @SuppressWarnings("StatementSwitchToExpressionSwitch")
+  private static String retryExhaustionMessage(Tasks.RetryExhaustionReason reason) {
+    switch (reason) {
+      case ATTEMPT_LIMIT:
+        return String.format("To allow more retry attempts, adjust %s.", COMMIT_NUM_RETRIES);
+
+      case TIMEOUT:
+        return String.format(
+            "To allow more total retry time, adjust %s.", COMMIT_TOTAL_RETRY_TIME_MS);
+
+      case ATTEMPT_LIMIT_AND_TIMEOUT:
+        return String.format(
+            "To allow more retry attempts and total retry time, adjust %s and %s.",
+            COMMIT_NUM_RETRIES, COMMIT_TOTAL_RETRY_TIME_MS);
+    }
+
+    throw new IllegalArgumentException("Unknown retry exhaustion reason: " + reason);
+  }
+}

core/src/main/java/org/apache/iceberg/util/Tasks.java

@@ -32,6 +32,7 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.BiFunction;
 import java.util.function.Predicate;
 import java.util.stream.Stream;
 import org.apache.iceberg.metrics.Counter;
@@ -44,6 +45,12 @@ public class Tasks {
 
   private Tasks() {}
 
+  public enum RetryExhaustionReason {
+    ATTEMPT_LIMIT,
+    TIMEOUT,
+    ATTEMPT_LIMIT_AND_TIMEOUT
+  }
+
   public static class UnrecoverableException extends RuntimeException {
     public UnrecoverableException(String message) {
       super(message);
@@ -88,6 +95,8 @@ public static class Builder<I> {
     private long maxDurationMs = 600000; // 10 minutes
     private double scaleFactor = 2.0; // exponential
     private Counter attemptsCounter;
+    private BiFunction<Exception, RetryExhaustionReason, RuntimeException> retryExhaustedHandler =
+        null;
 
     public Builder(Iterable<I> items) {
       this.items = items;
@@ -180,6 +189,12 @@ public Builder<I> countAttempts(Counter counter) {
       return this;
     }
 
+    public Builder<I> onRetryExhausted(
+        BiFunction<Exception, RetryExhaustionReason, RuntimeException> handler) {
+      this.retryExhaustedHandler = handler;
+      return this;
+    }
+
     public Builder<I> exponentialBackoff(
         long backoffMinSleepTimeMs,
         long backoffMaxSleepTimeMs,
@@ -414,39 +429,24 @@ private <E extends Exception> void runTaskWithRetry(Task<I, E> task, I item) thr
           break;
 
         } catch (Exception e) {
-          long durationMs = System.currentTimeMillis() - start;
-          if (attempt >= maxAttempts || (durationMs > maxDurationMs && attempt > 1)) {
-            if (durationMs > maxDurationMs) {
-              LOG.info("Stopping retries after {} ms", durationMs);
-            }
+          if (!shouldRetry(e)) {
             throw e;
           }
 
-          if (shouldRetryPredicate != null) {
-            if (!shouldRetryPredicate.test(e)) {
-              throw e;
+          long durationMs = System.currentTimeMillis() - start;
+          boolean attemptsExhausted = attempt >= maxAttempts;
+          boolean timeoutExhausted = durationMs > maxDurationMs && attempt > 1;
+          if (attemptsExhausted || timeoutExhausted) {
+            if (timeoutExhausted) {
+              LOG.info("Stopping retries after {} ms", durationMs);
             }
 
-          } else if (onlyRetryExceptions != null) {
-            // if onlyRetryExceptions are present, then this retries if one is found
-            boolean matchedRetryException = false;
-            for (Class<? extends Exception> exClass : onlyRetryExceptions) {
-              if (exClass.isInstance(e)) {
-                matchedRetryException = true;
-                break;
-              }
-            }
-            if (!matchedRetryException) {
-              throw e;
+            if (retryExhaustedHandler != null) {
+              throw retryExhaustedHandler.apply(
+                  e, retryExhaustionReason(attemptsExhausted, timeoutExhausted));
             }
 
-          } else {
-            // otherwise, always retry unless one of the stop exceptions is found
-            for (Class<? extends Exception> exClass : stopRetryExceptions) {
-              if (exClass.isInstance(e)) {
-                throw e;
-              }
-            }
+            throw e;
           }
 
           int delayMs =
@@ -468,6 +468,41 @@ private <E extends Exception> void runTaskWithRetry(Task<I, E> task, I item) thr
         }
       }
     }
+
+    private boolean shouldRetry(Exception exception) {
+      if (shouldRetryPredicate != null) {
+        return shouldRetryPredicate.test(exception);
+      } else if (onlyRetryExceptions != null) {
+        // if onlyRetryExceptions are present, then this retries if one is found
+        for (Class<? extends Exception> exClass : onlyRetryExceptions) {
+          if (exClass.isInstance(exception)) {
+            return true;
+          }
+        }
+
+        return false;
+      } else {
+        // otherwise, always retry unless one of the stop exceptions is found
+        for (Class<? extends Exception> exClass : stopRetryExceptions) {
+          if (exClass.isInstance(exception)) {
+            return false;
+          }
+        }
+
+        return true;
+      }
+    }
+
+    private RetryExhaustionReason retryExhaustionReason(
+        boolean attemptsExhausted, boolean timeoutExhausted) {
+      if (attemptsExhausted && timeoutExhausted) {
+        return RetryExhaustionReason.ATTEMPT_LIMIT_AND_TIMEOUT;
+      } else if (attemptsExhausted) {
+        return RetryExhaustionReason.ATTEMPT_LIMIT;
+      } else {
+        return RetryExhaustionReason.TIMEOUT;
+      }
+    }
   }
 
   @SuppressWarnings("checkstyle:CyclomaticComplexity")

core/src/main/java/org/apache/iceberg/view/PropertiesUpdate.java

@@ -33,6 +33,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.Tasks;
 
@@ -73,6 +74,7 @@ public void commit() {
                 base.properties(), COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(taskOps -> taskOps.commit(base, internalApply()));
   }
 

core/src/main/java/org/apache/iceberg/view/SetViewLocation.java

@@ -30,6 +30,7 @@
 import org.apache.iceberg.UpdateLocation;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.Tasks;
 
@@ -63,6 +64,7 @@ public void commit() {
                 base.properties(), COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(
             taskOps ->
                 taskOps.commit(base, ViewMetadata.buildFrom(base).setLocation(apply()).build()));

core/src/main/java/org/apache/iceberg/view/ViewVersionReplace.java

@@ -34,6 +34,7 @@
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.util.CommitRetry;
 import org.apache.iceberg.util.PropertyUtil;
 import org.apache.iceberg.util.Tasks;
 
@@ -100,6 +101,7 @@ public void commit() {
                 base.properties(), COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
             2.0 /* exponential */)
         .onlyRetryOn(CommitFailedException.class)
+        .onRetryExhausted(CommitRetry::retryExhaustedException)
         .run(taskOps -> taskOps.commit(base, internalApply()));
   }