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 6ae57e6d2c0e..286cfd495644 100644 --- a/core/src/main/java/org/apache/iceberg/rest/ExponentialHttpRequestRetryStrategy.java +++ b/core/src/main/java/org/apache/iceberg/rest/ExponentialHttpRequestRetryStrategy.java @@ -23,6 +23,7 @@ import java.net.ConnectException; import java.net.NoRouteToHostException; import java.net.UnknownHostException; +import java.time.Duration; import java.time.Instant; import java.util.Set; import java.util.concurrent.ThreadLocalRandom; @@ -42,6 +43,8 @@ import org.apache.hc.core5.util.TimeValue; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Defines an exponential HTTP request retry strategy and provides the same characteristics as the @@ -82,15 +85,26 @@ * {@link #getRetryInterval(HttpResponse, int, HttpContext)} to achieve exponential backoff. */ class ExponentialHttpRequestRetryStrategy implements HttpRequestRetryStrategy { + private static final Logger LOG = + LoggerFactory.getLogger(ExponentialHttpRequestRetryStrategy.class); + + static final String FIRST_ATTEMPT_EPOCH = "first-attempt-epoch"; + private final int maxRetries; + private final Duration keyLifetime; private final Set> nonRetriableExceptions; private final Set retriableCodes; private final Set idempotentRetriableCodes; ExponentialHttpRequestRetryStrategy(int maximumRetries) { + this(maximumRetries, null); + } + + ExponentialHttpRequestRetryStrategy(int maximumRetries, Duration keyLifetime) { Preconditions.checkArgument( maximumRetries > 0, "Cannot set retries to %s, the value must be positive", maximumRetries); this.maxRetries = maximumRetries; + this.keyLifetime = keyLifetime; this.retriableCodes = ImmutableSet.of(HttpStatus.SC_TOO_MANY_REQUESTS); this.idempotentRetriableCodes = ImmutableSet.of( @@ -133,10 +147,11 @@ public boolean retryRequest( return false; } - // Retry if the request is idempotent, or carries an Idempotency-Key (server guarantees safe - // retry) - return Method.isIdempotent(request.getMethod()) - || request.containsHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER); + if (!isSafeToRetry(request)) { + return false; + } + + return !wouldExceedKeyLifetime(request, null, execCount, context); } @Override @@ -154,10 +169,17 @@ public boolean retryRequest(HttpResponse response, int execCount, HttpContext co // - It's in a predefined list of retriable codes. // - The request is idempotent, and the response code indicates a retry is safe. // - The response code is '503 Service Unavailable' and includes a 'Retry-After' header. - return execCount <= maxRetries - && (retriableCodes.contains(response.getCode()) - || shouldRetryIdempotent(request, response.getCode()) - || is503Retryable); + boolean shouldRetry = + execCount <= maxRetries + && (retriableCodes.contains(response.getCode()) + || (isSafeToRetry(request) && idempotentRetriableCodes.contains(response.getCode())) + || is503Retryable); + + if (!shouldRetry) { + return false; + } + + return !wouldExceedKeyLifetime(request, response, execCount, context); } @Override @@ -183,22 +205,47 @@ public TimeValue getRetryInterval(HttpResponse response, int execCount, HttpCont } } - int delayMillis = 1000 * (int) Math.min(Math.pow(2.0, (long) execCount - 1.0), 64.0); + long delayMillis = nextBackoffMillis(execCount); int jitter = ThreadLocalRandom.current().nextInt(Math.max(1, (int) (delayMillis * 0.1))); return TimeValue.ofMilliseconds(delayMillis + jitter); } - private boolean shouldRetryIdempotent(HttpRequest request, int responseCode) { - if (request == null) { + private static boolean isSafeToRetry(HttpRequest request) { + return request != null + && (Method.isIdempotent(request.getMethod()) + || request.containsHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER)); + } + + private static long nextBackoffMillis(int execCount) { + return 1000L * (long) Math.min(Math.pow(2.0, (long) execCount - 1.0), 64.0); + } + + private boolean wouldExceedKeyLifetime( + HttpRequest request, HttpResponse response, int execCount, HttpContext context) { + if (keyLifetime == null + || context == null + || request == null + || !request.containsHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER)) { + return false; + } + + Object attr = context.getAttribute(FIRST_ATTEMPT_EPOCH); + if (!(attr instanceof Long)) { + LOG.warn( + "Idempotency-Key lifetime not enforced: first-attempt time not recorded for {}", + request.getMethod()); return false; } - // A request is retry-safe if its HTTP method is idempotent or it carries an Idempotency-Key - // header (which lets the server replay a finalized result on retry). - boolean retrySafe = - Method.isIdempotent(request.getMethod()) - || request.containsHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER); - return retrySafe && idempotentRetriableCodes.contains(responseCode); + long firstAttemptMillis = (Long) attr; + long now = System.currentTimeMillis(); + long nextIntervalMillis = + response != null + ? getRetryInterval(response, execCount, context).toMilliseconds() + : nextBackoffMillis(execCount); + boolean nextAttemptExceedsLifetime = + (now - firstAttemptMillis) + nextIntervalMillis >= keyLifetime.toMillis(); + return nextAttemptExceedsLifetime; } } 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 e2f69c29b052..a2ad8005c426 100644 --- a/core/src/main/java/org/apache/iceberg/rest/HTTPClient.java +++ b/core/src/main/java/org/apache/iceberg/rest/HTTPClient.java @@ -24,6 +24,8 @@ import java.io.UncheckedIOException; import java.net.URI; import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.format.DateTimeParseException; import java.util.Map; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; @@ -81,6 +83,7 @@ public class HTTPClient extends BaseHTTPClient { static final String CLIENT_GIT_COMMIT_SHORT_HEADER = "X-Client-Git-Commit-Short"; private static final String REST_MAX_RETRIES = "rest.client.max-retries"; + static final String REST_IDEMPOTENCY_KEY_LIFETIME = "rest.client.idempotency-key-lifetime"; static final String REST_MAX_CONNECTIONS = "rest.client.max-connections"; static final int REST_MAX_CONNECTIONS_DEFAULT = 100; static final String REST_MAX_CONNECTIONS_PER_ROUTE = "rest.client.connections-per-route"; @@ -125,7 +128,9 @@ private HTTPClient( clientBuilder.setConnectionManager(connectionManager); int maxRetries = PropertyUtil.propertyAsInt(properties, REST_MAX_RETRIES, 5); - clientBuilder.setRetryStrategy(new ExponentialHttpRequestRetryStrategy(maxRetries)); + Duration keyLifetime = parseKeyLifetime(properties.get(REST_IDEMPOTENCY_KEY_LIFETIME)); + clientBuilder.setRetryStrategy( + new ExponentialHttpRequestRetryStrategy(maxRetries, keyLifetime)); String userAgent = PropertyUtil.propertyAsString(properties, REST_USER_AGENT, null); if (userAgent != null) { @@ -164,6 +169,18 @@ public HTTPClient withAuthSession(AuthSession session) { return new HTTPClient(this, session); } + private static Duration parseKeyLifetime(String lifetime) { + if (lifetime == null) { + return null; + } + try { + return Duration.parse(lifetime); + } catch (DateTimeParseException e) { + LOG.warn("Ignoring malformed idempotency-key lifetime: {}", lifetime, e); + return null; + } + } + private static String extractResponseBodyAsString(ClassicHttpResponse response) { try { if (response.getEntity() == null) { @@ -317,6 +334,9 @@ protected T execute( } HttpContext context = HttpClientContext.create(); + // Record first-send time so the retry strategy bounds keyed retries to the key lifetime. + context.setAttribute( + ExponentialHttpRequestRetryStrategy.FIRST_ATTEMPT_EPOCH, System.currentTimeMillis()); try { return httpClient.execute( request, diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java index e7b68dad4aae..4f242b4be622 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java @@ -214,13 +214,18 @@ public void initialize(String name, Map unresolved) { } // build the final configuration and set up the catalog's auth - Map mergedProps = config.merge(props); + ImmutableMap.Builder mergedPropsBuilder = + ImmutableMap.builder().putAll(config.merge(props)); // Enable Idempotency-Key header for mutation endpoints if the server advertises support if (config.idempotencyKeyLifetime() != null) { this.mutationHeaders = RESTUtil::idempotencyHeaders; + mergedPropsBuilder.put( + HTTPClient.REST_IDEMPOTENCY_KEY_LIFETIME, config.idempotencyKeyLifetime()); } + Map mergedProps = mergedPropsBuilder.buildKeepingLast(); + if (config.endpoints().isEmpty()) { this.endpoints = PropertyUtil.propertyAsBoolean( 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 7dcacc3a1216..3709b89f6cdd 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestExponentialHttpRequestRetryStrategy.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestExponentialHttpRequestRetryStrategy.java @@ -27,6 +27,7 @@ import java.net.NoRouteToHostException; import java.net.SocketTimeoutException; import java.net.UnknownHostException; +import java.time.Duration; import java.time.Instant; import java.time.temporal.ChronoUnit; import javax.net.ssl.SSLException; @@ -249,4 +250,141 @@ public void testRetryDoesNotHappenForNonIdempotentMethodWithoutIdempotencyKey(in context.setRequest(new BasicHttpRequest("POST", "/")); assertThat(retryStrategy.retryRequest(response, 3, context)).isFalse(); } + + @Test + public void testKeyedRetryAbandonedWhenLifetimeExceeded() { + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofSeconds(1)); + BasicHttpResponse response = + new BasicHttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR, "error"); + response.addHeader(new BasicHeader(HttpHeaders.RETRY_AFTER, "3600")); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-abandon"); + context.setRequest(request); + context.setAttribute( + ExponentialHttpRequestRetryStrategy.FIRST_ATTEMPT_EPOCH, System.currentTimeMillis()); + assertThat(strategy.retryRequest(response, 1, context)).isFalse(); + } + + @Test + public void testKeyedRetryContinuesWithinLifetime() { + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofHours(1)); + BasicHttpResponse response = + new BasicHttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR, "error"); + response.addHeader(new BasicHeader(HttpHeaders.RETRY_AFTER, "1")); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-continue"); + context.setRequest(request); + context.setAttribute( + ExponentialHttpRequestRetryStrategy.FIRST_ATTEMPT_EPOCH, System.currentTimeMillis()); + assertThat(strategy.retryRequest(response, 1, context)).isTrue(); + } + + @Test + public void testKeyedNetworkExceptionRetryAbandonedWhenLifetimeExceeded() { + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofMillis(1)); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-net-abandon"); + context.setRequest(request); + context.setAttribute( + ExponentialHttpRequestRetryStrategy.FIRST_ATTEMPT_EPOCH, System.currentTimeMillis()); + assertThat(strategy.retryRequest(request, new IOException("connection reset"), 1, context)) + .isFalse(); + } + + @Test + public void testKeyedNetworkExceptionRetryHonoredWithoutLifetime() { + HttpRequestRetryStrategy strategy = new ExponentialHttpRequestRetryStrategy(5); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-net-ok"); + context.setRequest(request); + assertThat(strategy.retryRequest(request, new IOException("connection reset"), 1, context)) + .isTrue(); + } + + @Test + public void testIdempotentGetNotConstrainedByKeyLifetime() { + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofMillis(1)); + BasicHttpResponse response = + new BasicHttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR, "error"); + response.addHeader(new BasicHeader(HttpHeaders.RETRY_AFTER, "3600")); + HttpClientContext context = HttpClientContext.create(); + context.setRequest(new BasicHttpRequest("GET", "/")); + assertThat(strategy.retryRequest(response, 1, context)).isTrue(); + } + + @Test + public void testKeyedRetryHonorsRecordedFirstAttemptTime() { + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofMinutes(30)); + BasicHttpResponse response = + new BasicHttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR, "error"); + response.addHeader(new BasicHeader(HttpHeaders.RETRY_AFTER, "1")); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-recorded"); + context.setRequest(request); + context.setAttribute( + ExponentialHttpRequestRetryStrategy.FIRST_ATTEMPT_EPOCH, + System.currentTimeMillis() - Duration.ofMinutes(31).toMillis()); + assertThat(strategy.retryRequest(response, 1, context)).isFalse(); + } + + @Test + public void testKeyedRetryAbandonedOnceCumulativeElapsedExceedsLifetime() + throws InterruptedException { + // 2s lifetime, 1s Retry-After: 1st retry allowed, 2nd abandoned once cumulative elapsed > 2s. + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofSeconds(2)); + BasicHttpResponse response = + new BasicHttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR, "error"); + response.addHeader(new BasicHeader(HttpHeaders.RETRY_AFTER, "1")); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-cumulative"); + context.setRequest(request); + context.setAttribute( + ExponentialHttpRequestRetryStrategy.FIRST_ATTEMPT_EPOCH, System.currentTimeMillis()); + + assertThat(strategy.retryRequest(response, 1, context)).isTrue(); + Thread.sleep(1100); + assertThat(strategy.retryRequest(response, 2, context)).isFalse(); + } + + @Test + public void testKeyedRetryProceedsWhenFirstAttemptTimeNotRecorded() { + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofSeconds(1)); + BasicHttpResponse response = + new BasicHttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR, "error"); + response.addHeader(new BasicHeader(HttpHeaders.RETRY_AFTER, "3600")); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-unrecorded"); + context.setRequest(request); + assertThat(strategy.retryRequest(response, 1, context)).isTrue(); + } + + @Test + public void testKeyedRetryAbandonedWhenExponentialIntervalExceedsLifetime() { + // No Retry-After -> interval falls back to exponential (>=1s), exceeding the 1ms lifetime. + HttpRequestRetryStrategy strategy = + new ExponentialHttpRequestRetryStrategy(5, Duration.ofMillis(1)); + BasicHttpResponse response = + new BasicHttpResponse(HttpStatus.SC_INTERNAL_SERVER_ERROR, "error"); + HttpClientContext context = HttpClientContext.create(); + BasicHttpRequest request = new BasicHttpRequest("POST", "/"); + request.addHeader(RESTUtil.IDEMPOTENCY_KEY_HEADER, "key-exp"); + context.setRequest(request); + context.setAttribute( + ExponentialHttpRequestRetryStrategy.FIRST_ATTEMPT_EPOCH, System.currentTimeMillis()); + assertThat(strategy.retryRequest(response, 1, context)).isFalse(); + } }