Skip to content

Core: Bound Idempotency-Key retries to the advertised key lifetime #18070

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
HotSushi wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Core: Bound Idempotency-Key retries to the advertised key lifetime #18070

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
81 changes: 64 additions & 17 deletions core/src/main/java/org/apache/iceberg/rest/ExponentialHttpRequestRetryStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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<Class<? extends IOException>> nonRetriableExceptions;
private final Set<Integer> retriableCodes;
private final Set<Integer> 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(
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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;
}
}
22 changes: 21 additions & 1 deletion core/src/main/java/org/apache/iceberg/rest/HTTPClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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";
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);

@huaxingao huaxingao Sep 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we also reject zero and negative values?

} 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) {
Expand Down Expand Up @@ -317,6 +334,9 @@ protected <T extends RESTResponse> 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,
Expand Down
7 changes: 6 additions & 1 deletion core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,13 +214,18 @@ public void initialize(String name, Map<String, String> unresolved) {
}

// build the final configuration and set up the catalog's auth
Map<String, String> mergedProps = config.merge(props);
ImmutableMap.Builder<String, String> mergedPropsBuilder =
ImmutableMap.<String, String>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<String, String> mergedProps = mergedPropsBuilder.buildKeepingLast();

if (config.endpoints().isEmpty()) {
this.endpoints =
PropertyUtil.propertyAsBoolean(
Expand Down
138 changes: 138 additions & 0 deletions core/src/test/java/org/apache/iceberg/rest/TestExponentialHttpRequestRetryStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();

@huaxingao huaxingao Sep 18, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this only tests that one Retry-After is longer than keyLifetime. context is new and retryRequest is called once, so wouldExceedKeyLifetime always takes the else branch: FIRST_ATTEMPT_EPOCH is never read back, firstAttemptMillis is set to now, and now - firstAttemptMillis is always 0. I deleted the context.setAttribute(FIRST_ATTEMPT_EPOCH, firstAttemptMillis) line locally and all five new tests still passed.

Can you call retryRequest twice on the same context with time passing in between? With a 2s keyLifetime and Retry-After: 1:

  assertThat(strategy.retryRequest(response, 1, context)).isTrue();
  Thread.sleep(1100);
  assertThat(strategy.retryRequest(response, 2, context)).isFalse();

@HotSushi HotSushi Sep 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

fixed the first-attempt time recording and also added a two-call test.

}

@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();
}
}
Loading