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..ec920883dd2c 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 @@ 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; + // 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..f323dd5a1b97 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,141 @@ 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") + .hasMessageContaining("maxRetries=0"); + 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, + "25")); + + 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( diff --git a/docs/docs/catalog-properties.md b/docs/docs/catalog-properties.md index a5eb07b24edc..75db72316ef0 100644 --- a/docs/docs/catalog-properties.md +++ b/docs/docs/catalog-properties.md @@ -55,6 +55,8 @@ 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. | ### Table cache properties