Core: Avoid table corruption from 409 on self conflicts after 5xx retries by throwing CommitStateUnknown (apache#12818)

Author: singhpk234

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

@@ -91,7 +91,19 @@ public void accept(ErrorResponse error) {
         case 404:
           throw new NoSuchTableException("%s", error.message());
         case 409:
-          throw new CommitFailedException("Commit failed: %s", error.message());
+          if (error.wasRetried()) {
+            // If a retried request finally fails with 409,
+            // the IRC service may have persisted the commit
+            // despite initial 5xx errors, resulting in a self-conflict on retry
+            // due to the base changing.
+            // Mark this failure as commit state unknown rather than failed to prevent file cleanup.
+            throw new CommitStateUnknownException(
+                new RESTException(
+                    "Commit status unknown, due to retries: %s: %s",
+                    error.code(), error.message()));
+          } else {
+            throw new CommitFailedException("Commit failed: %s", error.message());
+          }
         case 500:
         case 502:
         case 504:

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

@@ -118,12 +118,20 @@ public boolean retryRequest(
     }
 
     // Retry if the request is considered idempotent
-    return Method.isIdempotent(request.getMethod());
+    boolean shouldRetry = Method.isIdempotent(request.getMethod());
+    if (shouldRetry && context != null) {
+      context.setAttribute("was-retried", Boolean.TRUE);
+    }
+    return shouldRetry;
   }
 
   @Override
   public boolean retryRequest(HttpResponse response, int execCount, HttpContext context) {
-    return execCount <= maxRetries && retriableCodes.contains(response.getCode());
+    boolean shouldRetry = execCount <= maxRetries && retriableCodes.contains(response.getCode());
+    if (shouldRetry && context != null) {
+      context.setAttribute("was-retried", Boolean.TRUE);
+    }
+    return shouldRetry;
   }
 
   @Override

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

@@ -45,6 +45,8 @@
 import org.apache.hc.core5.http.impl.EnglishReasonPhraseCatalog;
 import org.apache.hc.core5.http.io.entity.EntityUtils;
 import org.apache.hc.core5.http.io.entity.StringEntity;
+import org.apache.hc.core5.http.protocol.BasicHttpContext;
+import org.apache.hc.core5.http.protocol.HttpContext;
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.iceberg.IcebergBuild;
 import org.apache.iceberg.exceptions.RESTException;
@@ -175,7 +177,10 @@ private static ErrorResponse buildDefaultErrorResponse(CloseableHttpResponse res
   // Process a failed response through the provided errorHandler, and throw a RESTException if the
   // provided error handler doesn't already throw.
   private static void throwFailure(
-      CloseableHttpResponse response, String responseBody, Consumer<ErrorResponse> errorHandler) {
+      CloseableHttpResponse response,
+      String responseBody,
+      Consumer<ErrorResponse> errorHandler,
+      Object wasRetried) {
     ErrorResponse errorResponse = null;
 
     if (responseBody != null) {
@@ -212,10 +217,19 @@ private static void throwFailure(
       errorResponse = buildDefaultErrorResponse(response);
     }
 
-    errorHandler.accept(errorResponse);
+    ErrorResponse enrichedErrorResponse =
+        ErrorResponse.builder()
+            .wasRetried(wasRetried == Boolean.TRUE)
+            .responseCode(errorResponse.code())
+            .withMessage(errorResponse.message())
+            .withType(errorResponse.type())
+            .withStackTrace(errorResponse.stack())
+            .build();
+
+    errorHandler.accept(enrichedErrorResponse);
 
     // Throw an exception in case the provided error handler does not throw.
-    throw new RESTException("Unhandled error: %s", errorResponse);
+    throw new RESTException("Unhandled error: %s", enrichedErrorResponse);
   }
 
   @Override
@@ -278,7 +292,8 @@ protected <T extends RESTResponse> T execute(
       request.setEntity(new StringEntity(encodedBody));
     }
 
-    try (CloseableHttpResponse response = httpClient.execute(request)) {
+    HttpContext context = new BasicHttpContext();
+    try (CloseableHttpResponse response = httpClient.execute(request, context)) {
       Map<String, String> respHeaders = Maps.newHashMap();
       for (Header header : response.getHeaders()) {
         respHeaders.put(header.getName(), header.getValue());
@@ -296,7 +311,7 @@ protected <T extends RESTResponse> T execute(
 
       if (!isSuccessful(response)) {
         // The provided error handler is expected to throw, but a RESTException is thrown if not.
-        throwFailure(response, responseBody, errorHandler);
+        throwFailure(response, responseBody, errorHandler, context.getAttribute("was-retried"));
       }
 
       if (responseBody == null) {

core/src/main/java/org/apache/iceberg/rest/responses/ErrorResponse.java

@@ -32,12 +32,15 @@ public class ErrorResponse implements RESTResponse {
   private final String type;
   private final int code;
   private final List<String> stack;
+  private final boolean wasRetried;
 
-  private ErrorResponse(String message, String type, int code, List<String> stack) {
+  private ErrorResponse(
+      String message, String type, int code, List<String> stack, boolean wasRetried) {
     this.message = message;
     this.type = type;
     this.code = code;
     this.stack = stack;
+    this.wasRetried = wasRetried;
     validate();
   }
 
@@ -62,6 +65,10 @@ public List<String> stack() {
     return stack;
   }
 
+  public boolean wasRetried() {
+    return wasRetried;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder();
@@ -92,6 +99,7 @@ public static class Builder {
     private String type;
     private Integer code;
     private List<String> stack;
+    private boolean wasRetried;
 
     private Builder() {}
 
@@ -126,9 +134,14 @@ public Builder responseCode(Integer responseCode) {
       return this;
     }
 
+    public Builder wasRetried(boolean hasBeenRetried) {
+      this.wasRetried = hasBeenRetried;
+      return this;
+    }
+
     public ErrorResponse build() {
       Preconditions.checkArgument(code != null, "Invalid response, missing field: code");
-      return new ErrorResponse(message, type, code, stack);
+      return new ErrorResponse(message, type, code, stack, wasRetried);
     }
   }
 }

core/src/test/java/org/apache/iceberg/rest/TestExponentialHttpRequestRetryStrategy.java

@@ -37,6 +37,8 @@
 import org.apache.hc.core5.http.HttpHeaders;
 import org.apache.hc.core5.http.HttpResponse;
 import org.apache.hc.core5.http.message.BasicHttpResponse;
+import org.apache.hc.core5.http.protocol.BasicHttpContext;
+import org.apache.hc.core5.http.protocol.HttpContext;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -80,14 +82,20 @@ public void exponentialRetry() {
 
   @Test
   public void basicRetry() {
+    HttpContext context503 = new BasicHttpContext();
     BasicHttpResponse response503 = new BasicHttpResponse(503, "Oopsie");
-    assertThat(retryStrategy.retryRequest(response503, 3, null)).isTrue();
+    assertThat(retryStrategy.retryRequest(response503, 3, context503)).isTrue();
+    assertThat(context503.getAttribute("was-retried") == Boolean.TRUE).isTrue();
 
+    HttpContext context429 = new BasicHttpContext();
     BasicHttpResponse response429 = new BasicHttpResponse(429, "Oopsie");
-    assertThat(retryStrategy.retryRequest(response429, 3, null)).isTrue();
+    assertThat(retryStrategy.retryRequest(response429, 3, context429)).isTrue();
+    assertThat(context429.getAttribute("was-retried") == Boolean.TRUE).isTrue();
 
+    HttpContext context404 = new BasicHttpContext();
     BasicHttpResponse response404 = new BasicHttpResponse(404, "Oopsie");
-    assertThat(retryStrategy.retryRequest(response404, 3, null)).isFalse();
+    assertThat(retryStrategy.retryRequest(response404, 3, context404)).isFalse();
+    assertThat(context429.getAttribute("was-retried") == Boolean.TRUE).isTrue();
   }
 
   @Test
@@ -155,8 +163,9 @@ public void noRetryOnAbortedRequests() {
   @Test
   public void retryOnNonAbortedRequests() {
     HttpGet request = new HttpGet("/");
+    HttpContext context = new BasicHttpContext();
 
-    assertThat(retryStrategy.retryRequest(request, new IOException(), 1, null)).isTrue();
+    assertThat(retryStrategy.retryRequest(request, new IOException(), 1, context)).isTrue();
   }
 
   @Test
@@ -199,13 +208,15 @@ public void invalidRetryAfterHeader() {
 
   @Test
   public void testRetryBadGateway() {
+    HttpContext context = new BasicHttpContext();
     BasicHttpResponse response502 = new BasicHttpResponse(502, "Bad gateway failure");
-    assertThat(retryStrategy.retryRequest(response502, 3, null)).isTrue();
+    assertThat(retryStrategy.retryRequest(response502, 3, context)).isTrue();
   }
 
   @Test
   public void testRetryGatewayTimeout() {
+    HttpContext context = new BasicHttpContext();
     BasicHttpResponse response504 = new BasicHttpResponse(504, "Gateway timeout");
-    assertThat(retryStrategy.retryRequest(response504, 3, null)).isTrue();
+    assertThat(retryStrategy.retryRequest(response504, 3, context)).isTrue();
   }
 }

core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java

@@ -62,6 +62,7 @@
 import org.apache.iceberg.catalog.TableCommit;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.CommitStateUnknownException;
 import org.apache.iceberg.exceptions.NotAuthorizedException;
 import org.apache.iceberg.exceptions.NotFoundException;
 import org.apache.iceberg.exceptions.ServiceFailureException;
@@ -92,6 +93,7 @@
 import org.eclipse.jetty.servlet.ServletContextHandler;
 import org.eclipse.jetty.servlet.ServletHolder;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
@@ -2648,6 +2650,24 @@ public void testTableExistsFallbackToGETRequest() {
         ConfigResponse.builder().withEndpoints(ImmutableList.of(Endpoint.V1_LOAD_TABLE)).build());
   }
 
+  @Test
+  public void testErrorHandlingForConflicts() {
+    Consumer<ErrorResponse> errorResponseConsumer = ErrorHandlers.tableCommitHandler();
+    // server returning 409 with client without retrying
+    ErrorResponse errorResponse409WithoutRetries =
+        ErrorResponse.builder().responseCode(409).wasRetried(false).build();
+    Assertions.assertThrows(
+        CommitFailedException.class,
+        () -> errorResponseConsumer.accept(errorResponse409WithoutRetries));
+
+    // server returning 409, with retries.
+    ErrorResponse errorResponse409WithRetries =
+        ErrorResponse.builder().responseCode(409).wasRetried(true).build();
+    Assertions.assertThrows(
+        CommitStateUnknownException.class,
+        () -> errorResponseConsumer.accept(errorResponse409WithRetries));
+  }
+
   private void verifyTableExistsFallbackToGETRequest(ConfigResponse configResponse) {
     RESTCatalogAdapter adapter =
         Mockito.spy(