This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 288f977711 [rest] Do not replay a POST the server cannot absorb twice
(#9122)
288f977711 is described below
commit 288f977711af972371a409d17c9e8b45fb97bdb4
Author: Dapeng Sun(孙大鹏) <[email protected]>
AuthorDate: Mon Aug 10 13:51:11 2026 +0800
[rest] Do not replay a POST the server cannot absorb twice (#9122)
---
.../rest/ExponentialHttpRequestRetryStrategy.java | 29 ++++
.../java/org/apache/paimon/rest/HttpClient.java | 25 +++-
.../java/org/apache/paimon/rest/RESTRequest.java | 24 +++-
.../paimon/rest/HttpClientRetrySafetyTest.java | 146 +++++++++++++++++++++
.../TestExponentialHttpRequestRetryStrategy.java | 18 +++
5 files changed, 237 insertions(+), 5 deletions(-)
diff --git
a/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java
b/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java
index e8e621a64d..08e7215c11 100644
---
a/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java
+++
b/paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java
@@ -23,6 +23,7 @@ import org.apache.paimon.utils.Preconditions;
import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableSet;
import org.apache.hc.client5.http.HttpRequestRetryStrategy;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
import org.apache.hc.client5.http.utils.DateUtils;
import org.apache.hc.core5.concurrent.CancellableDependency;
import org.apache.hc.core5.http.ConnectionClosedException;
@@ -35,6 +36,7 @@ import org.apache.hc.core5.http.Method;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.util.TimeValue;
+import javax.annotation.Nullable;
import javax.net.ssl.SSLException;
import java.io.IOException;
@@ -47,6 +49,16 @@ import java.util.Set;
import java.util.concurrent.ThreadLocalRandom;
class ExponentialHttpRequestRetryStrategy implements HttpRequestRetryStrategy {
+
+ /**
+ * Context attribute marking one exchange as "must not be sent twice". A
429 or a 503 can reach
+ * the client from a proxy after the server already applied the request,
so replaying it applies
+ * it again; for a request that is not idempotent by content that is a
silent double apply. The
+ * mark travels in the context rather than in the request, so it never
reaches the wire and
+ * survives whatever the exec chain does to the request object.
+ */
+ static final String RETRY_UNSAFE_ATTRIBUTE = "paimon.rest.retry-unsafe";
+
private final int maxRetries;
private final Set<Class<? extends IOException>> nonRetriableExceptions;
private final Set<Integer> retriableCodes;
@@ -98,9 +110,26 @@ class ExponentialHttpRequestRetryStrategy implements
HttpRequestRetryStrategy {
@Override
public boolean retryRequest(HttpResponse response, int execCount,
HttpContext context) {
+ if (isRetryUnsafe(context)) {
+ // The status says nothing about whether the server applied the
request: a 503 from an
+ // intermediary can follow a request that already took effect.
Replaying it would apply
+ // it twice with nobody the wiser, so the failure goes back to the
caller instead.
+ return false;
+ }
return execCount <= maxRetries &&
retriableCodes.contains(response.getCode());
}
+ /** A context for one exchange that must be sent exactly once. */
+ static HttpClientContext retryUnsafeContext() {
+ HttpClientContext context = HttpClientContext.create();
+ context.setAttribute(RETRY_UNSAFE_ATTRIBUTE, Boolean.TRUE);
+ return context;
+ }
+
+ static boolean isRetryUnsafe(@Nullable HttpContext context) {
+ return context != null &&
Boolean.TRUE.equals(context.getAttribute(RETRY_UNSAFE_ATTRIBUTE));
+ }
+
@Override
public TimeValue getRetryInterval(HttpResponse response, int execCount,
HttpContext context) {
// a server may send a 429 / 503 with a Retry-After header
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java
b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java
index 311a1a2d4a..8205dfe212 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java
@@ -36,8 +36,12 @@ import
org.apache.hc.client5.http.classic.methods.HttpUriRequestBase;
import org.apache.hc.core5.http.ClassicHttpResponse;
import org.apache.hc.core5.http.ContentType;
import org.apache.hc.core5.http.Header;
+import org.apache.hc.core5.http.io.HttpClientResponseHandler;
import org.apache.hc.core5.http.io.entity.StringEntity;
import org.apache.hc.core5.http.message.BasicHeader;
+import org.apache.hc.core5.http.protocol.HttpContext;
+
+import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Arrays;
@@ -100,7 +104,13 @@ public class HttpClient implements RESTClient {
}
Header[] authHeaders = getHeaders(path, "POST", encodedBody,
restAuthFunction);
httpPost.setHeaders(authHeaders);
- return exec(httpPost, responseType);
+ // A POST the server cannot absorb twice is sent exactly once,
whatever the status says.
+ return exec(
+ httpPost,
+ responseType,
+ body != null && !body.isRetrySafe()
+ ?
ExponentialHttpRequestRetryStrategy.retryUnsafeContext()
+ : null);
}
@Override
@@ -127,9 +137,13 @@ public class HttpClient implements RESTClient {
}
private <T extends RESTResponse> T exec(HttpUriRequestBase request,
Class<T> responseType) {
+ return exec(request, responseType, null);
+ }
+
+ private <T extends RESTResponse> T exec(
+ HttpUriRequestBase request, Class<T> responseType, @Nullable
HttpContext context) {
try {
- return DEFAULT_HTTP_CLIENT.execute(
- request,
+ HttpClientResponseHandler<T> handler =
response -> {
String responseBodyStr =
RESTUtil.extractResponseBodyAsString(response);
if (!RESTUtil.isSuccessful(response)) {
@@ -159,7 +173,10 @@ public class HttpClient implements RESTClient {
} else {
throw new RESTException("response body is null.");
}
- });
+ };
+ return context == null
+ ? DEFAULT_HTTP_CLIENT.execute(request, handler)
+ : DEFAULT_HTTP_CLIENT.execute(request, context, handler);
} catch (IOException e) {
// No cause: a redirect/protocol error message can echo the target
URL (a signed URL).
throw new RESTException(
diff --git a/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java
b/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java
index 9c6758df14..35412bcb22 100644
--- a/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java
+++ b/paimon-api/src/main/java/org/apache/paimon/rest/RESTRequest.java
@@ -18,5 +18,27 @@
package org.apache.paimon.rest;
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
+
/** Interface to mark a REST request. */
-public interface RESTRequest extends RESTMessage {}
+public interface RESTRequest extends RESTMessage {
+
+ /**
+ * Whether sending this request a second time leaves the server where
sending it once does.
+ *
+ * <p>This is how the client treats the request, not something the server
is told: it is a
+ * getter on a serialized type and must stay off the wire.
+ *
+ * <p>POST is not idempotent by method, but nearly every request Paimon
sends over it is by
+ * content — registering a partition, creating a database, committing a
snapshot the server
+ * already holds — so the client retries them after a 429 or a 503, which
is the only defence
+ * against a rate limiter or a restarting node. A request that reports an
increment is the
+ * exception: a proxy answering 503 after the server already applied it
turns an automatic retry
+ * into a double count that no caller can see. Such a request says so here
and is sent exactly
+ * once; the failure reaches the caller, which can decide.
+ */
+ @JsonIgnore
+ default boolean isRetrySafe() {
+ return true;
+ }
+}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java
b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java
new file mode 100644
index 0000000000..8aca4d9947
--- /dev/null
+++
b/paimon-api/src/test/java/org/apache/paimon/rest/HttpClientRetrySafetyTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.paimon.rest;
+
+import org.apache.paimon.rest.exceptions.ServiceUnavailableException;
+import org.apache.paimon.rest.responses.ListDatabasesResponse;
+
+import
org.apache.paimon.shade.jackson2.com.fasterxml.jackson.annotation.JsonGetter;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests that a POST declaring itself unsafe to replay is sent exactly once,
and that every other
+ * POST keeps the 429/503 retry it has always had.
+ *
+ * <p>The server here refuses only the first attempt, so a retried request
succeeds on its second
+ * one: the request count separates "sent once" from "sent again" without
waiting out five backoffs.
+ */
+public class HttpClientRetrySafetyTest {
+
+ private static final String PATH = "/databases";
+
+ private HttpServer server;
+ private HttpClient client;
+ private final AtomicInteger requests = new AtomicInteger();
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ server = HttpServer.create(new InetSocketAddress(0), 0);
+ server.createContext(
+ PATH,
+ exchange -> {
+ if (requests.incrementAndGet() == 1) {
+ // A proxy answering 503 says nothing about whether
the server applied the
+ // request; this is exactly the shape that applies a
request twice.
+ respond(exchange, 503,
"{\"message\":\"busy\",\"code\":503}");
+ } else {
+ respond(exchange, 200, "{\"databases\":[\"db\"]}");
+ }
+ });
+ server.start();
+ client = new HttpClient("http://127.0.0.1:"; +
server.getAddress().getPort());
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (server != null) {
+ server.stop(0);
+ }
+ }
+
+ @Test
+ public void testARequestThatDeclaresItselfUnsafeIsNotRetried() {
+ assertThatThrownBy(() -> post(new UnsafeToRetry()))
+ .isInstanceOf(ServiceUnavailableException.class);
+
+ // Retrying would apply the same request a second time, and nothing
downstream could see it.
+ assertThat(requests.get()).isEqualTo(1);
+ }
+
+ @Test
+ public void testARequestThatNeverHeardOfRetrySafetyKeepsItsRetry() {
+ // The regression this guards against is the global one: every request
type implements
+ // RESTRequest and rides the interface default, so the case above
would still pass if the
+ // default flipped to false and silently took 429/503 retry away from
commits, database
+ // creation and every other POST in the catalog.
+ assertThat(new DefaultRetrySafety().isRetrySafe()).isTrue();
+ assertThat(post(new DefaultRetrySafety())).isNotNull();
+
+ assertThat(requests.get()).isEqualTo(2);
+ }
+
+ @Test
+ public void testRetrySafetyNeverReachesTheWire() {
+ // isRetrySafe is how the client treats the request, not something the
server is told. It is
+ // a getter on a serialized type, so without @JsonIgnore it would show
up in the body.
+ assertThat(RESTUtil.encodedBody(new
UnsafeToRetry())).doesNotContain("retrySafe");
+ assertThat(RESTUtil.encodedBody(new
DefaultRetrySafety())).doesNotContain("retrySafe");
+ }
+
+ /** A request that leaves {@link RESTRequest#isRetrySafe()} at its
default, as all others do. */
+ private static class DefaultRetrySafety implements RESTRequest {
+
+ @JsonGetter("name")
+ public String getName() {
+ return "db";
+ }
+ }
+
+ /** A request that must reach the server at most once. */
+ private static class UnsafeToRetry implements RESTRequest {
+
+ @JsonGetter("name")
+ public String getName() {
+ return "db";
+ }
+
+ @Override
+ public boolean isRetrySafe() {
+ return false;
+ }
+ }
+
+ private ListDatabasesResponse post(RESTRequest request) {
+ return client.post(PATH, request, ListDatabasesResponse.class, null);
+ }
+
+ private static void respond(HttpExchange exchange, int statusCode, String
body)
+ throws IOException {
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().add("Content-Type", "application/json");
+ exchange.sendResponseHeaders(statusCode, bytes.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(bytes);
+ }
+ }
+}
diff --git
a/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java
b/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java
index 9a743a93d9..cc4863f4e8 100644
---
a/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java
+++
b/paimon-api/src/test/java/org/apache/paimon/rest/TestExponentialHttpRequestRetryStrategy.java
@@ -20,6 +20,7 @@ package org.apache.paimon.rest;
import org.apache.hc.client5.http.HttpRequestRetryStrategy;
import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
import org.apache.hc.client5.http.utils.DateUtils;
import org.apache.hc.core5.http.ConnectionClosedException;
import org.apache.hc.core5.http.HttpHeaders;
@@ -220,4 +221,21 @@ public class TestExponentialHttpRequestRetryStrategy {
BasicHttpResponse response = new BasicHttpResponse(statusCode,
String.valueOf(statusCode));
assertThat(retryStrategy.retryRequest(response, 3, null)).isFalse();
}
+
+ @ParameterizedTest
+ @ValueSource(ints = {429, 503})
+ public void testRetryDoesNotHappenOnAnExchangeMarkedUnsafe(int statusCode)
{
+ BasicHttpResponse response = new BasicHttpResponse(statusCode,
String.valueOf(statusCode));
+
+ // The status says the request failed, not that the server never saw
it: a 503 can come
+ // from an intermediary after the request already took effect.
+ assertThat(
+ retryStrategy.retryRequest(
+ response,
+ 1,
+
ExponentialHttpRequestRetryStrategy.retryUnsafeContext()))
+ .isFalse();
+ // Only the marked exchange loses the retry; a plain one keeps it.
+ assertThat(retryStrategy.retryRequest(response, 1,
HttpClientContext.create())).isTrue();
+ }
}
