From ce5c2a7a67dd17e03282bc0a8bed669fc730023a Mon Sep 17 00:00:00 2001 From: Bartekszost <81922839+Bartekszost@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:57:40 +0000 Subject: [PATCH 1/4] Core: Make REST scan planning poll retries configurable Allow catalogs to set rest-scan-planning.poll-num-retries so long-running remote plans are not stopped by the default retry count. Generated-by: Cursor --- .../iceberg/rest/RESTCatalogProperties.java | 4 + .../apache/iceberg/rest/RESTTableScan.java | 15 +- .../iceberg/rest/TestRESTScanPlanning.java | 135 ++++++++++++++++++ 3 files changed, 151 insertions(+), 3 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 4180534bd856..b47646906417 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,10 @@ private RESTCatalogProperties() {} public static final long REST_SCAN_PLANNING_POLL_TIMEOUT_MS_DEFAULT = TimeUnit.MINUTES.toMillis(5); + 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 9fa273ca169f..c3be4b78a105 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 FILEIO_TRACKER = @@ -257,12 +256,22 @@ private CloseableIterable fetchPlanningResult() { "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 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 @@ private CloseableIterable fetchPlanningResult() { + " (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 172fcae769bf..3891d48230b5 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 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; @@ -1284,6 +1285,140 @@ public void asyncPlanningRejectsInvalidTimeout() { .hasMessageContaining("must be positive"); } + @Test + public void asyncPlanningRespectsConfigurablePollRetries() { + // Create an adapter that always returns SUBMITTED (never completes) + List 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); + + AtomicInteger fetchAttempts = new AtomicInteger(); + RESTCatalogAdapter adapter = + Mockito.spy( + new RESTCatalogAdapter(backendCatalog) { + @Override + public T execute( + HTTPRequest request, + Class responseType, + Consumer errorHandler, + Consumer> 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) { + fetchAttempts.incrementAndGet(); + 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-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, + "0")); + + RESTTable table = restTableFor(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"); + assertThat(fetchAttempts).hasValue(1); + } + + @Test + public void asyncPlanningSucceedsWithCustomRetries() { + List 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()); + + catalogWithAdapter.catalog.initialize( + "test-custom-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, + "10")); + + RESTTable table = restTableFor(catalogWithAdapter.catalog, "custom_retries_success"); + setParserContext(table); + assertThat(table.newScan().planFiles()).hasSize(1); + } + + @Test + public void asyncPlanningRejectsInvalidRetries() { + List 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) + .hasMessageContaining("must be non-negative"); + } + @ParameterizedTest @EnumSource(PlanningMode.class) public void planningFailsWithServerError( From 20ffb9940a10564765eaabd1648173844a676186 Mon Sep 17 00:00:00 2001 From: Bartekszost <81922839+Bartekszost@users.noreply.github.com> Date: Fri, 28 Aug 2026 10:56:39 +0000 Subject: [PATCH 2/4] Core: Make REST scan planning poll retries and backoff configurable Expose min wait, max wait, and scale factor alongside poll retries so long-running remote plans can tune the full backoff policy, not only the retry count. Generated-by: Cursor --- .../iceberg/rest/RESTCatalogProperties.java | 17 +++ .../apache/iceberg/rest/RESTTableScan.java | 42 ++++++- .../iceberg/rest/TestRESTScanPlanning.java | 112 +++++++++++++++++- docs/docs/catalog-properties.md | 5 + 4 files changed, 171 insertions(+), 5 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 b47646906417..a260585b5764 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java @@ -58,10 +58,27 @@ private 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; + public static final String REST_SCAN_PLANNING_POLL_MIN_WAIT_MS = + "rest-scan-planning.poll-min-wait-ms"; + public static final long REST_SCAN_PLANNING_POLL_MIN_WAIT_MS_DEFAULT = + TimeUnit.SECONDS.toMillis(1); + + public static final String REST_SCAN_PLANNING_POLL_MAX_WAIT_MS = + "rest-scan-planning.poll-max-wait-ms"; + public static final long REST_SCAN_PLANNING_POLL_MAX_WAIT_MS_DEFAULT = + TimeUnit.MINUTES.toMillis(1); + + // Exponential backoff multiplier between poll attempts. Must be >= 1.0. + public static final String REST_SCAN_PLANNING_POLL_SCALE_FACTOR = + "rest-scan-planning.poll-scale-factor"; + public static final double REST_SCAN_PLANNING_POLL_SCALE_FACTOR_DEFAULT = 2.0; + // 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 c3be4b78a105..9be7a3579d57 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java @@ -59,9 +59,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 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 FILEIO_TRACKER = Caffeine.newBuilder() @@ -266,11 +263,48 @@ private CloseableIterable fetchPlanningResult() { "Invalid value for %s: %s (must be non-negative)", RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES, maxRetries); + long minWaitMs = + PropertyUtil.propertyAsLong( + catalogProperties, + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS_DEFAULT); + Preconditions.checkArgument( + minWaitMs > 0, + "Invalid value for %s: %s (must be positive)", + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, + minWaitMs); + long maxWaitMs = + PropertyUtil.propertyAsLong( + catalogProperties, + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS_DEFAULT); + Preconditions.checkArgument( + maxWaitMs > 0, + "Invalid value for %s: %s (must be positive)", + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, + maxWaitMs); + Preconditions.checkArgument( + maxWaitMs >= minWaitMs, + "Invalid values for %s (%s) and %s (%s): min wait must be <= max wait", + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, + minWaitMs, + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, + maxWaitMs); + double scaleFactor = + PropertyUtil.propertyAsDouble( + catalogProperties, + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR_DEFAULT); + Preconditions.checkArgument( + scaleFactor >= 1.0, + "Invalid value for %s: %s (must be >= 1.0)", + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, + scaleFactor); AtomicReference result = new AtomicReference<>(); try { Tasks.foreach(planId) - .exponentialBackoff(MIN_SLEEP_MS, MAX_SLEEP_MS, maxWaitTimeMs, SCALE_FACTOR) + .exponentialBackoff(minWaitMs, maxWaitMs, maxWaitTimeMs, scaleFactor) .retry(maxRetries) .onlyRetryOn(NotCompleteException.class) .onFailure( 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 3891d48230b5..903961c95ad4 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java @@ -41,6 +41,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.function.Function; +import java.util.stream.Stream; import org.apache.iceberg.BaseTable; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.DataFile; @@ -74,7 +75,9 @@ import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; @@ -1380,7 +1383,7 @@ public void asyncPlanningSucceedsWithCustomRetries() { RESTCatalogProperties.SCAN_PLANNING_MODE, RESTCatalogProperties.ScanPlanningMode.SERVER.modeName(), RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES, - "10")); + "25")); RESTTable table = restTableFor(catalogWithAdapter.catalog, "custom_retries_success"); setParserContext(table); @@ -1419,6 +1422,113 @@ public void asyncPlanningRejectsInvalidRetries() { .hasMessageContaining("must be non-negative"); } + @Test + public void asyncPlanningSucceedsWithCustomBackoff() { + List 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()); + + catalogWithAdapter.catalog.initialize( + "test-custom-backoff", + ImmutableMap.of( + CatalogProperties.FILE_IO_IMPL, + "org.apache.iceberg.inmemory.InMemoryFileIO", + RESTCatalogProperties.SCAN_PLANNING_MODE, + RESTCatalogProperties.ScanPlanningMode.SERVER.modeName(), + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, + "50", + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, + "200", + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, + "1.5")); + + RESTTable table = restTableFor(catalogWithAdapter.catalog, "custom_backoff_success"); + setParserContext(table); + assertThat(table.newScan().planFiles()).hasSize(1); + } + + @ParameterizedTest + @MethodSource("invalidPollBackoff") + public void asyncPlanningRejectsInvalidPollBackoff( + String property, String value, String message) { + List 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()); + + ImmutableMap.Builder properties = + ImmutableMap.builder() + .put(CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO") + .put( + RESTCatalogProperties.SCAN_PLANNING_MODE, + RESTCatalogProperties.ScanPlanningMode.SERVER.modeName()) + .put(property, value); + + catalogWithAdapter.catalog.initialize("test-invalid-backoff", properties.build()); + + RESTTable table = restTableFor(catalogWithAdapter.catalog, "invalid_backoff_test"); + setParserContext(table); + RESTTableScan scan = restTableScanFor(table); + + assertThatThrownBy(scan::planFiles) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining(message); + } + + private static Stream invalidPollBackoff() { + return Stream.of( + Arguments.of( + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, "-1", "must be positive"), + Arguments.of( + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, "0", "must be positive"), + Arguments.of( + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, "0.5", "must be >= 1.0")); + } + + @Test + public void asyncPlanningRejectsMinWaitGreaterThanMaxWait() { + List 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()); + + catalogWithAdapter.catalog.initialize( + "test-invalid-wait-range", + ImmutableMap.of( + CatalogProperties.FILE_IO_IMPL, + "org.apache.iceberg.inmemory.InMemoryFileIO", + RESTCatalogProperties.SCAN_PLANNING_MODE, + RESTCatalogProperties.ScanPlanningMode.SERVER.modeName(), + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, + "5000", + RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, + "1000")); + + RESTTable table = restTableFor(catalogWithAdapter.catalog, "invalid_wait_range_test"); + setParserContext(table); + RESTTableScan scan = restTableScanFor(table); + + assertThatThrownBy(scan::planFiles) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("min wait must be <= max wait"); + } + @ParameterizedTest @EnumSource(PlanningMode.class) public void planningFailsWithServerError( diff --git a/docs/docs/catalog-properties.md b/docs/docs/catalog-properties.md index a5eb07b24edc..442990436305 100644 --- a/docs/docs/catalog-properties.md +++ b/docs/docs/catalog-properties.md @@ -55,6 +55,11 @@ The following properties configure the behavior of the REST catalog client. | `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-scan-planning.poll-min-wait-ms` | `1000` (1s) | Minimum backoff in milliseconds between poll attempts. | +| `rest-scan-planning.poll-max-wait-ms` | `60000` (1 min) | Maximum backoff in milliseconds between poll attempts. | +| `rest-scan-planning.poll-scale-factor` | `2.0` | Exponential backoff multiplier between poll attempts. Must be `>= 1.0`. | ### Table cache properties From b2fe291adc38f7283bdcf1eaa828338ddd985cd7 Mon Sep 17 00:00:00 2001 From: Bartekszost <81922839+Bartekszost@users.noreply.github.com> Date: Fri, 28 Aug 2026 11:47:30 +0000 Subject: [PATCH 3/4] Core: Assert configured poll retry count in timeout exception Lock the RemotePlanTimeoutException message to include maxRetries so a dropped interpolation would fail the test. Generated-by: Cursor --- .../java/org/apache/iceberg/rest/TestRESTScanPlanning.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 903961c95ad4..d95641dfe16a 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java @@ -1359,7 +1359,8 @@ public T execute( // 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("did not complete within configured limits") + .hasMessageContaining("maxRetries=0"); assertThat(fetchAttempts).hasValue(1); } From 97ac20036d19b7ca16891e3f9ecd8d949c655fe0 Mon Sep 17 00:00:00 2001 From: Bartekszost <81922839+Bartekszost@users.noreply.github.com> Date: Mon, 31 Aug 2026 21:42:25 +0000 Subject: [PATCH 4/4] Core: Keep only configurable REST scan planning poll retries Drop min/max wait and scale-factor knobs so this PR only adds poll-num-retries; backoff stays at the previous hardcoded defaults. Generated-by: Cursor --- .../iceberg/rest/RESTCatalogProperties.java | 15 --- .../apache/iceberg/rest/RESTTableScan.java | 42 +------ .../iceberg/rest/TestRESTScanPlanning.java | 110 ------------------ docs/docs/catalog-properties.md | 3 - 4 files changed, 4 insertions(+), 166 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 a260585b5764..ec920883dd2c 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTCatalogProperties.java @@ -64,21 +64,6 @@ private RESTCatalogProperties() {} "rest-scan-planning.poll-num-retries"; public static final int REST_SCAN_PLANNING_POLL_NUM_RETRIES_DEFAULT = 10; - public static final String REST_SCAN_PLANNING_POLL_MIN_WAIT_MS = - "rest-scan-planning.poll-min-wait-ms"; - public static final long REST_SCAN_PLANNING_POLL_MIN_WAIT_MS_DEFAULT = - TimeUnit.SECONDS.toMillis(1); - - public static final String REST_SCAN_PLANNING_POLL_MAX_WAIT_MS = - "rest-scan-planning.poll-max-wait-ms"; - public static final long REST_SCAN_PLANNING_POLL_MAX_WAIT_MS_DEFAULT = - TimeUnit.MINUTES.toMillis(1); - - // Exponential backoff multiplier between poll attempts. Must be >= 1.0. - public static final String REST_SCAN_PLANNING_POLL_SCALE_FACTOR = - "rest-scan-planning.poll-scale-factor"; - public static final double REST_SCAN_PLANNING_POLL_SCALE_FACTOR_DEFAULT = 2.0; - // 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 9be7a3579d57..c3be4b78a105 100644 --- a/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java +++ b/core/src/main/java/org/apache/iceberg/rest/RESTTableScan.java @@ -59,6 +59,9 @@ 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 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 FILEIO_TRACKER = Caffeine.newBuilder() @@ -263,48 +266,11 @@ private CloseableIterable fetchPlanningResult() { "Invalid value for %s: %s (must be non-negative)", RESTCatalogProperties.REST_SCAN_PLANNING_POLL_NUM_RETRIES, maxRetries); - long minWaitMs = - PropertyUtil.propertyAsLong( - catalogProperties, - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS_DEFAULT); - Preconditions.checkArgument( - minWaitMs > 0, - "Invalid value for %s: %s (must be positive)", - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, - minWaitMs); - long maxWaitMs = - PropertyUtil.propertyAsLong( - catalogProperties, - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS_DEFAULT); - Preconditions.checkArgument( - maxWaitMs > 0, - "Invalid value for %s: %s (must be positive)", - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, - maxWaitMs); - Preconditions.checkArgument( - maxWaitMs >= minWaitMs, - "Invalid values for %s (%s) and %s (%s): min wait must be <= max wait", - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, - minWaitMs, - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, - maxWaitMs); - double scaleFactor = - PropertyUtil.propertyAsDouble( - catalogProperties, - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR_DEFAULT); - Preconditions.checkArgument( - scaleFactor >= 1.0, - "Invalid value for %s: %s (must be >= 1.0)", - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, - scaleFactor); AtomicReference result = new AtomicReference<>(); try { Tasks.foreach(planId) - .exponentialBackoff(minWaitMs, maxWaitMs, maxWaitTimeMs, scaleFactor) + .exponentialBackoff(MIN_SLEEP_MS, MAX_SLEEP_MS, maxWaitTimeMs, SCALE_FACTOR) .retry(maxRetries) .onlyRetryOn(NotCompleteException.class) .onFailure( 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 d95641dfe16a..f323dd5a1b97 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java @@ -41,7 +41,6 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Consumer; import java.util.function.Function; -import java.util.stream.Stream; import org.apache.iceberg.BaseTable; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.DataFile; @@ -75,9 +74,7 @@ import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.EnumSource; -import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import org.mockito.ArgumentCaptor; import org.mockito.Mockito; @@ -1423,113 +1420,6 @@ public void asyncPlanningRejectsInvalidRetries() { .hasMessageContaining("must be non-negative"); } - @Test - public void asyncPlanningSucceedsWithCustomBackoff() { - List 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()); - - catalogWithAdapter.catalog.initialize( - "test-custom-backoff", - ImmutableMap.of( - CatalogProperties.FILE_IO_IMPL, - "org.apache.iceberg.inmemory.InMemoryFileIO", - RESTCatalogProperties.SCAN_PLANNING_MODE, - RESTCatalogProperties.ScanPlanningMode.SERVER.modeName(), - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, - "50", - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, - "200", - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, - "1.5")); - - RESTTable table = restTableFor(catalogWithAdapter.catalog, "custom_backoff_success"); - setParserContext(table); - assertThat(table.newScan().planFiles()).hasSize(1); - } - - @ParameterizedTest - @MethodSource("invalidPollBackoff") - public void asyncPlanningRejectsInvalidPollBackoff( - String property, String value, String message) { - List 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()); - - ImmutableMap.Builder properties = - ImmutableMap.builder() - .put(CatalogProperties.FILE_IO_IMPL, "org.apache.iceberg.inmemory.InMemoryFileIO") - .put( - RESTCatalogProperties.SCAN_PLANNING_MODE, - RESTCatalogProperties.ScanPlanningMode.SERVER.modeName()) - .put(property, value); - - catalogWithAdapter.catalog.initialize("test-invalid-backoff", properties.build()); - - RESTTable table = restTableFor(catalogWithAdapter.catalog, "invalid_backoff_test"); - setParserContext(table); - RESTTableScan scan = restTableScanFor(table); - - assertThatThrownBy(scan::planFiles) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining(message); - } - - private static Stream invalidPollBackoff() { - return Stream.of( - Arguments.of( - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, "-1", "must be positive"), - Arguments.of( - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, "0", "must be positive"), - Arguments.of( - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_SCALE_FACTOR, "0.5", "must be >= 1.0")); - } - - @Test - public void asyncPlanningRejectsMinWaitGreaterThanMaxWait() { - List 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()); - - catalogWithAdapter.catalog.initialize( - "test-invalid-wait-range", - ImmutableMap.of( - CatalogProperties.FILE_IO_IMPL, - "org.apache.iceberg.inmemory.InMemoryFileIO", - RESTCatalogProperties.SCAN_PLANNING_MODE, - RESTCatalogProperties.ScanPlanningMode.SERVER.modeName(), - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MIN_WAIT_MS, - "5000", - RESTCatalogProperties.REST_SCAN_PLANNING_POLL_MAX_WAIT_MS, - "1000")); - - RESTTable table = restTableFor(catalogWithAdapter.catalog, "invalid_wait_range_test"); - setParserContext(table); - RESTTableScan scan = restTableScanFor(table); - - assertThatThrownBy(scan::planFiles) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("min wait must be <= max wait"); - } - @ParameterizedTest @EnumSource(PlanningMode.class) public void planningFailsWithServerError( diff --git a/docs/docs/catalog-properties.md b/docs/docs/catalog-properties.md index 442990436305..75db72316ef0 100644 --- a/docs/docs/catalog-properties.md +++ b/docs/docs/catalog-properties.md @@ -57,9 +57,6 @@ The following properties configure the behavior of the REST catalog client. | `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-scan-planning.poll-min-wait-ms` | `1000` (1s) | Minimum backoff in milliseconds between poll attempts. | -| `rest-scan-planning.poll-max-wait-ms` | `60000` (1 min) | Maximum backoff in milliseconds between poll attempts. | -| `rest-scan-planning.poll-scale-factor` | `2.0` | Exponential backoff multiplier between poll attempts. Must be `>= 1.0`. | ### Table cache properties