You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2026/09/14 05:24:27 UTC
(iceberg) branch main updated: Core: Make REST scan planning poll retries configurable (#17846)
This is an automated email from the ASF dual-hosted git repository.
nastra pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new ee1acca581 Core: Make REST scan planning poll retries configurable (#17846)
ee1acca581 is described below
commit ee1acca581d6a2f389dd561bf59ec951c19d9caf
Author: Bartosz Szostakiewicz <81...@users.noreply.github.com>
AuthorDate: Mon Sep 14 07:24:19 2026 +0200
Core: Make REST scan planning poll retries configurable (#17846)
---
.../apache/iceberg/rest/RESTCatalogProperties.java | 6 +
.../org/apache/iceberg/rest/RESTTableScan.java | 15 +-
.../apache/iceberg/rest/TestRESTScanPlanning.java | 219 +++++++++++++++------
docs/docs/rest-catalog.md | 2 +
docs/docs/rest-protocol.md | 4 +-
5 files changed, 182 insertions(+), 64 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java b/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java
index 4180534bd8..ec920883dd 100644
--- a/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java
+++ b/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java
@@ -58,6 +58,12 @@ public final class RESTCatalogProperties {
public static final long REST_SCAN_PLANNING_POLL_TIMEOUT_MS_DEFAULT =
TimeUnit.MINUTES.toMillis(5);
+ // Extra poll attempts after the first fetch of a submitted plan. Must be >= 0; 0 means a single
+ // fetch attempt with no retries.
+ public static final String REST_SCAN_PLANNING_POLL_NUM_RETRIES =
+ "rest-scan-planning.poll-num-retries";
+ public static final int REST_SCAN_PLANNING_POLL_NUM_RETRIES_DEFAULT = 10;
+
// Properties that control the behaviour of the table cache used for freshness-aware table
// loading.
public static final String TABLE_CACHE_EXPIRE_AFTER_WRITE_MS =
diff --git a/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java b/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java
index 9fa273ca16..c3be4b78a1 100644
--- a/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java
+++ b/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java
@@ -61,7 +61,6 @@ class RESTTableScan extends DataTableScan {
private static final Logger LOG = LoggerFactory.getLogger(RESTTableScan.class);
private static final long MIN_SLEEP_MS = 1000; // Initial delay
private static final long MAX_SLEEP_MS = 60 * 1000; // Max backoff delay (1 minute)
- private static final int MAX_RETRIES = 10; // Max number of poll retries
private static final double SCALE_FACTOR = 2.0; // Exponential scale factor
private static final String DEFAULT_FILE_IO_IMPL = "org.apache.iceberg.io.ResolvingFileIO";
private static final Cache<RESTTableScan, FileIO> FILEIO_TRACKER =
@@ -257,12 +256,22 @@ class RESTTableScan extends DataTableScan {
"Invalid value for %s: %s (must be positive)",
RESTCatalogProperties.REST_SCAN_PLANNING_POLL_TIMEOUT_MS,
maxWaitTimeMs);
+ int maxRetries =
+ PropertyUtil.propertyAsInt(
+ catalogProperties,
+ RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES,
+ RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES_DEFAULT);
+ Preconditions.checkArgument(
+ maxRetries >= 0,
+ "Invalid value for %s: %s (must be non-negative)",
+ RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES,
+ maxRetries);
AtomicReference<FetchPlanningResultResponse> result = new AtomicReference<>();
try {
Tasks.foreach(planId)
.exponentialBackoff(MIN_SLEEP_MS, MAX_SLEEP_MS, maxWaitTimeMs, SCALE_FACTOR)
- .retry(MAX_RETRIES)
+ .retry(maxRetries)
.onlyRetryOn(NotCompleteException.class)
.onFailure(
(id, err) -> {
@@ -310,7 +319,7 @@ class RESTTableScan extends DataTableScan {
+ " (timeout=%d ms, maxRetries=%d)",
planId,
maxWaitTimeMs,
- MAX_RETRIES),
+ maxRetries),
e);
}
diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java
index 172fcae769..7c6639f170 100644
--- a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java
+++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java
@@ -38,6 +38,7 @@ import com.fasterxml.jackson.databind.ObjectReader;
import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.iceberg.BaseTable;
@@ -1061,6 +1062,80 @@ public class TestRESTScanPlanning extends TestBaseWithRESTServer {
return new CatalogWithAdapter(catalog, adapter);
}
+ /**
+ * Catalog whose fetch-planning responses stay {@link PlanStatus#SUBMITTED} so client polling can
+ * be tested. When {@code submittedFetchLimit} is null, every fetch stays submitted. Otherwise
+ * only the first {@code submittedFetchLimit} fetches are overridden and later calls return the
+ * real completed plan.
+ */
+ private CatalogWithAdapter catalogWithSubmittedPlanningResults(
+ String catalogName,
+ Map<String, String> extraProperties,
+ AtomicInteger fetchAttempts,
+ Integer submittedFetchLimit) {
+ List<Endpoint> endpoints =
+ endpointsWithPlanning(
+ Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN,
+ Endpoint.V1_FETCH_TABLE_SCAN_PLAN,
+ Endpoint.V1_CANCEL_TABLE_SCAN_PLAN,
+ Endpoint.V1_FETCH_TABLE_SCAN_PLAN_TASKS);
+
+ RESTCatalogAdapter adapter =
+ Mockito.spy(
+ new RESTCatalogAdapter(backendCatalog) {
+ @Override
+ public <T extends RESTResponse> T execute(
+ HTTPRequest request,
+ Class<T> responseType,
+ Consumer<ErrorResponse> errorHandler,
+ Consumer<Map<String, String>> responseHeaders,
+ ParserContext parserContext) {
+ if (ResourcePaths.config().equals(request.path())) {
+ return castResponse(
+ responseType, ConfigResponse.builder().withEndpoints(endpoints).build());
+ }
+ T response =
+ super.execute(
+ request, responseType, errorHandler, responseHeaders, parserContext);
+ if (response instanceof LoadTableResponse) {
+ return castResponse(
+ responseType,
+ withPlanningMode(
+ (LoadTableResponse) response,
+ RESTCatalogProperties.ScanPlanningMode.SERVER.modeName()));
+ }
+
+ if (response instanceof FetchPlanningResultResponse) {
+ int attempt = fetchAttempts == null ? 0 : fetchAttempts.incrementAndGet();
+ if (submittedFetchLimit == null || attempt <= submittedFetchLimit) {
+ return castResponse(
+ responseType,
+ FetchPlanningResultResponse.builder()
+ .withPlanStatus(PlanStatus.SUBMITTED)
+ .build());
+ }
+ }
+
+ return response;
+ }
+ });
+
+ adapter.setPlanningBehavior(TestPlanningBehavior.builder().asynchronous().build());
+
+ RESTCatalog catalog =
+ new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
+ catalog.initialize(
+ catalogName,
+ ImmutableMap.<String, String>builder()
+ .put(CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO")
+ .put(
+ RESTCatalogProperties.SCAN_PLANNING_MODE,
+ RESTCatalogProperties.ScanPlanningMode.SERVER.modeName())
+ .putAll(extraProperties)
+ .build());
+ return new CatalogWithAdapter(catalog, adapter);
+ }
+
@Test
public void serverDoesNotSupportPlanningEndpoint() throws IOException {
// Server requires server-side planning but doesn't support the endpoint - should fail
@@ -1155,67 +1230,14 @@ public class TestRESTScanPlanning extends TestBaseWithRESTServer {
@Test
public void asyncPlanningRespectsConfigurablePollTimeout() {
- // Create an adapter that always returns SUBMITTED (never completes)
- List<Endpoint> endpoints =
- endpointsWithPlanning(
- Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN,
- Endpoint.V1_FETCH_TABLE_SCAN_PLAN,
- Endpoint.V1_CANCEL_TABLE_SCAN_PLAN,
- Endpoint.V1_FETCH_TABLE_SCAN_PLAN_TASKS);
-
- RESTCatalogAdapter adapter =
- Mockito.spy(
- new RESTCatalogAdapter(backendCatalog) {
- @Override
- public <T extends RESTResponse> T execute(
- HTTPRequest request,
- Class<T> responseType,
- Consumer<ErrorResponse> errorHandler,
- Consumer<Map<String, String>> responseHeaders,
- ParserContext parserContext) {
- if (ResourcePaths.config().equals(request.path())) {
- return castResponse(
- responseType, ConfigResponse.builder().withEndpoints(endpoints).build());
- }
- T response =
- super.execute(
- request, responseType, errorHandler, responseHeaders, parserContext);
- if (response instanceof LoadTableResponse) {
- return castResponse(
- responseType,
- withPlanningMode(
- (LoadTableResponse) response,
- RESTCatalogProperties.ScanPlanningMode.SERVER.modeName()));
- }
-
- // Override fetch responses to always return SUBMITTED so the poll never completes
- if (response instanceof FetchPlanningResultResponse) {
- return castResponse(
- responseType,
- FetchPlanningResultResponse.builder()
- .withPlanStatus(PlanStatus.SUBMITTED)
- .build());
- }
-
- return response;
- }
- });
-
- adapter.setPlanningBehavior(TestPlanningBehavior.builder().asynchronous().build());
-
- RESTCatalog catalog =
- new RESTCatalog(SessionCatalog.SessionContext.createEmpty(), (config) -> adapter);
- catalog.initialize(
- "test-poll-timeout",
- ImmutableMap.of(
- CatalogProperties.FILE_IO_IMPL,
- "org.apache.iceberg.inmemory.InMemoryFileIO",
- RESTCatalogProperties.SCAN_PLANNING_MODE,
- RESTCatalogProperties.ScanPlanningMode.SERVER.modeName(),
- RESTCatalogProperties.REST_SCAN_PLANNING_POLL_TIMEOUT_MS,
- "1"));
+ CatalogWithAdapter catalogWithAdapter =
+ catalogWithSubmittedPlanningResults(
+ "test-poll-timeout",
+ ImmutableMap.of(RESTCatalogProperties.REST_SCAN_PLANNING_POLL_TIMEOUT_MS, "1"),
+ null,
+ null);
- RESTTable table = restTableFor(catalog, "poll_timeout_test");
+ RESTTable table = restTableFor(catalogWithAdapter.catalog, "poll_timeout_test");
setParserContext(table);
RESTTableScan scan = restTableScanFor(table);
@@ -1284,6 +1306,83 @@ public class TestRESTScanPlanning extends TestBaseWithRESTServer {
.hasMessageContaining("must be positive");
}
+ @Test
+ public void asyncPlanningRespectsConfigurablePollRetries() {
+ AtomicInteger fetchAttempts = new AtomicInteger();
+ CatalogWithAdapter catalogWithAdapter =
+ catalogWithSubmittedPlanningResults(
+ "test-poll-retries",
+ ImmutableMap.of(RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES, "0"),
+ fetchAttempts,
+ null);
+
+ RESTTable table = restTableFor(catalogWithAdapter.catalog, "poll_retries_test");
+ setParserContext(table);
+ RESTTableScan scan = restTableScanFor(table);
+
+ // With 0 retries and a server that never completes, planFiles should fail after one attempt
+ assertThatThrownBy(scan::planFiles)
+ .isInstanceOf(RemotePlanTimeoutException.class)
+ .hasMessageContaining("did not complete within configured limits")
+ .hasMessageContaining("maxRetries=0");
+ assertThat(fetchAttempts).hasValue(1);
+ }
+
+ @Test
+ public void asyncPlanningSucceedsWithCustomRetries() {
+ // Keep the plan SUBMITTED for the first two fetches so planning only completes because more
+ // than one retry is allowed. A retry count of 1 would exhaust the budget before the third
+ // fetch.
+ int submittedResponses = 2;
+ AtomicInteger fetchAttempts = new AtomicInteger();
+ CatalogWithAdapter catalogWithAdapter =
+ catalogWithSubmittedPlanningResults(
+ "test-custom-retries",
+ ImmutableMap.of(RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES, "2"),
+ fetchAttempts,
+ submittedResponses);
+
+ RESTTable table = restTableFor(catalogWithAdapter.catalog, "custom_retries_success");
+ setParserContext(table);
+
+ assertThat(table.newScan().planFiles()).hasSize(1);
+ // initial fetch plus retries: two SUBMITTED responses followed by a completed one
+ assertThat(fetchAttempts).hasValue(submittedResponses + 1);
+ }
+
+ @Test
+ public void asyncPlanningRejectsInvalidRetries() {
+ List<Endpoint> endpoints =
+ endpointsWithPlanning(
+ Endpoint.V1_SUBMIT_TABLE_SCAN_PLAN,
+ Endpoint.V1_FETCH_TABLE_SCAN_PLAN,
+ Endpoint.V1_CANCEL_TABLE_SCAN_PLAN,
+ Endpoint.V1_FETCH_TABLE_SCAN_PLAN_TASKS);
+
+ CatalogWithAdapter catalogWithAdapter =
+ catalogWithEndpoints(endpoints, TestPlanningBehavior.builder().asynchronous().build());
+
+ // re-initialize with an invalid retry count
+ catalogWithAdapter.catalog.initialize(
+ "test-invalid-retries",
+ ImmutableMap.of(
+ CatalogProperties.FILE_IO_IMPL,
+ "org.apache.iceberg.inmemory.InMemoryFileIO",
+ RESTCatalogProperties.SCAN_PLANNING_MODE,
+ RESTCatalogProperties.ScanPlanningMode.SERVER.modeName(),
+ RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES,
+ "-1"));
+
+ RESTTable table = restTableFor(catalogWithAdapter.catalog, "invalid_retries_test");
+ setParserContext(table);
+ RESTTableScan scan = restTableScanFor(table);
+
+ assertThatThrownBy(scan::planFiles)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage(
+ "Invalid value for rest-scan-planning.poll-num-retries: -1 (must be non-negative)");
+ }
+
@ParameterizedTest
@EnumSource(PlanningMode.class)
public void planningFailsWithServerError(
diff --git a/docs/docs/rest-catalog.md b/docs/docs/rest-catalog.md
index 77a7fc8766..5660d45c8e 100644
--- a/docs/docs/rest-catalog.md
+++ b/docs/docs/rest-catalog.md
@@ -83,6 +83,8 @@ at connection time through
| `rest-page-size` | null | The page size to use when listing namespaces, tables, or other paginated resources. |
| `namespace-separator` | `%1F` | The separator character used for namespace levels when communicating with the REST server. |
| `scan-planning-mode` | `client` | Controls where scan planning is performed. Supported values: `client` (client-side planning), `server` (server-side planning). Can be overridden per-table by the server in LoadTableResponse. |
+| `rest-scan-planning.poll-timeout-ms` | `300000` (5 min) | Maximum time in milliseconds to wait when polling for async server-side scan planning results. |
+| `rest-scan-planning.poll-num-retries` | `10` | Extra poll attempts after the first fetch of a submitted plan. `0` means a single fetch with no retries. |
| `rest.client.max-retries` | `5` | Maximum number of times to retry a failed HTTP request, using exponential backoff between attempts. |
| `rest.client.max-connections` | `100` | Maximum total number of connections in the HTTP client connection pool. A JVM system property with the same name takes precedence over this catalog property. |
| `rest.client.connections-per-route` | `100` | Maximum number of pooled connections per route. |
diff --git a/docs/docs/rest-protocol.md b/docs/docs/rest-protocol.md
index 11aa4f4873..18467d0b7c 100644
--- a/docs/docs/rest-protocol.md
+++ b/docs/docs/rest-protocol.md
@@ -134,7 +134,9 @@ server either answers immediately with the completed result or returns a plan
ID that the client polls until planning finishes. Large results are returned
in batches of plan tasks that the client fetches separately. The client
cancels a plan it no longer needs, and gives up after
-`rest-scan-planning.poll-timeout-ms` (default 5 minutes).
+`rest-scan-planning.poll-timeout-ms` (default 5 minutes) or
+the initial fetch plus `rest-scan-planning.poll-num-retries` retry attempts
+(default 10), whichever limit is reached first.
The planning mode is controlled by the `scan-planning-mode` catalog property
(`client`, the default, or `server`), and the server can override the mode per