[rest] Do not replay a POST the server cannot absorb twice

paimon-api/src/main/java/org/apache/paimon/partition/PartitionStatistics.java

@@ -30,15 +30,39 @@
 import java.util.Objects;
 
 /**
- * Statistics of a partition, fields inside may be negative, indicating that some data has been
- * removed.
+ * Statistics of a partition.
+ *
+ * <p>The numeric fields are read on two planes, and a negative value means a different thing on
+ * each. Which plane an instance belongs to follows from where it came from, never from the value:
+ *
+ * <ul>
+ * <li><b>Delta plane</b> — what a commit changed. A negative value is a decrement, and the server
+ * adds it to what it already holds. This is what a table snapshot commit reports.
+ * <li><b>Observation plane</b> — what a partition currently holds, as returned by {@code
+ * listPartitions}. A negative value ({@link #UNKNOWN}) means nobody ever reported that field,
+ * and {@code 0} means an exact zero. The two are not interchangeable: a consumer that treats
+ * unknown as zero plans against an empty partition that may hold a billion rows.
+ * </ul>
+ *
+ * <p>Unknown is per field, not per partition: a reporter that only knows the file count leaves the
+ * record count {@link #UNKNOWN} and fills the rest. Use {@link #isKnown(long)} rather than
+ * comparing against {@code -1}; any negative value on the observation plane is unknown.
  */
 @JsonIgnoreProperties(ignoreUnknown = true)
 @Public
 public class PartitionStatistics implements Serializable {
 
  private static final long serialVersionUID = 1L;
 
+ /**
+ * Canonical encoding of "this field was never reported" on the observation plane. Any negative
+ * value carries the same meaning; this is the one to write.
+ */
+ public static final long UNKNOWN = -1L;
+
+ /** Format tables have no buckets, so their bucket count is always unknown. */
+ public static final int UNKNOWN_TOTAL_BUCKETS = -1;
+
  public static final String FIELD_SPEC = "spec";
  public static final String FIELD_RECORD_COUNT = "recordCount";
  public static final String FIELD_FILE_SIZE_IN_BYTES = "fileSizeInBytes";
@@ -82,6 +106,20 @@ public PartitionStatistics(
  this.totalBuckets = totalBuckets;
  }
 
+ /** Statistics of a partition nobody ever reported on: every field {@link #UNKNOWN}. */
+ public static PartitionStatistics unknown(Map<String, String> spec) {
+ return new PartitionStatistics(
+ spec, UNKNOWN, UNKNOWN, UNKNOWN, UNKNOWN, UNKNOWN_TOTAL_BUCKETS);
+ }
+
+ /**
+ * Whether an observation-plane field carries a real measurement. Never apply this to a
+ * delta-plane value, where a negative number is a decrement rather than a missing measurement.
+ */
+ public static boolean isKnown(long value) {
+ return value >= 0;
+ }
+
  @JsonGetter(FIELD_SPEC)
  public Map<String, String> spec() {
  return spec;

paimon-api/src/main/java/org/apache/paimon/rest/ExponentialHttpRequestRetryStrategy.java

@@ -23,6 +23,7 @@
 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.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.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 @@ public boolean retryRequest(
 
  @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

paimon-api/src/main/java/org/apache/paimon/rest/HttpClient.java

@@ -36,8 +36,12 @@
 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 <T extends RESTResponse> T post(
  }
  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 @@ void setErrorHandler(ErrorHandler errorHandler) {
  }
 
  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 @@ private <T extends RESTResponse> T exec(HttpUriRequestBase request, Class<T> res
  } 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(

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;
+ }
+}

paimon-api/src/test/java/org/apache/paimon/partition/PartitionStatisticsTest.java

@@ -22,6 +22,9 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Collections;
+import java.util.Map;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Test for {@link PartitionStatistics}. */
@@ -41,4 +44,33 @@ void testLegacyPartitionStatisticsDeserialization() {
  assertThat(stats.lastFileCreationTime()).isEqualTo(123456789L);
  assertThat(stats.totalBuckets()).isEqualTo(0);
  }
+
+ @Test
+ void testZeroIsAKnownMeasurement() {
+ // The boundary the whole observation-plane contract rests on: an empty partition was
+ // measured, and a consumer that reads its zero as "nobody looked" plans against the wrong
+ // table.
+ assertThat(PartitionStatistics.isKnown(0L)).isTrue();
+ assertThat(PartitionStatistics.isKnown(1L)).isTrue();
+ assertThat(PartitionStatistics.isKnown(Long.MAX_VALUE)).isTrue();
+
+ assertThat(PartitionStatistics.isKnown(PartitionStatistics.UNKNOWN)).isFalse();
+ // Unknown is any negative value, not only the canonical -1.
+ assertThat(PartitionStatistics.isKnown(-2L)).isFalse();
+ assertThat(PartitionStatistics.isKnown(Long.MIN_VALUE)).isFalse();
+ }
+
+ @Test
+ void testUnknownLeavesEveryFieldUnknown() {
+ Map<String, String> spec = Collections.singletonMap("pt", "1");
+
+ PartitionStatistics stats = PartitionStatistics.unknown(spec);
+
+ assertThat(stats.spec()).isEqualTo(spec);
+ assertThat(PartitionStatistics.isKnown(stats.recordCount())).isFalse();
+ assertThat(PartitionStatistics.isKnown(stats.fileSizeInBytes())).isFalse();
+ assertThat(PartitionStatistics.isKnown(stats.fileCount())).isFalse();
+ assertThat(PartitionStatistics.isKnown(stats.lastFileCreationTime())).isFalse();
+ assertThat(PartitionStatistics.isKnown(stats.totalBuckets())).isFalse();
+ }
 }

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);
+ }
+ }
+}