[#10684] feat(iceberg-rest): Support vended credentials on registerTable endpoint

Wire the X-Iceberg-Access-Delegation header through the registerTable
REST endpoint so that clients can request credential vending, matching
the existing createTable and loadTable behavior.

Vend WRITE credentials because the registering user becomes the table
owner (IcebergNamespaceHookDispatcher.setTableOwner), consistent with
createTable.

Author: sachinnn99

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/CatalogWrapperForREST.java

@@ -80,6 +80,7 @@
 import org.apache.iceberg.rest.RESTCatalog;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
 import org.apache.iceberg.rest.requests.PlanTableScanRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
@@ -156,6 +157,21 @@ public LoadTableResponse loadTable(
     return loadTableResponse;
   }
 
+  public LoadTableResponse registerTable(
+      Namespace namespace, RegisterTableRequest request, boolean requestCredential) {
+    LoadTableResponse loadTableResponse = super.registerTable(namespace, request);
+    if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
+      // Vend WRITE credentials: the registering user becomes the table owner
+      // (IcebergNamespaceHookDispatcher.setTableOwner runs after this call
+      // returns), consistent with createTable which also vends WRITE.
+      return injectCredentialConfig(
+          TableIdentifier.of(namespace, request.name()),
+          loadTableResponse,
+          CredentialPrivilege.WRITE);
+    }
+    return loadTableResponse;
+  }
+
   @Override
   public LoadTableResponse updateTable(
       TableIdentifier tableIdentifier, UpdateTableRequest updateTableRequest) {
@@ -286,7 +302,8 @@ protected boolean useDifferentClassLoader() {
     return false;
   }
 
-  private LoadTableResponse injectCredentialConfig(
+  @VisibleForTesting
+  protected LoadTableResponse injectCredentialConfig(
       TableIdentifier tableIdentifier,
       LoadTableResponse loadTableResponse,
       CredentialPrivilege privilege) {
@@ -336,7 +353,8 @@ private Credential getCredential(
     return credential;
   }
 
-  private boolean shouldGenerateCredential(
+  @VisibleForTesting
+  protected boolean shouldGenerateCredential(
       LoadTableResponse loadTableResponse, boolean requestCredential) {
     if (!requestCredential) {
       return false;

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java

@@ -123,6 +123,6 @@ public LoadTableResponse registerTable(
       RegisterTableRequest registerTableRequest) {
     return icebergCatalogWrapperManager
         .getCatalogWrapper(context.catalogName())
-        .registerTable(namespace, registerTableRequest);
+        .registerTable(namespace, registerTableRequest, context.requestCredentialVending());
   }
 }

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergNamespaceOperations.java

@@ -33,6 +33,7 @@
 import javax.ws.rs.Encoded;
 import javax.ws.rs.GET;
 import javax.ws.rs.HEAD;
+import javax.ws.rs.HeaderParam;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
@@ -300,20 +301,25 @@ public Response registerTable(
       @AuthorizationMetadata(type = Entity.EntityType.CATALOG) @PathParam("prefix") String prefix,
       @AuthorizationMetadata(type = Entity.EntityType.SCHEMA) @Encoded() @PathParam("namespace")
           String namespace,
-      RegisterTableRequest registerTableRequest) {
+      RegisterTableRequest registerTableRequest,
+      @HeaderParam(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION) String accessDelegation) {
+    boolean isCredentialVending = IcebergTableOperations.isCredentialVending(accessDelegation);
     String catalogName = IcebergRESTUtils.getCatalogName(prefix);
     Namespace icebergNS = RESTUtil.decodeNamespace(namespace);
     LOG.info(
-        "Register Iceberg table, catalog: {}, namespace: {}, registerTableRequest: {}",
+        "Register Iceberg table, catalog: {}, namespace: {}, registerTableRequest: {}, "
+            + "accessDelegation: {}, isCredentialVending: {}",
         catalogName,
         icebergNS,
-        registerTableRequest);
+        registerTableRequest,
+        accessDelegation,
+        isCredentialVending);
     try {
       return Utils.doAs(
           httpRequest,
           () -> {
             IcebergRequestContext context =
-                new IcebergRequestContext(httpServletRequest(), catalogName);
+                new IcebergRequestContext(httpServletRequest(), catalogName, isCredentialVending);
             LoadTableResponse loadTableResponse =
                 namespaceOperationDispatcher.registerTable(
                     context, icebergNS, registerTableRequest);

iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/rest/IcebergTableOperations.java

@@ -603,7 +603,13 @@ private static boolean etagMatches(String ifNoneMatch, EntityTag etag) {
     return etag.getValue().equals(clientEtag);
   }
 
-  private boolean isCredentialVending(String accessDelegation) {
+  /**
+   * Parses the {@code X-Iceberg-Access-Delegation} header value and returns whether the client is
+   * requesting credential vending. Package-private and static so that {@link
+   * IcebergNamespaceOperations#registerTable} can reuse the same parsing logic from the same
+   * package.
+   */
+  static boolean isCredentialVending(String accessDelegation) {
     if (StringUtils.isBlank(accessDelegation)) {
       return false;
     }

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/CatalogWrapperForTest.java

@@ -22,6 +22,7 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import org.apache.gravitino.credential.CredentialPrivilege;
 import org.apache.gravitino.iceberg.common.IcebergConfig;
 import org.apache.gravitino.iceberg.service.CatalogWrapperForREST;
 import org.apache.iceberg.DataFile;
@@ -41,7 +42,9 @@
 import org.apache.iceberg.types.Types.NestedField;
 import org.apache.iceberg.types.Types.StringType;
 
-// Used to override registerTable
+// Test wrapper that mocks operations the in-memory catalog cannot perform natively
+// (e.g. registerTable, which requires a real metadata.json file at the given location),
+// and adds test-only hooks (e.g. plan-task data generation for createTable).
 @SuppressWarnings("deprecation")
 public class CatalogWrapperForTest extends CatalogWrapperForREST {
   public static final String GENERATE_PLAN_TASKS_DATA_PROP = "test.generate-plan-data";
@@ -61,23 +64,44 @@ public LoadTableResponse createTable(
   }
 
   @Override
-  public LoadTableResponse registerTable(Namespace namespace, RegisterTableRequest request) {
+  public LoadTableResponse registerTable(
+      Namespace namespace, RegisterTableRequest request, boolean requestCredential) {
     if (request.name().contains("fail")) {
       throw new AlreadyExistsException("Already exits exception for test");
     }
 
+    // The in-memory test catalog cannot natively registerTable (it would need a real
+    // metadata.json file at the given location), so build a mock LoadTableResponse here.
+    // Honor cloud URIs (e.g. s3://) in metadataLocation so credential vending tests can
+    // verify the vended path; default to /mock otherwise for existing tests.
+    String location =
+        request.metadataLocation().contains("://") ? request.metadataLocation() : "/mock";
     Schema mockSchema = new Schema(NestedField.of(1, false, "foo_string", StringType.get()));
     TableMetadata baseMetadata =
         TableMetadata.newTableMetadata(
-            mockSchema, PartitionSpec.unpartitioned(), "/mock", ImmutableMap.of());
+            mockSchema, PartitionSpec.unpartitioned(), location, ImmutableMap.of());
     String json = TableMetadataParser.toJson(baseMetadata);
     TableMetadata tableMetadata =
-        TableMetadataParser.fromJson("/mock/metadata/v1.metadata.json", json);
+        TableMetadataParser.fromJson(location + "/metadata/v1.metadata.json", json);
     LoadTableResponse loadTableResponse =
         LoadTableResponse.builder()
             .withTableMetadata(tableMetadata)
             .addAllConfig(ImmutableMap.of())
             .build();
+    // We must replicate the credential-vending check + injection here (rather than reuse
+    // CatalogWrapperForREST.registerTable via super) because the in-memory test catalog
+    // cannot natively perform registerTable. Above, we synthesized a mock LoadTableResponse
+    // instead of delegating to super.registerTable; that means the parent class never sees
+    // this call and therefore never runs its vending logic. Calling shouldGenerateCredential
+    // / injectCredentialConfig directly (now protected on the parent) re-applies the exact
+    // same vending behavior to the mock response, so credential-vending tests exercise the
+    // production code path end-to-end. Privilege must match the production wrapper (WRITE).
+    if (shouldGenerateCredential(loadTableResponse, requestCredential)) {
+      return injectCredentialConfig(
+          TableIdentifier.of(namespace, request.name()),
+          loadTableResponse,
+          CredentialPrivilege.WRITE);
+    }
     return loadTableResponse;
   }
 

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/IcebergNamespaceTestBase.java

@@ -59,6 +59,18 @@ protected Response doRegisterTable(String tableName, Namespace ns) {
         .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
   }
 
+  protected Response doRegisterTableWithCredentialVending(
+      String tableName, Namespace ns, String metadataLocation) {
+    RegisterTableRequest request =
+        ImmutableRegisterTableRequest.builder()
+            .name(tableName)
+            .metadataLocation(metadataLocation)
+            .build();
+    return getNamespaceClientBuilder(Optional.of(ns), Optional.of("register"), Optional.empty())
+        .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "vended-credentials")
+        .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+  }
+
   private Response doListNamespace(Optional<Namespace> parent) {
     Optional<Map<String, String>> queryParam =
         parent.isPresent()

iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/rest/TestIcebergNamespaceOperations.java

@@ -21,10 +21,15 @@
 import java.util.Arrays;
 import java.util.Optional;
 import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.client.Entity;
 import javax.ws.rs.core.Application;
 import javax.ws.rs.core.EntityTag;
+import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import org.apache.gravitino.credential.Credential;
 import org.apache.gravitino.iceberg.service.IcebergRESTUtils;
+import org.apache.gravitino.iceberg.service.extension.DummyCredentialProvider;
 import org.apache.gravitino.listener.api.event.Event;
 import org.apache.gravitino.listener.api.event.IcebergCreateNamespaceEvent;
 import org.apache.gravitino.listener.api.event.IcebergCreateNamespaceFailureEvent;
@@ -44,6 +49,9 @@
 import org.apache.gravitino.listener.api.event.IcebergUpdateNamespaceFailureEvent;
 import org.apache.gravitino.listener.api.event.IcebergUpdateNamespacePreEvent;
 import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.rest.requests.ImmutableRegisterTableRequest;
+import org.apache.iceberg.rest.requests.RegisterTableRequest;
+import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.glassfish.jersey.internal.inject.AbstractBinder;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.junit.jupiter.api.Assertions;
@@ -257,4 +265,68 @@ void testUpdateNamespace() {
     dummyEventListener.clearEvent();
     verifyUpdateNamespaceSucc(Namespace.of("update_foo3", "a"));
   }
+
+  @Test
+  void testRegisterTableWithCredentialVending() {
+    // register without credential vending -- no credentials in response
+    verifyRegisterTableSucc("register_cred_foo1", Namespace.of("register_cred_ns"));
+
+    // register with credential vending but local location -- should NOT vend
+    Response response =
+        doRegisterTableWithCredentialVending(
+            "register_cred_foo2", Namespace.of("register_cred_ns2"), "mock");
+    Assertions.assertEquals(Status.OK.getStatusCode(), response.getStatus());
+    LoadTableResponse loadTableResponse = response.readEntity(LoadTableResponse.class);
+    Assertions.assertFalse(loadTableResponse.config().containsKey(Credential.CREDENTIAL_TYPE));
+
+    // register with credential vending and S3 location -- SHOULD vend
+    String s3Location = "s3://dummy-bucket/register_cred_foo3";
+    response =
+        doRegisterTableWithCredentialVending(
+            "register_cred_foo3", Namespace.of("register_cred_ns3"), s3Location);
+    Assertions.assertEquals(Status.OK.getStatusCode(), response.getStatus());
+    loadTableResponse = response.readEntity(LoadTableResponse.class);
+    Assertions.assertEquals(
+        DummyCredentialProvider.DUMMY_CREDENTIAL_TYPE,
+        loadTableResponse.config().get(Credential.CREDENTIAL_TYPE));
+    // DummyCredentialProvider.SimpleCredential is not one of the typed credentials handled
+    // in CredentialPropertyUtils#toIcebergProperties, so it falls through to
+    // Credential#toProperties, which always emits credential-type and expire-time-in-ms.
+    // Asserting both guards against partial-injection regressions.
+    Assertions.assertEquals("0", loadTableResponse.config().get(Credential.EXPIRE_TIME_IN_MS));
+  }
+
+  @Test
+  void testRegisterTableRemoteSigningNotSupported() {
+    RegisterTableRequest request =
+        ImmutableRegisterTableRequest.builder()
+            .name("remote_signing_test")
+            .metadataLocation("mock")
+            .build();
+    Response response =
+        getNamespaceClientBuilder(
+                Optional.of(Namespace.of("register_remote_ns")),
+                Optional.of("register"),
+                Optional.empty())
+            .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "remote-signing")
+            .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+    Assertions.assertEquals(406, response.getStatus());
+  }
+
+  @Test
+  void testRegisterTableInvalidAccessDelegation() {
+    RegisterTableRequest request =
+        ImmutableRegisterTableRequest.builder()
+            .name("invalid_delegation_test")
+            .metadataLocation("mock")
+            .build();
+    Response response =
+        getNamespaceClientBuilder(
+                Optional.of(Namespace.of("register_invalid_ns")),
+                Optional.of("register"),
+                Optional.empty())
+            .header(IcebergTableOperations.X_ICEBERG_ACCESS_DELEGATION, "invalid-value")
+            .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+    Assertions.assertEquals(400, response.getStatus());
+  }
 }