diff --git a/core/src/main/java/org/apache/iceberg/rest/ErrorHandlers.java b/core/src/main/java/org/apache/iceberg/rest/ErrorHandlers.java index 0fca2a5bc9dd..acc3439a6ca7 100644 --- a/core/src/main/java/org/apache/iceberg/rest/ErrorHandlers.java +++ b/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: diff --git a/core/src/main/java/org/apache/iceberg/rest/ExponentialHttpRequestRetryStrategy.java b/core/src/main/java/org/apache/iceberg/rest/ExponentialHttpRequestRetryStrategy.java index 263b3c305af0..bbda16e6aa4b 100644 --- a/core/src/main/java/org/apache/iceberg/rest/ExponentialHttpRequestRetryStrategy.java +++ b/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 diff --git a/core/src/main/java/org/apache/iceberg/rest/HTTPClient.java b/core/src/main/java/org/apache/iceberg/rest/HTTPClient.java index 532fff240141..bd301bc4f2ce 100644 --- a/core/src/main/java/org/apache/iceberg/rest/HTTPClient.java +++ b/core/src/main/java/org/apache/iceberg/rest/HTTPClient.java @@ -48,6 +48,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; @@ -183,7 +185,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 errorHandler) { + CloseableHttpResponse response, + String responseBody, + Consumer errorHandler, + Object wasRetried) { ErrorResponse errorResponse = null; if (responseBody != null) { @@ -220,10 +225,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 @@ -286,7 +300,8 @@ protected 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 respHeaders = Maps.newHashMap(); for (Header header : response.getHeaders()) { respHeaders.put(header.getName(), header.getValue()); @@ -304,7 +319,7 @@ protected 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) { diff --git a/core/src/main/java/org/apache/iceberg/rest/responses/ErrorResponse.java b/core/src/main/java/org/apache/iceberg/rest/responses/ErrorResponse.java index da93b821cedd..e9a7b0f27574 100644 --- a/core/src/main/java/org/apache/iceberg/rest/responses/ErrorResponse.java +++ b/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 stack; + private final boolean wasRetried; - private ErrorResponse(String message, String type, int code, List stack) { + private ErrorResponse( + String message, String type, int code, List 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 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 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); } } } diff --git a/core/src/test/java/org/apache/iceberg/rest/TestExponentialHttpRequestRetryStrategy.java b/core/src/test/java/org/apache/iceberg/rest/TestExponentialHttpRequestRetryStrategy.java index 7d8c58701a1e..00a4efa1f956 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestExponentialHttpRequestRetryStrategy.java +++ b/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(); } } diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java index 30b1ddef5b17..df199959c553 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTCatalog.java @@ -64,6 +64,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; @@ -94,6 +95,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; @@ -2650,6 +2652,24 @@ public void testTableExistsFallbackToGETRequest() { ConfigResponse.builder().withEndpoints(ImmutableList.of(Endpoint.V1_LOAD_TABLE)).build()); } + @Test + public void testErrorHandlingForConflicts() { + Consumer 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(