Core: Make REST scan planning poll retries configurable

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 =

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 @@ private CloseableIterable<FileScanTask> 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<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 @@ private CloseableIterable<FileScanTask> fetchPlanningResult() {
                   + " (timeout=%d ms, maxRetries=%d)",
               planId,
               maxWaitTimeMs,
-              MAX_RETRIES),
+              maxRetries),
           e);
     }
 

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<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);
+
+    AtomicInteger fetchAttempts = new AtomicInteger();
+    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) {
+                  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<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());
+
+    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<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)
+        .hasMessageContaining("must be non-negative");
+  }
+
   @ParameterizedTest
   @EnumSource(PlanningMode.class)
   public void planningFailsWithServerError(

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