This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new a69d93ec24 [Cherry-pick to branch-1.3] [#14473] fix(iceberg-rest):
keep the owner property in step with Gravitino ownership (#14475)
a69d93ec24 is described below
commit a69d93ec244a1e7ea31e72a7425d8a6410beaa6d
Author: Mark Hoerth <[email protected]>
AuthorDate: Mon Nov 2 01:28:56 2026 -0800
[Cherry-pick to branch-1.3] [#14473] fix(iceberg-rest): keep the owner
property in step with Gravitino ownership (#14475)
Backport to branch-1.3 of the fix for #14473. Same change as the PR to
main; the dispatcher is wired where branch-1.3 builds its owner
dispatcher.
### What changes were proposed in this pull request?
- New `OwnerPropertySyncDispatcher` in the public owner dispatcher chain
(`OwnerEventManager` -> `OwnerPropertySyncDispatcher` ->
`OwnerManager`). After `setOwner` or `setOwners` succeeds, it rewrites
the `owner` property of each namespace or table in a `lakehouse-iceberg`
catalog that has one, through the internal schema and table dispatchers.
Objects without the property, such as those created through the
Gravitino API, are left alone. Ownership has already changed when this
runs, so a failed update is logged and does not fail the transfer. The
internal owner dispatcher used by the IRC create hooks is unchanged,
since the property already names the first owner.
- New `IcebergOwnerPropertyGuard`, applied by the IRC table and
namespace executors: `owner` is dropped from `set-properties` and
`remove-properties` in table commits and from namespace property updates
and removals. A commit that creates a table (staged create) sets it to
the committing user, as a direct create does. The key is dropped rather
than rejected because engines send it on their own, for example Spark
adds the current user as `owner` on CREATE OR REPLACE, and rejecting it
would fail those statements.
### Why are the changes needed?
Gravitino and engines gave different answers for an object's owner, and
the property could be set to any name.
Fixes #14473
### Does this PR introduce _any_ user-facing change?
Yes. An ownership transfer updates the `owner` property of IRC-created
namespaces and tables; IRC property updates no longer change it. Objects
transferred before this change keep their old value until their next
transfer.
### How was this patch tested?
- `TestOwnerPropertySyncDispatcher`: a transfer rewrites the property of
an Iceberg table and namespace; objects without it or outside Iceberg
catalogs are untouched; a failed property update does not fail the
transfer; a failed transfer does not touch the property.
- `TestIcebergOwnerPropertyGuard`: commits and namespace updates cannot
set or remove `owner`; a create commit sets it to the committer; other
requests are unchanged.
`./gradlew :core:test --tests '*Owner*'
:iceberg:iceberg-rest-server:test -PskipITs`
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01Eu6fb9pLfmrSJzGQDVLi1g
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/gravitino/GravitinoEnv.java | 7 +-
.../authorization/OwnerPropertySyncDispatcher.java | 143 +++++++++++++++++++
.../TestOwnerPropertySyncDispatcher.java | 154 +++++++++++++++++++++
.../IcebergNamespaceOperationExecutor.java | 5 +-
.../dispatcher/IcebergOwnerPropertyGuard.java | 132 ++++++++++++++++++
.../dispatcher/IcebergTableOperationExecutor.java | 4 +-
.../dispatcher/TestIcebergOwnerPropertyGuard.java | 105 ++++++++++++++
7 files changed, 547 insertions(+), 3 deletions(-)
diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
index 4fee8c7d27..f39a2ecc25 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -27,6 +27,7 @@ import org.apache.gravitino.authorization.GravitinoAuthorizer;
import org.apache.gravitino.authorization.OwnerDispatcher;
import org.apache.gravitino.authorization.OwnerEventManager;
import org.apache.gravitino.authorization.OwnerManager;
+import org.apache.gravitino.authorization.OwnerPropertySyncDispatcher;
import org.apache.gravitino.auxiliary.AuxiliaryServiceManager;
import org.apache.gravitino.catalog.CatalogDispatcher;
import org.apache.gravitino.catalog.CatalogManager;
@@ -885,7 +886,11 @@ public class GravitinoEnv {
new AccessControlEventDispatcher(eventBus,
accessControlHookDispatcher);
OwnerDispatcher ownerManager = new OwnerManager(entityStore);
this.internalOwnerDispatcher = ownerManager;
- this.ownerDispatcher = new OwnerEventManager(eventBus, ownerManager);
+ // Transfers through the public dispatcher also update the owner
property of Iceberg
+ // namespaces and tables. Internal callers, such as the Iceberg REST
catalog's create hooks,
+ // set the first owner, which the property already names.
+ this.ownerDispatcher =
+ new OwnerEventManager(eventBus, new
OwnerPropertySyncDispatcher(ownerManager));
this.futureGrantManager = new FutureGrantManager(entityStore,
ownerManager);
} else {
this.accessControlDispatcher = null;
diff --git
a/core/src/main/java/org/apache/gravitino/authorization/OwnerPropertySyncDispatcher.java
b/core/src/main/java/org/apache/gravitino/authorization/OwnerPropertySyncDispatcher.java
new file mode 100644
index 0000000000..4337920152
--- /dev/null
+++
b/core/src/main/java/org/apache/gravitino/authorization/OwnerPropertySyncDispatcher.java
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.authorization;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.catalog.SchemaDispatcher;
+import org.apache.gravitino.catalog.TableDispatcher;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableChange;
+import org.apache.gravitino.utils.MetadataObjectUtil;
+import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Keeps the {@code owner} property of Iceberg namespaces and tables in step
with Gravitino
+ * ownership. The Iceberg REST catalog writes that property, the conventional
Iceberg and Hive owner
+ * key that engines display, when it creates a namespace or table. After an
ownership transfer this
+ * dispatcher rewrites the property to the new owner, so engines and Gravitino
give the same answer.
+ * Objects without the property, such as those created through the Gravitino
API, are left alone.
+ */
+public class OwnerPropertySyncDispatcher implements OwnerDispatcher {
+
+ /** The Iceberg and Hive property that names an object's owner. */
+ @VisibleForTesting static final String OWNER_PROPERTY = "owner";
+
+ private static final Logger LOG =
LoggerFactory.getLogger(OwnerPropertySyncDispatcher.class);
+ private static final String ICEBERG_PROVIDER = "lakehouse-iceberg";
+
+ private final OwnerDispatcher dispatcher;
+
+ /**
+ * Creates a dispatcher that sets owners through {@code dispatcher} and then
updates the owner
+ * property.
+ *
+ * @param dispatcher the dispatcher that records ownership
+ */
+ public OwnerPropertySyncDispatcher(OwnerDispatcher dispatcher) {
+ this.dispatcher = dispatcher;
+ }
+
+ @Override
+ public void setOwner(
+ String metalake, MetadataObject metadataObject, String ownerName,
Owner.Type ownerType) {
+ dispatcher.setOwner(metalake, metadataObject, ownerName, ownerType);
+ syncOwnerProperty(metalake, metadataObject, ownerName);
+ }
+
+ @Override
+ public void setOwners(
+ String metalake,
+ List<MetadataObject> metadataObjects,
+ String ownerName,
+ Owner.Type ownerType) {
+ dispatcher.setOwners(metalake, metadataObjects, ownerName, ownerType);
+ if (metadataObjects != null) {
+ metadataObjects.forEach(object -> syncOwnerProperty(metalake, object,
ownerName));
+ }
+ }
+
+ @Override
+ public Optional<Owner> getOwner(String metalake, MetadataObject
metadataObject) {
+ return dispatcher.getOwner(metalake, metadataObject);
+ }
+
+ /**
+ * Rewrites the owner property of an Iceberg namespace or table that has
one. Ownership has
+ * already changed when this runs, so a failure is logged rather than
failing the transfer.
+ */
+ @VisibleForTesting
+ void syncOwnerProperty(String metalake, MetadataObject metadataObject,
String ownerName) {
+ MetadataObject.Type type = metadataObject.type();
+ if (type != MetadataObject.Type.SCHEMA && type !=
MetadataObject.Type.TABLE) {
+ return;
+ }
+ NameIdentifier identifier = MetadataObjectUtil.toEntityIdent(metalake,
metadataObject);
+ try {
+ if
(!isIcebergCatalog(NameIdentifierUtil.getCatalogIdentifier(identifier))) {
+ return;
+ }
+ GravitinoEnv env = GravitinoEnv.getInstance();
+ if (type == MetadataObject.Type.SCHEMA) {
+ SchemaDispatcher schemas = env.internalSchemaDispatcher();
+ Schema schema = schemas.loadSchema(identifier);
+ if (needsUpdate(schema.properties(), ownerName)) {
+ schemas.alterSchema(identifier,
SchemaChange.setProperty(OWNER_PROPERTY, ownerName));
+ }
+ } else {
+ TableDispatcher tables = env.internalTableDispatcher();
+ Table table = tables.loadTable(identifier);
+ if (needsUpdate(table.properties(), ownerName)) {
+ tables.alterTable(identifier,
TableChange.setProperty(OWNER_PROPERTY, ownerName));
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn(
+ "Ownership of {} {} moved to {}, but its {} property could not be
updated",
+ type,
+ identifier,
+ ownerName,
+ OWNER_PROPERTY,
+ e);
+ }
+ }
+
+ private static boolean isIcebergCatalog(NameIdentifier catalogIdent) {
+ Catalog catalog =
+
GravitinoEnv.getInstance().internalCatalogDispatcher().loadCatalog(catalogIdent);
+ return ICEBERG_PROVIDER.equalsIgnoreCase(catalog.provider());
+ }
+
+ @VisibleForTesting
+ static boolean needsUpdate(Map<String, String> properties, String ownerName)
{
+ return properties != null
+ && properties.containsKey(OWNER_PROPERTY)
+ && !ownerName.equals(properties.get(OWNER_PROPERTY));
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestOwnerPropertySyncDispatcher.java
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerPropertySyncDispatcher.java
new file mode 100644
index 0000000000..5a9f0389b7
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestOwnerPropertySyncDispatcher.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.authorization;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Arrays;
+import java.util.Collections;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Schema;
+import org.apache.gravitino.SchemaChange;
+import org.apache.gravitino.catalog.CatalogDispatcher;
+import org.apache.gravitino.catalog.SchemaDispatcher;
+import org.apache.gravitino.catalog.TableDispatcher;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableChange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class TestOwnerPropertySyncDispatcher {
+
+ private static final String METALAKE = "m";
+ private static final MetadataObject TABLE =
+ MetadataObjects.of(Arrays.asList("cat", "ns", "t"),
MetadataObject.Type.TABLE);
+ private static final MetadataObject SCHEMA =
+ MetadataObjects.of(Arrays.asList("cat", "ns"),
MetadataObject.Type.SCHEMA);
+ private static final NameIdentifier TABLE_IDENT =
NameIdentifier.of(METALAKE, "cat", "ns", "t");
+ private static final NameIdentifier SCHEMA_IDENT =
NameIdentifier.of(METALAKE, "cat", "ns");
+
+ private final GravitinoEnv env = GravitinoEnv.getInstance();
+ private Object previousCatalogs;
+ private Object previousSchemas;
+ private Object previousTables;
+ private OwnerDispatcher delegate;
+ private Catalog catalog;
+ private SchemaDispatcher schemas;
+ private TableDispatcher tables;
+ private OwnerPropertySyncDispatcher dispatcher;
+
+ @BeforeEach
+ public void setUp() throws IllegalAccessException {
+ previousCatalogs = FieldUtils.readField(env, "internalCatalogDispatcher",
true);
+ previousSchemas = FieldUtils.readField(env, "internalSchemaDispatcher",
true);
+ previousTables = FieldUtils.readField(env, "internalTableDispatcher",
true);
+ CatalogDispatcher catalogs = mock(CatalogDispatcher.class);
+ catalog = mock(Catalog.class);
+ when(catalogs.loadCatalog(NameIdentifier.of(METALAKE,
"cat"))).thenReturn(catalog);
+ when(catalog.provider()).thenReturn("lakehouse-iceberg");
+ schemas = mock(SchemaDispatcher.class);
+ tables = mock(TableDispatcher.class);
+ FieldUtils.writeField(env, "internalCatalogDispatcher", catalogs, true);
+ FieldUtils.writeField(env, "internalSchemaDispatcher", schemas, true);
+ FieldUtils.writeField(env, "internalTableDispatcher", tables, true);
+ delegate = mock(OwnerDispatcher.class);
+ dispatcher = new OwnerPropertySyncDispatcher(delegate);
+ }
+
+ @AfterEach
+ public void tearDown() throws IllegalAccessException {
+ FieldUtils.writeField(env, "internalCatalogDispatcher", previousCatalogs,
true);
+ FieldUtils.writeField(env, "internalSchemaDispatcher", previousSchemas,
true);
+ FieldUtils.writeField(env, "internalTableDispatcher", previousTables,
true);
+ }
+
+ @Test
+ public void testTransferRewritesTheOwnerPropertyOfAnIcebergTable() {
+ Table table = mock(Table.class);
+ when(table.properties()).thenReturn(ImmutableMap.of("owner", "admin"));
+ when(tables.loadTable(TABLE_IDENT)).thenReturn(table);
+
+ dispatcher.setOwner(METALAKE, TABLE, "writer", Owner.Type.USER);
+
+ verify(delegate).setOwner(METALAKE, TABLE, "writer", Owner.Type.USER);
+ verify(tables).alterTable(TABLE_IDENT, TableChange.setProperty("owner",
"writer"));
+ }
+
+ @Test
+ public void testTransferRewritesTheOwnerPropertyOfIcebergNamespaces() {
+ Schema schema = mock(Schema.class);
+ when(schema.properties()).thenReturn(ImmutableMap.of("owner", "admin"));
+ when(schemas.loadSchema(SCHEMA_IDENT)).thenReturn(schema);
+
+ dispatcher.setOwners(METALAKE, Collections.singletonList(SCHEMA),
"writers", Owner.Type.GROUP);
+
+ verify(schemas).alterSchema(SCHEMA_IDENT,
SchemaChange.setProperty("owner", "writers"));
+ }
+
+ @Test
+ public void testObjectsWithoutThePropertyOrOutsideIcebergAreLeftAlone() {
+ Table table = mock(Table.class);
+ when(table.properties()).thenReturn(ImmutableMap.of("k", "v"));
+ when(tables.loadTable(TABLE_IDENT)).thenReturn(table);
+ dispatcher.setOwner(METALAKE, TABLE, "writer", Owner.Type.USER);
+
+ when(catalog.provider()).thenReturn("hive");
+ when(table.properties()).thenReturn(ImmutableMap.of("owner", "admin"));
+ dispatcher.setOwner(METALAKE, TABLE, "writer", Owner.Type.USER);
+
+ verify(tables, never()).alterTable(any(), any());
+ }
+
+ @Test
+ public void testAFailedPropertyUpdateDoesNotFailTheTransfer() {
+ Table table = mock(Table.class);
+ when(table.properties()).thenReturn(ImmutableMap.of("owner", "admin"));
+ when(tables.loadTable(TABLE_IDENT)).thenReturn(table);
+ when(tables.alterTable(any(), any())).thenThrow(new
IllegalStateException("catalog down"));
+
+ Assertions.assertDoesNotThrow(
+ () -> dispatcher.setOwner(METALAKE, TABLE, "writer", Owner.Type.USER));
+ verify(delegate).setOwner(METALAKE, TABLE, "writer", Owner.Type.USER);
+ }
+
+ @Test
+ public void testAFailedTransferDoesNotTouchTheProperty() {
+ Mockito.doThrow(new IllegalArgumentException("no such user"))
+ .when(delegate)
+ .setOwner(METALAKE, TABLE, "nobody", Owner.Type.USER);
+
+ Assertions.assertThrows(
+ IllegalArgumentException.class,
+ () -> dispatcher.setOwner(METALAKE, TABLE, "nobody", Owner.Type.USER));
+ verify(tables, never()).loadTable(any());
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
index 353d46b68e..c0be4675ac 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergNamespaceOperationExecutor.java
@@ -104,7 +104,10 @@ public class IcebergNamespaceOperationExecutor implements
IcebergNamespaceOperat
validatePropertyUpdate(updateNamespacePropertiesRequest);
return icebergCatalogWrapperManager
.getCatalogWrapper(context.catalogName())
- .updateNamespaceProperties(namespace,
updateNamespacePropertiesRequest);
+ .updateNamespaceProperties(
+ namespace,
+ IcebergOwnerPropertyGuard.protect(
+ updateNamespacePropertiesRequest, context.userName()));
}
@Override
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergOwnerPropertyGuard.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergOwnerPropertyGuard.java
new file mode 100644
index 0000000000..6085cacdd2
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergOwnerPropertyGuard.java
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.iceberg.service.dispatcher;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.gravitino.auth.AuthConstants;
+import org.apache.gravitino.catalog.lakehouse.iceberg.IcebergConstants;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+
+/**
+ * Keeps the {@code owner} property of namespaces and tables under Gravitino's
control. The Iceberg
+ * REST catalog sets it to the creator, and an ownership transfer in Gravitino
rewrites it; a
+ * property update through the Iceberg REST catalog cannot change it.
+ *
+ * <p>Updates to the property are dropped rather than rejected: engines send
it on their own, for
+ * example Spark adds the current user as {@code owner} when it creates or
replaces a table, and
+ * rejecting it would fail those statements.
+ */
+final class IcebergOwnerPropertyGuard {
+
+ private IcebergOwnerPropertyGuard() {}
+
+ /**
+ * Returns the request without changes to the owner property. A commit that
creates the table sets
+ * the property to the authenticated user instead, as a direct create does.
+ *
+ * @param request the table commit
+ * @param user the authenticated user
+ * @return the request to apply
+ */
+ static UpdateTableRequest protect(UpdateTableRequest request, String user) {
+ if (AuthConstants.ANONYMOUS_USER.equals(user) || !touchesOwner(request)) {
+ return request;
+ }
+ boolean creates =
+ request.requirements().stream()
+
.anyMatch(UpdateRequirement.AssertTableDoesNotExist.class::isInstance);
+ List<MetadataUpdate> updates = new ArrayList<>();
+ for (MetadataUpdate update : request.updates()) {
+ if (update instanceof MetadataUpdate.SetProperties) {
+ Map<String, String> properties =
+ new HashMap<>(((MetadataUpdate.SetProperties) update).updated());
+ if (creates) {
+ properties.put(IcebergConstants.OWNER, user);
+ } else {
+ properties.remove(IcebergConstants.OWNER);
+ }
+ if (!properties.isEmpty()) {
+ updates.add(new MetadataUpdate.SetProperties(properties));
+ }
+ } else if (update instanceof MetadataUpdate.RemoveProperties) {
+ List<String> removed =
+ ((MetadataUpdate.RemoveProperties) update)
+ .removed().stream()
+ .filter(key -> !IcebergConstants.OWNER.equals(key))
+ .collect(Collectors.toList());
+ if (!removed.isEmpty()) {
+ updates.add(new MetadataUpdate.RemoveProperties(new
HashSet<>(removed)));
+ }
+ } else {
+ updates.add(update);
+ }
+ }
+ return request.identifier() == null
+ ? new UpdateTableRequest(request.requirements(), updates)
+ : UpdateTableRequest.create(request.identifier(),
request.requirements(), updates);
+ }
+
+ /**
+ * Returns the request without changes to the owner property.
+ *
+ * @param request the namespace property update
+ * @param user the authenticated user
+ * @return the request to apply
+ */
+ static UpdateNamespacePropertiesRequest protect(
+ UpdateNamespacePropertiesRequest request, String user) {
+ if (AuthConstants.ANONYMOUS_USER.equals(user)
+ || (!request.updates().containsKey(IcebergConstants.OWNER)
+ && !request.removals().contains(IcebergConstants.OWNER))) {
+ return request;
+ }
+ Map<String, String> updates = new HashMap<>(request.updates());
+ updates.remove(IcebergConstants.OWNER);
+ List<String> removals =
+ request.removals().stream()
+ .filter(key -> !IcebergConstants.OWNER.equals(key))
+ .collect(Collectors.toList());
+ return UpdateNamespacePropertiesRequest.builder()
+ .updateAll(updates)
+ .removeAll(removals)
+ .build();
+ }
+
+ private static boolean touchesOwner(UpdateTableRequest request) {
+ return request.updates().stream()
+ .anyMatch(
+ update ->
+ (update instanceof MetadataUpdate.SetProperties
+ && ((MetadataUpdate.SetProperties) update)
+ .updated()
+ .containsKey(IcebergConstants.OWNER))
+ || (update instanceof MetadataUpdate.RemoveProperties
+ && ((MetadataUpdate.RemoveProperties) update)
+ .removed()
+ .contains(IcebergConstants.OWNER)));
+ }
+}
diff --git
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
index d95ee792e8..4fd9c39709 100644
---
a/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
+++
b/iceberg/iceberg-rest-server/src/main/java/org/apache/gravitino/iceberg/service/dispatcher/IcebergTableOperationExecutor.java
@@ -124,7 +124,9 @@ public class IcebergTableOperationExecutor implements
IcebergTableOperationDispa
IcebergColumnFieldValidator.validateUpdate(updateTableRequest);
return icebergCatalogWrapperManager
.getCatalogWrapper(context.catalogName())
- .updateTable(tableIdentifier, updateTableRequest);
+ .updateTable(
+ tableIdentifier,
+ IcebergOwnerPropertyGuard.protect(updateTableRequest,
context.userName()));
}
@Override
diff --git
a/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergOwnerPropertyGuard.java
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergOwnerPropertyGuard.java
new file mode 100644
index 0000000000..ffbe4f11a9
--- /dev/null
+++
b/iceberg/iceberg-rest-server/src/test/java/org/apache/gravitino/iceberg/service/dispatcher/TestIcebergOwnerPropertyGuard.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.iceberg.service.dispatcher;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import org.apache.iceberg.MetadataUpdate;
+import org.apache.iceberg.UpdateRequirement;
+import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
+import org.apache.iceberg.rest.requests.UpdateTableRequest;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class TestIcebergOwnerPropertyGuard {
+
+ @Test
+ public void testCommitCannotChangeTheOwnerProperty() {
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ Collections.emptyList(),
+ Arrays.asList(
+ new MetadataUpdate.SetProperties(ImmutableMap.of("owner",
"mallory", "k", "v")),
+ new MetadataUpdate.RemoveProperties(ImmutableSet.of("owner")),
+ new MetadataUpdate.SetCurrentSchema(0)));
+
+ List<MetadataUpdate> updates = IcebergOwnerPropertyGuard.protect(request,
"bob").updates();
+
+ Assertions.assertEquals(2, updates.size());
+ Assertions.assertEquals(
+ ImmutableMap.of("k", "v"), ((MetadataUpdate.SetProperties)
updates.get(0)).updated());
+ Assertions.assertInstanceOf(MetadataUpdate.SetCurrentSchema.class,
updates.get(1));
+ }
+
+ @Test
+ public void testCreateCommitSetsTheOwnerToTheCommitter() {
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ Collections.singletonList(new
UpdateRequirement.AssertTableDoesNotExist()),
+ Collections.singletonList(
+ new MetadataUpdate.SetProperties(ImmutableMap.of("owner",
"spark_user"))));
+
+ MetadataUpdate.SetProperties update =
+ (MetadataUpdate.SetProperties)
+ IcebergOwnerPropertyGuard.protect(request, "bob").updates().get(0);
+
+ Assertions.assertEquals(ImmutableMap.of("owner", "bob"), update.updated());
+ }
+
+ @Test
+ public void testRequestsWithoutTheOwnerPropertyOrUserAreUnchanged() {
+ UpdateTableRequest request =
+ new UpdateTableRequest(
+ Collections.emptyList(),
+ Collections.singletonList(new
MetadataUpdate.SetProperties(ImmutableMap.of("k", "v"))));
+ Assertions.assertSame(request, IcebergOwnerPropertyGuard.protect(request,
"bob"));
+
+ UpdateTableRequest withOwner =
+ new UpdateTableRequest(
+ Collections.emptyList(),
+ Collections.singletonList(
+ new MetadataUpdate.SetProperties(ImmutableMap.of("owner",
"x"))));
+ Assertions.assertSame(withOwner,
IcebergOwnerPropertyGuard.protect(withOwner, "anonymous"));
+ }
+
+ @Test
+ public void testNamespaceUpdateCannotChangeTheOwnerProperty() {
+ UpdateNamespacePropertiesRequest request =
+ UpdateNamespacePropertiesRequest.builder()
+ .update("owner", "mallory")
+ .update("k", "v")
+ .remove("old")
+ .build();
+
+ UpdateNamespacePropertiesRequest protectedRequest =
+ IcebergOwnerPropertyGuard.protect(request, "bob");
+
+ Assertions.assertEquals(ImmutableMap.of("k", "v"),
protectedRequest.updates());
+ Assertions.assertEquals(Collections.singletonList("old"),
protectedRequest.removals());
+
+ UpdateNamespacePropertiesRequest removal =
+ IcebergOwnerPropertyGuard.protect(
+
UpdateNamespacePropertiesRequest.builder().remove("owner").remove("old").build(),
+ "bob");
+ Assertions.assertEquals(Collections.singletonList("old"),
removal.removals());
+ }
+}
