This is an automated email from the ASF dual-hosted git repository.
roryqi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 94d2d02ace [#12287] feat(server): Add bulk user access-control APIs
(#12288)
94d2d02ace is described below
commit 94d2d02ace136bf339c3dbd5aec82d3c0df8064b
Author: jarred0214 <[email protected]>
AuthorDate: Wed Aug 19 15:21:58 2026 +0800
[#12287] feat(server): Add bulk user access-control APIs (#12288)
### What changes were proposed in this pull request?
This PR adds the shared foundation for best-effort bulk access-control
APIs and introduces bulk user operations.
Changes include:
- Add shared bulk request/response DTOs for item-level errors and
summary counts.
- Add `gravitino.server.bulk.maxItems`, defaulting to 100.
- Add bulk user add/remove REST APIs.
- Register the bulk REST resource.
- Add OpenAPI definitions and access-control documentation for bulk user
APIs.
- Add tests for bulk user best-effort behavior, duplicate-name
validation, and request size limit validation.
### Why are the changes needed?
Bulk access-control operations are needed to efficiently add or remove
multiple users under a metalake while preserving item-level best-effort
results.
Fix: #12287
### Does this PR introduce _any_ user-facing change?
Yes.
New REST APIs:
- `POST /api/bulk/metalakes/{metalake}/users/add`
- `POST /api/bulk/metalakes/{metalake}/users/remove`
New server configuration:
- `gravitino.server.bulk.maxItems`, default value: `100`
### How was this patch tested?
-
`JAVA_HOME=/opt/homebrew/Cellar/openjdk@17/17.0.18/libexec/openjdk.jdk/Contents/Home
./gradlew :server:spotlessApply :common:spotlessApply :server:test
--tests org.apache.gravitino.server.web.rest.TestBulkOperations
-PskipITs`
-
`JAVA_HOME=/opt/homebrew/Cellar/openjdk@17/17.0.18/libexec/openjdk.jdk/Contents/Home
./gradlew :docs:build`
- `git diff --check`
---
.../gravitino/dto/requests/BulkRemoveRequest.java | 67 +++++
.../gravitino/dto/requests/BulkUserAddRequest.java | 62 +++++
.../apache/gravitino/dto/responses/BulkError.java | 80 ++++++
.../dto/responses/BulkRemoveResponse.java | 71 ++++++
.../gravitino/dto/responses/BulkSummary.java | 67 +++++
.../gravitino/dto/responses/BulkUserResponse.java | 72 ++++++
conf/gravitino.conf.template | 2 +
.../main/java/org/apache/gravitino/Configs.java | 10 +
.../java/org/apache/gravitino/GravitinoEnv.java | 13 +
.../authorization/AccessControlDispatcher.java | 29 +++
.../authorization/AccessControlManager.java | 73 ++++++
.../org/apache/gravitino/bulk/BulkItemResult.java | 121 +++++++++
.../org/apache/gravitino/bulk/BulkManager.java | 95 +++++++
.../java/org/apache/gravitino/bulk/UserAdd.java | 80 ++++++
.../hook/AccessControlHookDispatcher.java | 16 ++
.../listener/AccessControlEventDispatcher.java | 65 +++++
.../authorization/TestAccessControlManager.java | 53 ++++
.../listener/api/event/TestUserEvent.java | 76 +++++-
.../gravitino/rewrite_gravitino_server_config.py | 4 +-
docs/gravitino-server-config.md | 2 +
docs/open-api/bulk.yaml | 275 +++++++++++++++++++++
docs/open-api/openapi.yaml | 6 +
docs/security/access-control.md | 36 +++
.../web/filter/GravitinoInterceptionService.java | 2 +
.../gravitino/server/web/rest/BulkOperations.java | 181 ++++++++++++++
.../server/web/rest/TestBulkOperations.java | 253 +++++++++++++++++++
26 files changed, 1808 insertions(+), 3 deletions(-)
diff --git
a/common/src/main/java/org/apache/gravitino/dto/requests/BulkRemoveRequest.java
b/common/src/main/java/org/apache/gravitino/dto/requests/BulkRemoveRequest.java
new file mode 100644
index 0000000000..cc89bd87f5
--- /dev/null
+++
b/common/src/main/java/org/apache/gravitino/dto/requests/BulkRemoveRequest.java
@@ -0,0 +1,67 @@
+/*
+ * 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.dto.requests;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.gravitino.rest.RESTRequest;
+
+/** Represents a request to remove named entities in bulk. */
+@Getter
+@EqualsAndHashCode
+@ToString
+public class BulkRemoveRequest implements RESTRequest {
+
+ @JsonProperty("names")
+ private final String[] names;
+
+ /**
+ * Creates a new BulkRemoveRequest.
+ *
+ * @param names The entity names.
+ */
+ public BulkRemoveRequest(String[] names) {
+ this.names = names;
+ }
+
+ /** Default constructor for BulkRemoveRequest. (Used for Jackson
deserialization.) */
+ public BulkRemoveRequest() {
+ this(null);
+ }
+
+ @Override
+ public void validate() throws IllegalArgumentException {
+ Preconditions.checkArgument(names != null && names.length > 0, "\"names\"
must not be empty");
+ Set<String> seen = new HashSet<>();
+ Arrays.stream(names)
+ .forEach(
+ name -> {
+ Preconditions.checkArgument(
+ StringUtils.isNotBlank(name), "name must not be null or
empty");
+ Preconditions.checkArgument(seen.add(name), "Duplicate name in
request: %s", name);
+ });
+ }
+}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/requests/BulkUserAddRequest.java
b/common/src/main/java/org/apache/gravitino/dto/requests/BulkUserAddRequest.java
new file mode 100644
index 0000000000..a840c2bc73
--- /dev/null
+++
b/common/src/main/java/org/apache/gravitino/dto/requests/BulkUserAddRequest.java
@@ -0,0 +1,62 @@
+/*
+ * 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.dto.requests;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.apache.gravitino.rest.RESTRequest;
+
+/** Represents a request to add users in bulk. */
+@Getter
+@EqualsAndHashCode
+@ToString
+public class BulkUserAddRequest implements RESTRequest {
+
+ @JsonProperty("users")
+ private final UserAddRequest[] users;
+
+ /**
+ * Creates a new BulkUserAddRequest.
+ *
+ * @param users The user add requests.
+ */
+ public BulkUserAddRequest(UserAddRequest[] users) {
+ this.users = users;
+ }
+
+ /** Default constructor for BulkUserAddRequest. (Used for Jackson
deserialization.) */
+ public BulkUserAddRequest() {
+ this(null);
+ }
+
+ @Override
+ public void validate() throws IllegalArgumentException {
+ Preconditions.checkArgument(users != null && users.length > 0, "\"users\"
must not be empty");
+ Arrays.stream(users)
+ .forEach(
+ user -> {
+ Preconditions.checkArgument(user != null, "user must not be
null");
+ user.validate();
+ });
+ }
+}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/responses/BulkError.java
b/common/src/main/java/org/apache/gravitino/dto/responses/BulkError.java
new file mode 100644
index 0000000000..a5b1ec74ed
--- /dev/null
+++ b/common/src/main/java/org/apache/gravitino/dto/responses/BulkError.java
@@ -0,0 +1,80 @@
+/*
+ * 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.dto.responses;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import javax.annotation.Nullable;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.apache.commons.lang3.StringUtils;
+
+/** Represents an item-level error in a bulk response. */
+@Getter
+@ToString
+@EqualsAndHashCode
+public class BulkError {
+
+ @JsonProperty("index")
+ private final int index;
+
+ @Nullable
+ @JsonProperty("name")
+ private final String name;
+
+ @JsonProperty("code")
+ private final int code;
+
+ @JsonProperty("type")
+ private final String type;
+
+ @JsonProperty("message")
+ private final String message;
+
+ /**
+ * Creates a new BulkError.
+ *
+ * @param index The zero-based index of the failed request item.
+ * @param name The name of the failed request item.
+ * @param code The Gravitino error code.
+ * @param type The error type.
+ * @param message The error message.
+ */
+ public BulkError(int index, @Nullable String name, int code, String type,
String message) {
+ this.index = index;
+ this.name = name;
+ this.code = code;
+ this.type = type;
+ this.message = message;
+ }
+
+ /** Default constructor for BulkError. (Used for Jackson deserialization.) */
+ public BulkError() {
+ this(-1, null, 0, null, null);
+ }
+
+ /** Validates the bulk error. */
+ public void validate() {
+ Preconditions.checkArgument(index >= 0, "index must be >= 0");
+ Preconditions.checkArgument(code > 0, "code must be > 0");
+ Preconditions.checkArgument(StringUtils.isNotBlank(type), "type must not
be blank");
+ Preconditions.checkArgument(StringUtils.isNotBlank(message), "message must
not be blank");
+ }
+}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/responses/BulkRemoveResponse.java
b/common/src/main/java/org/apache/gravitino/dto/responses/BulkRemoveResponse.java
new file mode 100644
index 0000000000..b02b3fb8c2
--- /dev/null
+++
b/common/src/main/java/org/apache/gravitino/dto/responses/BulkRemoveResponse.java
@@ -0,0 +1,71 @@
+/*
+ * 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.dto.responses;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+
+/** Represents a bulk remove response. */
+@Getter
+@ToString
+@EqualsAndHashCode(callSuper = true)
+public class BulkRemoveResponse extends BaseResponse {
+
+ @JsonProperty("names")
+ private final String[] names;
+
+ @JsonProperty("errors")
+ private final BulkError[] errors;
+
+ @JsonProperty("summary")
+ private final BulkSummary summary;
+
+ /**
+ * Creates a new BulkRemoveResponse.
+ *
+ * @param names The successfully removed names.
+ * @param errors The item-level errors.
+ * @param summary The summary counts.
+ */
+ public BulkRemoveResponse(String[] names, BulkError[] errors, BulkSummary
summary) {
+ super(0);
+ this.names = names;
+ this.errors = errors;
+ this.summary = summary;
+ }
+
+ /** Default constructor for BulkRemoveResponse. (Used for Jackson
deserialization.) */
+ public BulkRemoveResponse() {
+ this(null, null, null);
+ }
+
+ @Override
+ public void validate() throws IllegalArgumentException {
+ super.validate();
+ Preconditions.checkArgument(names != null, "names must not be null");
+ Preconditions.checkArgument(errors != null, "errors must not be null");
+ Preconditions.checkArgument(summary != null, "summary must not be null");
+ Arrays.stream(errors).forEach(BulkError::validate);
+ summary.validate();
+ }
+}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/responses/BulkSummary.java
b/common/src/main/java/org/apache/gravitino/dto/responses/BulkSummary.java
new file mode 100644
index 0000000000..28c86a9118
--- /dev/null
+++ b/common/src/main/java/org/apache/gravitino/dto/responses/BulkSummary.java
@@ -0,0 +1,67 @@
+/*
+ * 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.dto.responses;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+
+/** Represents summary counts for a bulk response. */
+@Getter
+@ToString
+@EqualsAndHashCode
+public class BulkSummary {
+
+ @JsonProperty("total")
+ private final int total;
+
+ @JsonProperty("succeeded")
+ private final int succeeded;
+
+ @JsonProperty("failed")
+ private final int failed;
+
+ /**
+ * Creates a new BulkSummary.
+ *
+ * @param total The total number of request items.
+ * @param succeeded The number of succeeded request items.
+ * @param failed The number of failed request items.
+ */
+ public BulkSummary(int total, int succeeded, int failed) {
+ this.total = total;
+ this.succeeded = succeeded;
+ this.failed = failed;
+ }
+
+ /** Default constructor for BulkSummary. (Used for Jackson deserialization.)
*/
+ public BulkSummary() {
+ this(0, 0, 0);
+ }
+
+ /** Validates the bulk summary. */
+ public void validate() {
+ Preconditions.checkArgument(total >= 0, "total must be >= 0");
+ Preconditions.checkArgument(succeeded >= 0, "succeeded must be >= 0");
+ Preconditions.checkArgument(failed >= 0, "failed must be >= 0");
+ Preconditions.checkArgument(total == succeeded + failed, "total must equal
succeeded + failed");
+ }
+}
diff --git
a/common/src/main/java/org/apache/gravitino/dto/responses/BulkUserResponse.java
b/common/src/main/java/org/apache/gravitino/dto/responses/BulkUserResponse.java
new file mode 100644
index 0000000000..96deff2a76
--- /dev/null
+++
b/common/src/main/java/org/apache/gravitino/dto/responses/BulkUserResponse.java
@@ -0,0 +1,72 @@
+/*
+ * 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.dto.responses;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.base.Preconditions;
+import java.util.Arrays;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+import org.apache.gravitino.dto.authorization.UserDTO;
+
+/** Represents a bulk user response. */
+@Getter
+@ToString
+@EqualsAndHashCode(callSuper = true)
+public class BulkUserResponse extends BaseResponse {
+
+ @JsonProperty("users")
+ private final UserDTO[] users;
+
+ @JsonProperty("errors")
+ private final BulkError[] errors;
+
+ @JsonProperty("summary")
+ private final BulkSummary summary;
+
+ /**
+ * Creates a new BulkUserResponse.
+ *
+ * @param users The successfully added users.
+ * @param errors The item-level errors.
+ * @param summary The summary counts.
+ */
+ public BulkUserResponse(UserDTO[] users, BulkError[] errors, BulkSummary
summary) {
+ super(0);
+ this.users = users;
+ this.errors = errors;
+ this.summary = summary;
+ }
+
+ /** Default constructor for BulkUserResponse. (Used for Jackson
deserialization.) */
+ public BulkUserResponse() {
+ this(null, null, null);
+ }
+
+ @Override
+ public void validate() throws IllegalArgumentException {
+ super.validate();
+ Preconditions.checkArgument(users != null, "users must not be null");
+ Preconditions.checkArgument(errors != null, "errors must not be null");
+ Preconditions.checkArgument(summary != null, "summary must not be null");
+ Arrays.stream(errors).forEach(BulkError::validate);
+ summary.validate();
+ }
+}
diff --git a/conf/gravitino.conf.template b/conf/gravitino.conf.template
index 4d936dda34..c9afcc025c 100644
--- a/conf/gravitino.conf.template
+++ b/conf/gravitino.conf.template
@@ -21,6 +21,8 @@
gravitino.server.shutdown.timeout = 3000
# Timeout in milliseconds for the entity-store readiness probe used by
/api/health/ready
# gravitino.server.health.entityStore.probeTimeoutMs = 2000
+# Maximum number of items allowed in a single bulk request
+gravitino.server.bulk.maxItems = 100
# THE CONFIGURATION FOR Gravitino WEB SERVER
# The host name of the built-in web server
diff --git a/core/src/main/java/org/apache/gravitino/Configs.java
b/core/src/main/java/org/apache/gravitino/Configs.java
index 4ed70e1bec..ed47ef8b5a 100644
--- a/core/src/main/java/org/apache/gravitino/Configs.java
+++ b/core/src/main/java/org/apache/gravitino/Configs.java
@@ -97,6 +97,8 @@ public class Configs {
public static final int DEFAULT_GRAVITINO_AUTHORIZATION_THREAD_POOL_SIZE =
100;
+ public static final int DEFAULT_BULK_MAX_ITEMS = 100;
+
public static final long
DEFAULT_RELATIONAL_JDBC_BACKEND_MAX_WAIT_MILLISECONDS = 1000L;
public static final int GARBAGE_COLLECTOR_SINGLE_DELETION_LIMIT = 100;
@@ -346,6 +348,14 @@ public class Configs {
.intConf()
.createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_THREAD_POOL_SIZE);
+ public static final ConfigEntry<Integer> BULK_MAX_ITEMS =
+ new ConfigBuilder("gravitino.server.bulk.maxItems")
+ .doc("The maximum number of items allowed in a single bulk request")
+ .version(ConfigConstants.VERSION_2_0_0)
+ .intConf()
+ .checkValue(value -> value > 0,
ConfigConstants.POSITIVE_NUMBER_ERROR_MSG)
+ .createWithDefault(DEFAULT_BULK_MAX_ITEMS);
+
public static final long
DEFAULT_GRAVITINO_AUTHORIZATION_CACHE_EXPIRATION_SECS = 3600L;
public static final ConfigEntry<Long>
GRAVITINO_AUTHORIZATION_CACHE_EXPIRATION_SECS =
diff --git a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
index def00e8cbc..e0a671b844 100644
--- a/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
+++ b/core/src/main/java/org/apache/gravitino/GravitinoEnv.java
@@ -28,6 +28,7 @@ import org.apache.gravitino.authorization.OwnerDispatcher;
import org.apache.gravitino.authorization.OwnerEventManager;
import org.apache.gravitino.authorization.OwnerManager;
import org.apache.gravitino.auxiliary.AuxiliaryServiceManager;
+import org.apache.gravitino.bulk.BulkManager;
import org.apache.gravitino.catalog.CatalogDispatcher;
import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.catalog.CatalogNormalizeDispatcher;
@@ -185,6 +186,7 @@ public class GravitinoEnv {
private EventBus eventBus;
private OwnerDispatcher ownerDispatcher;
private OwnerDispatcher internalOwnerDispatcher;
+ private BulkManager bulkManager;
private FutureGrantManager futureGrantManager;
private GravitinoAuthorizer gravitinoAuthorizer;
private StatisticDispatcher statisticDispatcher;
@@ -526,6 +528,15 @@ public class GravitinoEnv {
return internalAccessControlDispatcher;
}
+ /**
+ * Get the BulkManager associated with the Gravitino environment.
+ *
+ * @return The BulkManager instance.
+ */
+ public BulkManager bulkManager() {
+ return bulkManager;
+ }
+
/**
* Get the tagDispatcher associated with the Gravitino environment.
*
@@ -858,12 +869,14 @@ public class GravitinoEnv {
OwnerDispatcher ownerManager = new OwnerManager(entityStore);
this.internalOwnerDispatcher = ownerManager;
this.ownerDispatcher = new OwnerEventManager(eventBus, ownerManager);
+ this.bulkManager = new BulkManager(config);
this.futureGrantManager = new FutureGrantManager(entityStore,
ownerManager);
} else {
this.accessControlDispatcher = null;
this.internalAccessControlDispatcher = null;
this.ownerDispatcher = null;
this.internalOwnerDispatcher = null;
+ this.bulkManager = null;
this.futureGrantManager = null;
}
diff --git
a/core/src/main/java/org/apache/gravitino/authorization/AccessControlDispatcher.java
b/core/src/main/java/org/apache/gravitino/authorization/AccessControlDispatcher.java
index 25b48b1067..e91af8c6e9 100644
---
a/core/src/main/java/org/apache/gravitino/authorization/AccessControlDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/authorization/AccessControlDispatcher.java
@@ -20,8 +20,11 @@ package org.apache.gravitino.authorization;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.UserAdd;
import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
import org.apache.gravitino.exceptions.IllegalRoleException;
import org.apache.gravitino.exceptions.NoSuchGroupException;
@@ -66,6 +69,18 @@ public interface AccessControlDispatcher {
User addUser(String metalake, String user, String externalId, boolean
enabled)
throws UserAlreadyExistsException, NoSuchMetalakeException;
+ /**
+ * Adds users in bulk.
+ *
+ * @param metalake The Metalake of the Users.
+ * @param users The Users to add.
+ * @return The item-level bulk results.
+ * @throws NoSuchMetalakeException If the Metalake with the given name does
not exist.
+ * @throws RuntimeException If adding the Users encounters storage issues.
+ */
+ List<BulkItemResult<User>> addUsers(String metalake, List<UserAdd> users)
+ throws NoSuchMetalakeException;
+
/**
* Removes a User.
*
@@ -78,6 +93,20 @@ public interface AccessControlDispatcher {
*/
boolean removeUser(String metalake, String user) throws
NoSuchMetalakeException;
+ /**
+ * Removes Users in bulk.
+ *
+ * @param metalake The Metalake of the Users.
+ * @param users The names of the Users.
+ * @param metalakeOwner The Metalake owner.
+ * @return The item-level bulk results.
+ * @throws NoSuchMetalakeException If the Metalake with the given name does
not exist.
+ * @throws RuntimeException If removing the Users encounters storage issues.
+ */
+ List<BulkItemResult<String>> removeUsers(
+ String metalake, List<String> users, Optional<Owner> metalakeOwner)
+ throws NoSuchMetalakeException;
+
/**
* Removes a User by external identifier.
*
diff --git
a/core/src/main/java/org/apache/gravitino/authorization/AccessControlManager.java
b/core/src/main/java/org/apache/gravitino/authorization/AccessControlManager.java
index 91dc0d1781..05b44dc4c2 100644
---
a/core/src/main/java/org/apache/gravitino/authorization/AccessControlManager.java
+++
b/core/src/main/java/org/apache/gravitino/authorization/AccessControlManager.java
@@ -18,14 +18,18 @@
*/
package org.apache.gravitino.authorization;
+import com.google.common.collect.Lists;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import org.apache.gravitino.Config;
import org.apache.gravitino.Configs;
import org.apache.gravitino.EntityStore;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.UserAdd;
import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
import org.apache.gravitino.exceptions.IllegalRoleException;
import org.apache.gravitino.exceptions.NoSuchGroupException;
@@ -80,6 +84,34 @@ public class AccessControlManager implements
AccessControlDispatcher {
() -> userGroupExternalManager.addUser(metalake, user, externalId,
enabled));
}
+ @Override
+ public List<BulkItemResult<User>> addUsers(String metalake, List<UserAdd>
users)
+ throws NoSuchMetalakeException {
+ return TreeLockUtils.doWithTreeLock(
+
NameIdentifier.of(AuthorizationUtils.ofUserNamespace(metalake).levels()),
+ LockType.WRITE,
+ () -> {
+ List<BulkItemResult<User>> results =
Lists.newArrayListWithCapacity(users.size());
+ for (int index = 0; index < users.size(); index++) {
+ UserAdd user = users.get(index);
+ try {
+ User addedUser =
+ user.hasExternalId()
+ ? userGroupExternalManager.addUser(
+ metalake,
+ user.name(),
+ user.externalId(),
+ Optional.ofNullable(user.enabled()).orElse(true))
+ : userGroupManager.addUser(metalake, user.name());
+ results.add(BulkItemResult.success(index, user.name(),
addedUser));
+ } catch (Exception e) {
+ results.add(BulkItemResult.failure(index, user.name(), e));
+ }
+ }
+ return results;
+ });
+ }
+
@Override
public boolean removeUser(String metalake, String user) throws
NoSuchMetalakeException {
return TreeLockUtils.doWithTreeLock(
@@ -88,6 +120,35 @@ public class AccessControlManager implements
AccessControlDispatcher {
() -> userGroupManager.removeUser(metalake, user));
}
+ @Override
+ public List<BulkItemResult<String>> removeUsers(
+ String metalake, List<String> users, Optional<Owner> metalakeOwner)
+ throws NoSuchMetalakeException {
+ return TreeLockUtils.doWithTreeLock(
+
NameIdentifier.of(AuthorizationUtils.ofUserNamespace(metalake).levels()),
+ LockType.WRITE,
+ () -> {
+ List<BulkItemResult<String>> results =
Lists.newArrayListWithCapacity(users.size());
+ for (int index = 0; index < users.size(); index++) {
+ String user = users.get(index);
+ try {
+ ensureNotMetalakeOwner(metalakeOwner, metalake, user);
+ boolean removed = userGroupManager.removeUser(metalake, user);
+ if (!removed) {
+ results.add(
+ BulkItemResult.failure(
+ index, user, new NoSuchUserException("User does not
exist: %s", user)));
+ continue;
+ }
+ results.add(BulkItemResult.success(index, user));
+ } catch (Exception e) {
+ results.add(BulkItemResult.failure(index, user, e));
+ }
+ }
+ return results;
+ });
+ }
+
@Override
public boolean removeUserByExternalId(String metalake, String externalId)
throws NoSuchMetalakeException {
@@ -402,4 +463,16 @@ public class AccessControlManager implements
AccessControlDispatcher {
() ->
permissionManager.overridePrivilegesInRole(metalake, role,
securableObjectsToOverride));
}
+
+ private void ensureNotMetalakeOwner(Optional<Owner> metalakeOwner, String
metalake, String user) {
+ metalakeOwner.ifPresent(
+ owner -> {
+ if (owner.type() == Owner.Type.USER && owner.name().equals(user)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot remove user %s from metalake %s because the user
is the owner of the metalake.",
+ user, metalake));
+ }
+ });
+ }
}
diff --git a/core/src/main/java/org/apache/gravitino/bulk/BulkItemResult.java
b/core/src/main/java/org/apache/gravitino/bulk/BulkItemResult.java
new file mode 100644
index 0000000000..074a8b88d0
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/bulk/BulkItemResult.java
@@ -0,0 +1,121 @@
+/*
+ * 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.bulk;
+
+import java.util.Optional;
+import javax.annotation.Nullable;
+
+/** Represents the result of one item in a best-effort bulk operation. */
+public final class BulkItemResult<T> {
+
+ private final int index;
+ private final String name;
+ @Nullable private final T value;
+ @Nullable private final Exception error;
+
+ private BulkItemResult(int index, String name, @Nullable T value, @Nullable
Exception error) {
+ this.index = index;
+ this.name = name;
+ this.value = value;
+ this.error = error;
+ }
+
+ /**
+ * Creates a successful item result with a value.
+ *
+ * @param index The item index in the request.
+ * @param name The item name.
+ * @param value The successful value.
+ * @return The successful item result.
+ * @param <T> The successful value type.
+ */
+ public static <T> BulkItemResult<T> success(int index, String name, T value)
{
+ return new BulkItemResult<>(index, name, value, null);
+ }
+
+ /**
+ * Creates a successful item result without a value.
+ *
+ * @param index The item index in the request.
+ * @param name The item name.
+ * @return The successful item result.
+ * @param <T> The successful value type.
+ */
+ public static <T> BulkItemResult<T> success(int index, String name) {
+ return new BulkItemResult<>(index, name, null, null);
+ }
+
+ /**
+ * Creates a failed item result.
+ *
+ * @param index The item index in the request.
+ * @param name The item name.
+ * @param error The item-level error.
+ * @return The failed item result.
+ * @param <T> The successful value type.
+ */
+ public static <T> BulkItemResult<T> failure(int index, String name,
Exception error) {
+ return new BulkItemResult<>(index, name, null, error);
+ }
+
+ /**
+ * Returns the item index in the request.
+ *
+ * @return The item index.
+ */
+ public int index() {
+ return index;
+ }
+
+ /**
+ * Returns the item name.
+ *
+ * @return The item name.
+ */
+ public String name() {
+ return name;
+ }
+
+ /**
+ * Returns whether the item succeeded.
+ *
+ * @return True if the item succeeded, otherwise false.
+ */
+ public boolean succeeded() {
+ return error == null;
+ }
+
+ /**
+ * Returns the successful value.
+ *
+ * @return The successful value.
+ */
+ public Optional<T> value() {
+ return Optional.ofNullable(value);
+ }
+
+ /**
+ * Returns the item-level error.
+ *
+ * @return The item-level error.
+ */
+ public Optional<Exception> error() {
+ return Optional.ofNullable(error);
+ }
+}
diff --git a/core/src/main/java/org/apache/gravitino/bulk/BulkManager.java
b/core/src/main/java/org/apache/gravitino/bulk/BulkManager.java
new file mode 100644
index 0000000000..1fccbc55c4
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/bulk/BulkManager.java
@@ -0,0 +1,95 @@
+/*
+ * 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.bulk;
+
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.dto.responses.BulkError;
+import org.apache.gravitino.dto.responses.ErrorConstants;
+import org.apache.gravitino.exceptions.AlreadyExistsException;
+import org.apache.gravitino.exceptions.ForbiddenException;
+import org.apache.gravitino.exceptions.NotFoundException;
+import org.apache.gravitino.exceptions.NotInUseException;
+
+/** Manages best-effort bulk operations. */
+public class BulkManager {
+
+ private final int maxBulkItems;
+
+ /**
+ * Creates a new {@link BulkManager}.
+ *
+ * @param config The Gravitino configuration.
+ */
+ public BulkManager(Config config) {
+ this.maxBulkItems =
+ config == null
+ ? Configs.BULK_MAX_ITEMS.getDefaultValue()
+ : config.get(Configs.BULK_MAX_ITEMS);
+ }
+
+ /**
+ * Checks whether the request item size exceeds the configured bulk limit.
+ *
+ * @param fieldName The request field name.
+ * @param size The request item size.
+ */
+ public void checkBulkSize(String fieldName, int size) {
+ if (size > maxBulkItems) {
+ throw new IllegalArgumentException(
+ String.format(
+ "\"%s\" size %d exceeds the maximum allowed bulk items %d",
+ fieldName, size, maxBulkItems));
+ }
+ }
+
+ private int errorCode(Exception e) {
+ if (e instanceof IllegalArgumentException) {
+ return ErrorConstants.ILLEGAL_ARGUMENTS_CODE;
+ } else if (e instanceof NotFoundException) {
+ return ErrorConstants.NOT_FOUND_CODE;
+ } else if (e instanceof AlreadyExistsException) {
+ return ErrorConstants.ALREADY_EXISTS_CODE;
+ } else if (e instanceof ForbiddenException) {
+ return ErrorConstants.FORBIDDEN_CODE;
+ } else if (e instanceof NotInUseException) {
+ return ErrorConstants.NOT_IN_USE_CODE;
+ }
+ return ErrorConstants.INTERNAL_ERROR_CODE;
+ }
+
+ /**
+ * Converts an item-level exception to a bulk error.
+ *
+ * @param result The failed item result.
+ * @return The bulk error.
+ */
+ public BulkError toBulkError(BulkItemResult<?> result) {
+ Exception error =
+ result
+ .error()
+ .orElseThrow(() -> new IllegalArgumentException("Bulk item result
has no error"));
+ return new BulkError(
+ result.index(),
+ result.name(),
+ errorCode(error),
+ error.getClass().getSimpleName(),
+ error.getMessage());
+ }
+}
diff --git a/core/src/main/java/org/apache/gravitino/bulk/UserAdd.java
b/core/src/main/java/org/apache/gravitino/bulk/UserAdd.java
new file mode 100644
index 0000000000..d62fd46a4c
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/bulk/UserAdd.java
@@ -0,0 +1,80 @@
+/*
+ * 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.bulk;
+
+import javax.annotation.Nullable;
+
+/** Represents one user to add in a bulk operation. */
+public final class UserAdd {
+
+ private final String name;
+ @Nullable private final String externalId;
+ @Nullable private final Boolean enabled;
+
+ /**
+ * Creates a user add item.
+ *
+ * @param name The user name.
+ * @param externalId The external identifier, or null if unset.
+ * @param enabled Whether the user is enabled, or null to use the default
value.
+ */
+ public UserAdd(String name, @Nullable String externalId, @Nullable Boolean
enabled) {
+ this.name = name;
+ this.externalId = externalId;
+ this.enabled = enabled;
+ }
+
+ /**
+ * Returns the user name.
+ *
+ * @return The user name.
+ */
+ public String name() {
+ return name;
+ }
+
+ /**
+ * Returns the external identifier.
+ *
+ * @return The external identifier, or null if unset.
+ */
+ @Nullable
+ public String externalId() {
+ return externalId;
+ }
+
+ /**
+ * Returns whether the user has an external identifier.
+ *
+ * @return True if the user has an external identifier, otherwise false.
+ */
+ public boolean hasExternalId() {
+ return externalId != null && !externalId.isEmpty();
+ }
+
+ /**
+ * Returns whether the user is enabled.
+ *
+ * @return Whether the user is enabled, or null to use the default value.
+ */
+ @Nullable
+ public Boolean enabled() {
+ return enabled;
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/hook/AccessControlHookDispatcher.java
b/core/src/main/java/org/apache/gravitino/hook/AccessControlHookDispatcher.java
index 3d856a61f8..570626eaee 100644
---
a/core/src/main/java/org/apache/gravitino/hook/AccessControlHookDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/hook/AccessControlHookDispatcher.java
@@ -20,6 +20,7 @@ package org.apache.gravitino.hook;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import org.apache.gravitino.Entity;
import org.apache.gravitino.GravitinoEnv;
@@ -37,6 +38,8 @@ import org.apache.gravitino.authorization.Role;
import org.apache.gravitino.authorization.SecurableObject;
import org.apache.gravitino.authorization.User;
import org.apache.gravitino.authorization.UserChange;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.UserAdd;
import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
import org.apache.gravitino.exceptions.IllegalRoleException;
import org.apache.gravitino.exceptions.NoSuchGroupException;
@@ -78,11 +81,24 @@ public class AccessControlHookDispatcher implements
AccessControlDispatcher {
return dispatcher.addUser(metalake, user, externalId, enabled);
}
+ @Override
+ public List<BulkItemResult<User>> addUsers(String metalake, List<UserAdd>
users)
+ throws NoSuchMetalakeException {
+ return dispatcher.addUsers(metalake, users);
+ }
+
@Override
public boolean removeUser(String metalake, String user) throws
NoSuchMetalakeException {
return dispatcher.removeUser(metalake, user);
}
+ @Override
+ public List<BulkItemResult<String>> removeUsers(
+ String metalake, List<String> users, Optional<Owner> metalakeOwner)
+ throws NoSuchMetalakeException {
+ return dispatcher.removeUsers(metalake, users, metalakeOwner);
+ }
+
@Override
public boolean removeUserByExternalId(String metalake, String externalId)
throws NoSuchMetalakeException {
diff --git
a/core/src/main/java/org/apache/gravitino/listener/AccessControlEventDispatcher.java
b/core/src/main/java/org/apache/gravitino/listener/AccessControlEventDispatcher.java
index a0e2517b1a..95e4aa7125 100644
---
a/core/src/main/java/org/apache/gravitino/listener/AccessControlEventDispatcher.java
+++
b/core/src/main/java/org/apache/gravitino/listener/AccessControlEventDispatcher.java
@@ -21,17 +21,21 @@ package org.apache.gravitino.listener;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.authorization.AccessControlDispatcher;
import org.apache.gravitino.authorization.Group;
import org.apache.gravitino.authorization.GroupChange;
+import org.apache.gravitino.authorization.Owner;
import org.apache.gravitino.authorization.PagedResult;
import org.apache.gravitino.authorization.Privilege;
import org.apache.gravitino.authorization.Role;
import org.apache.gravitino.authorization.SecurableObject;
import org.apache.gravitino.authorization.User;
import org.apache.gravitino.authorization.UserChange;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.UserAdd;
import org.apache.gravitino.exceptions.GroupAlreadyExistsException;
import org.apache.gravitino.exceptions.IllegalRoleException;
import org.apache.gravitino.exceptions.NoSuchGroupException;
@@ -208,6 +212,26 @@ public class AccessControlEventDispatcher implements
AccessControlDispatcher {
}
}
+ /** {@inheritDoc} */
+ @Override
+ public List<BulkItemResult<User>> addUsers(String metalake, List<UserAdd>
users)
+ throws NoSuchMetalakeException {
+ String initiator = PrincipalUtils.getCurrentUserName();
+ users.forEach(
+ user -> eventBus.dispatchEvent(new AddUserPreEvent(initiator,
metalake, user.name())));
+
+ try {
+ List<BulkItemResult<User>> results = dispatcher.addUsers(metalake,
users);
+ results.forEach(result -> dispatchAddUserResultEvent(initiator,
metalake, result));
+ return results;
+ } catch (Exception e) {
+ users.forEach(
+ user ->
+ eventBus.dispatchEvent(new AddUserFailureEvent(initiator,
metalake, e, user.name())));
+ throw e;
+ }
+ }
+
/** {@inheritDoc} */
@Override
public boolean removeUser(String metalake, String user) throws
NoSuchMetalakeException {
@@ -225,6 +249,26 @@ public class AccessControlEventDispatcher implements
AccessControlDispatcher {
}
}
+ /** {@inheritDoc} */
+ @Override
+ public List<BulkItemResult<String>> removeUsers(
+ String metalake, List<String> users, Optional<Owner> metalakeOwner)
+ throws NoSuchMetalakeException {
+ String initiator = PrincipalUtils.getCurrentUserName();
+ users.forEach(
+ user -> eventBus.dispatchEvent(new RemoveUserPreEvent(initiator,
metalake, user)));
+
+ try {
+ List<BulkItemResult<String>> results = dispatcher.removeUsers(metalake,
users, metalakeOwner);
+ results.forEach(result -> dispatchRemoveUserResultEvent(initiator,
metalake, result));
+ return results;
+ } catch (Exception e) {
+ users.forEach(
+ user -> eventBus.dispatchEvent(new RemoveUserFailureEvent(initiator,
metalake, e, user)));
+ throw e;
+ }
+ }
+
/** {@inheritDoc} */
@Override
public boolean removeUserByExternalId(String metalake, String externalId)
@@ -900,4 +944,25 @@ public class AccessControlEventDispatcher implements
AccessControlDispatcher {
throw e;
}
}
+
+ private void dispatchAddUserResultEvent(
+ String initiator, String metalake, BulkItemResult<User> result) {
+ if (result.succeeded()) {
+ eventBus.dispatchEvent(
+ new AddUserEvent(initiator, metalake, new
UserInfo(result.value().get())));
+ } else {
+ eventBus.dispatchEvent(
+ new AddUserFailureEvent(initiator, metalake, result.error().get(),
result.name()));
+ }
+ }
+
+ private void dispatchRemoveUserResultEvent(
+ String initiator, String metalake, BulkItemResult<String> result) {
+ if (result.succeeded()) {
+ eventBus.dispatchEvent(new RemoveUserEvent(initiator, metalake,
result.name(), true));
+ } else {
+ eventBus.dispatchEvent(
+ new RemoveUserFailureEvent(initiator, metalake,
result.error().get(), result.name()));
+ }
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
index 7b6686c6c1..d72cefa451 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestAccessControlManager.java
@@ -53,6 +53,7 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.UUID;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Catalog;
@@ -63,6 +64,8 @@ import org.apache.gravitino.EntityStoreFactory;
import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.Namespace;
import org.apache.gravitino.StringIdentifier;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.UserAdd;
import org.apache.gravitino.catalog.CatalogManager;
import org.apache.gravitino.connector.BaseCatalog;
import org.apache.gravitino.connector.authorization.AuthorizationPlugin;
@@ -246,6 +249,56 @@ public class TestAccessControlManager {
Assertions.assertFalse(removed1);
}
+ @Test
+ public void testBulkAddUsers() {
+ List<BulkItemResult<User>> results =
+ accessControlManager.addUsers(
+ METALAKE,
+ Lists.newArrayList(
+ new UserAdd("bulk_user_1", "bulk-user-ext-1", false),
+ new UserAdd("bulk_user_2", null, null),
+ new UserAdd("bulk_user_1", null, null)));
+
+ Assertions.assertEquals(3, results.size());
+ Assertions.assertTrue(results.get(0).succeeded());
+ Assertions.assertEquals("bulk_user_1",
results.get(0).value().get().name());
+ Assertions.assertEquals("bulk-user-ext-1",
results.get(0).value().get().externalId());
+ Assertions.assertFalse(results.get(0).value().get().enabled());
+ Assertions.assertTrue(results.get(1).succeeded());
+ Assertions.assertFalse(results.get(2).succeeded());
+ Assertions.assertTrue(results.get(2).error().get() instanceof
UserAlreadyExistsException);
+ }
+
+ @Test
+ public void testBulkRemoveUsers() {
+ accessControlManager.addUser(METALAKE, "bulk_remove_user");
+
+ List<BulkItemResult<String>> results =
+ accessControlManager.removeUsers(
+ METALAKE,
+ Lists.newArrayList("bulk_remove_user", "missing_bulk_user",
"metalake_owner"),
+ Optional.of(
+ new Owner() {
+ @Override
+ public String name() {
+ return "metalake_owner";
+ }
+
+ @Override
+ public Type type() {
+ return Type.USER;
+ }
+ }));
+
+ Assertions.assertEquals(3, results.size());
+ Assertions.assertTrue(results.get(0).succeeded());
+ Assertions.assertEquals("bulk_remove_user", results.get(0).name());
+ Assertions.assertFalse(results.get(1).succeeded());
+ Assertions.assertTrue(results.get(1).error().get() instanceof
NoSuchUserException);
+ Assertions.assertFalse(results.get(2).succeeded());
+ Assertions.assertTrue(results.get(2).error().get() instanceof
IllegalArgumentException);
+ }
+
@Test
public void testListUsers() {
accessControlManager.addUser("metalake_list", "testList1");
diff --git
a/core/src/test/java/org/apache/gravitino/listener/api/event/TestUserEvent.java
b/core/src/test/java/org/apache/gravitino/listener/api/event/TestUserEvent.java
index b38b0bc453..8845a36bf5 100644
---
a/core/src/test/java/org/apache/gravitino/listener/api/event/TestUserEvent.java
+++
b/core/src/test/java/org/apache/gravitino/listener/api/event/TestUserEvent.java
@@ -34,6 +34,8 @@ import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.authorization.PagedResult;
import org.apache.gravitino.authorization.User;
import org.apache.gravitino.authorization.UserChange;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.UserAdd;
import org.apache.gravitino.exceptions.GravitinoRuntimeException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
import org.apache.gravitino.exceptions.NoSuchUserException;
@@ -87,8 +89,6 @@ public class TestUserEvent {
this.dispatcher = new AccessControlEventDispatcher(eventBus,
mockUserDispatcher());
this.failureDispatcher =
new AccessControlEventDispatcher(eventBus,
mockExceptionUserDispatcher());
-
- System.out.println(failureDispatcher);
}
@Test
@@ -583,6 +583,37 @@ public class TestUserEvent {
Assertions.assertEquals(OperationType.ADD_USER, event.operationType());
}
+ @Test
+ void testBulkAddUsersDispatchesPerUserEvents() {
+ dummyEventListener.clear();
+
+ dispatcher.addUsers(
+ METALAKE,
+ Arrays.asList(
+ new UserAdd(userName, USER_EXT_ID, true), new
UserAdd(otherUserName, null, null)));
+
+ Assertions.assertEquals(2, dummyEventListener.getPreEvents().size());
+ PreEvent firstPreEvent = dummyEventListener.getPreEvents().get(0);
+ Assertions.assertEquals(AddUserPreEvent.class, firstPreEvent.getClass());
+ Assertions.assertEquals(identifier, firstPreEvent.identifier());
+ Assertions.assertEquals(userName, ((AddUserPreEvent)
firstPreEvent).userName());
+ PreEvent secondPreEvent = dummyEventListener.getPreEvents().get(1);
+ Assertions.assertEquals(AddUserPreEvent.class, secondPreEvent.getClass());
+ Assertions.assertEquals(otherIdentifier, secondPreEvent.identifier());
+ Assertions.assertEquals(otherUserName, ((AddUserPreEvent)
secondPreEvent).userName());
+
+ Assertions.assertEquals(2, dummyEventListener.getPostEvents().size());
+ Event firstEvent = dummyEventListener.getPostEvents().get(0);
+ Assertions.assertEquals(AddUserEvent.class, firstEvent.getClass());
+ Assertions.assertEquals(OperationStatus.SUCCESS,
firstEvent.operationStatus());
+ validateUserInfo(((AddUserEvent) firstEvent).addedUserInfo(), user);
+ Event secondEvent = dummyEventListener.getPostEvents().get(1);
+ Assertions.assertEquals(AddUserFailureEvent.class, secondEvent.getClass());
+ Assertions.assertEquals(OperationStatus.FAILURE,
secondEvent.operationStatus());
+ Assertions.assertEquals(otherIdentifier, secondEvent.identifier());
+ Assertions.assertEquals(otherUserName, ((AddUserFailureEvent)
secondEvent).userName());
+ }
+
@Test
void testGetUserByExternalIdEvent() {
dispatcher.getUserByExternalId(METALAKE, USER_EXT_ID);
@@ -610,6 +641,35 @@ public class TestUserEvent {
Assertions.assertEquals(OperationType.REMOVE_USER_BY_EXTERNAL_ID,
event.operationType());
}
+ @Test
+ void testBulkRemoveUsersDispatchesPerUserEvents() {
+ dummyEventListener.clear();
+
+ dispatcher.removeUsers(METALAKE, Arrays.asList(userName, inExistUserName),
Optional.empty());
+
+ Assertions.assertEquals(2, dummyEventListener.getPreEvents().size());
+ PreEvent firstPreEvent = dummyEventListener.getPreEvents().get(0);
+ Assertions.assertEquals(RemoveUserPreEvent.class,
firstPreEvent.getClass());
+ Assertions.assertEquals(identifier, firstPreEvent.identifier());
+ Assertions.assertEquals(userName, ((RemoveUserPreEvent)
firstPreEvent).userName());
+ PreEvent secondPreEvent = dummyEventListener.getPreEvents().get(1);
+ Assertions.assertEquals(RemoveUserPreEvent.class,
secondPreEvent.getClass());
+ Assertions.assertEquals(inExistIdentifier, secondPreEvent.identifier());
+ Assertions.assertEquals(inExistUserName, ((RemoveUserPreEvent)
secondPreEvent).userName());
+
+ Assertions.assertEquals(2, dummyEventListener.getPostEvents().size());
+ Event firstEvent = dummyEventListener.getPostEvents().get(0);
+ Assertions.assertEquals(RemoveUserEvent.class, firstEvent.getClass());
+ Assertions.assertEquals(OperationStatus.SUCCESS,
firstEvent.operationStatus());
+ Assertions.assertEquals(userName, ((RemoveUserEvent)
firstEvent).removedUserName());
+ Assertions.assertTrue(((RemoveUserEvent) firstEvent).isExists());
+ Event secondEvent = dummyEventListener.getPostEvents().get(1);
+ Assertions.assertEquals(RemoveUserFailureEvent.class,
secondEvent.getClass());
+ Assertions.assertEquals(OperationStatus.FAILURE,
secondEvent.operationStatus());
+ Assertions.assertEquals(inExistIdentifier, secondEvent.identifier());
+ Assertions.assertEquals(inExistUserName, ((RemoveUserFailureEvent)
secondEvent).userName());
+ }
+
@Test
void testGetUserByExternalIdFailureEvent() {
Assertions.assertThrowsExactly(
@@ -709,10 +769,22 @@ public class TestUserEvent {
when(dispatcher.addUser(METALAKE, userName)).thenReturn(user);
when(dispatcher.addUser(METALAKE, otherUserName)).thenReturn(otherUser);
when(dispatcher.addUser(METALAKE, userName, USER_EXT_ID,
true)).thenReturn(externalIdUser);
+ when(dispatcher.addUsers(eq(METALAKE), any()))
+ .thenReturn(
+ Arrays.asList(
+ BulkItemResult.success(0, userName, user),
+ BulkItemResult.failure(
+ 1, otherUserName, new GravitinoRuntimeException("Failed to
add user"))));
when(dispatcher.removeUser(METALAKE, userName)).thenReturn(true);
when(dispatcher.removeUser(METALAKE, inExistUserName)).thenReturn(false);
when(dispatcher.removeUserByExternalId(METALAKE,
USER_EXT_ID)).thenReturn(true);
+ when(dispatcher.removeUsers(eq(METALAKE), any(), any()))
+ .thenReturn(
+ Arrays.asList(
+ BulkItemResult.success(0, userName),
+ BulkItemResult.failure(
+ 1, inExistUserName, new NoSuchUserException("user not
found"))));
when(dispatcher.listUsers(METALAKE)).thenReturn(new User[] {user,
otherUser});
when(dispatcher.listUsers(eq(METALAKE), eq(0), eq(10)))
diff --git a/dev/docker/gravitino/rewrite_gravitino_server_config.py
b/dev/docker/gravitino/rewrite_gravitino_server_config.py
index 6e27611e5a..c5aca5d72c 100755
--- a/dev/docker/gravitino/rewrite_gravitino_server_config.py
+++ b/dev/docker/gravitino/rewrite_gravitino_server_config.py
@@ -28,6 +28,7 @@ env_map = {
"GRAVITINO_SERVER_WEBSERVER_THREAD_POOL_WORK_QUEUE_SIZE":
"server.webserver.threadPoolWorkQueueSize",
"GRAVITINO_SERVER_WEBSERVER_REQUEST_HEADER_SIZE":
"server.webserver.requestHeaderSize",
"GRAVITINO_SERVER_WEBSERVER_RESPONSE_HEADER_SIZE":
"server.webserver.responseHeaderSize",
+ "GRAVITINO_SERVER_BULK_MAX_ITEMS": "server.bulk.maxItems",
"GRAVITINO_ENTITY_STORE": "entity.store",
"GRAVITINO_ENTITY_STORE_RELATIONAL": "entity.store.relational",
"GRAVITINO_ENTITY_STORE_RELATIONAL_JDBC_URL":
"entity.store.relational.jdbcUrl",
@@ -89,6 +90,7 @@ init_config = {
"server.webserver.threadPoolWorkQueueSize": "100",
"server.webserver.requestHeaderSize": "131072",
"server.webserver.responseHeaderSize": "131072",
+ "server.bulk.maxItems": "100",
"entity.store": "relational",
"entity.store.relational": "JDBCBackend",
"entity.store.relational.jdbcUrl": "jdbc:h2",
@@ -142,4 +144,4 @@ if os.path.exists(config_file_path):
with open(config_file_path, "w") as file:
for key, value in config_map.items():
line = "{} = {}\n".format(key, value)
- file.write(line)
\ No newline at end of file
+ file.write(line)
diff --git a/docs/gravitino-server-config.md b/docs/gravitino-server-config.md
index 963774a51b..c8d5ba2f10 100644
--- a/docs/gravitino-server-config.md
+++ b/docs/gravitino-server-config.md
@@ -172,6 +172,7 @@ empty string or list; `(none)` means it has no default at
all.
| `gravitino.server.webserver.customFilters` | Comma-separated list
of servlet filter class names to apply to the API.
| (empty) |
| `gravitino.server.rest.extensionPackages` | Comma-separated list
of packages to scan for additional REST resources.
| (empty) |
| `gravitino.server.visibleConfigs` | Comma-separated list
of extra properties to expose on the unauthenticated `GET /configs` endpoint,
on top of the fixed set it always returns. Additive, so each entry widens what
is public. | (empty) |
+| `gravitino.server.bulk.maxItems` | Maximum number of
items allowed in a single bulk request.
| `100` |
Filters named in `customFilters` must be standard `javax.servlet` filters.
Pass parameters to a
filter with properties of the form
@@ -613,6 +614,7 @@ means the property is left alone.
| `GRAVITINO_SERVER_WEBSERVER_THREAD_POOL_WORK_QUEUE_SIZE` |
`gravitino.server.webserver.threadPoolWorkQueueSize` | `100`
|
| `GRAVITINO_SERVER_WEBSERVER_REQUEST_HEADER_SIZE` |
`gravitino.server.webserver.requestHeaderSize` | `131072`
|
| `GRAVITINO_SERVER_WEBSERVER_RESPONSE_HEADER_SIZE` |
`gravitino.server.webserver.responseHeaderSize` | `131072`
|
+| `GRAVITINO_SERVER_BULK_MAX_ITEMS` |
`gravitino.server.bulk.maxItems` | `100`
|
| `GRAVITINO_ENTITY_STORE` |
`gravitino.entity.store` | `relational`
|
| `GRAVITINO_ENTITY_STORE_RELATIONAL` |
`gravitino.entity.store.relational` | `JDBCBackend`
|
| `GRAVITINO_ENTITY_STORE_RELATIONAL_JDBC_URL` |
`gravitino.entity.store.relational.jdbcUrl` | `jdbc:h2`
|
diff --git a/docs/open-api/bulk.yaml b/docs/open-api/bulk.yaml
new file mode 100644
index 0000000000..7377d39d21
--- /dev/null
+++ b/docs/open-api/bulk.yaml
@@ -0,0 +1,275 @@
+# 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.
+
+---
+
+paths:
+
+ /bulk/metalakes/{metalake}/users/add:
+ parameters:
+ - $ref: "./openapi.yaml#/components/parameters/metalake"
+
+ post:
+ tags:
+ - access control
+ summary: Add users in bulk
+ operationId: bulkAddUsers
+ description: Adds users in best-effort mode. Failed items are returned
in the top-level errors array. The maximum request size is controlled by
`gravitino.server.bulk.maxItems`, which defaults to 100.
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: "#/components/schemas/BulkUserAddRequest"
+ examples:
+ BulkUserAddRequest:
+ $ref: "#/components/examples/BulkUserAddRequest"
+ responses:
+ "200":
+ description: Returns successfully added users and item-level errors
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "#/components/schemas/BulkUserResponse"
+ examples:
+ BulkUserResponse:
+ $ref: "#/components/examples/BulkUserResponse"
+ "400":
+ $ref: "./openapi.yaml#/components/responses/BadRequestErrorResponse"
+ "404":
+ description: Not Found - The specified metalake does not exist
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "./openapi.yaml#/components/schemas/ErrorModel"
+ examples:
+ NoSuchMetalakeException:
+ $ref:
"./metalakes.yaml#/components/examples/NoSuchMetalakeException"
+ "5xx":
+ $ref: "./openapi.yaml#/components/responses/ServerErrorResponse"
+
+ /bulk/metalakes/{metalake}/users/remove:
+ parameters:
+ - $ref: "./openapi.yaml#/components/parameters/metalake"
+
+ post:
+ tags:
+ - access control
+ summary: Remove users in bulk
+ operationId: bulkRemoveUsers
+ description: Removes users in best-effort mode. Failed items are
returned in the top-level errors array. User names in the same request must be
unique. The maximum request size is controlled by
`gravitino.server.bulk.maxItems`, which defaults to 100.
+ requestBody:
+ content:
+ application/json:
+ schema:
+ $ref: "#/components/schemas/BulkRemoveRequest"
+ examples:
+ BulkRemoveRequest:
+ $ref: "#/components/examples/BulkRemoveRequest"
+ responses:
+ "200":
+ description: Returns successfully removed names and item-level errors
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "#/components/schemas/BulkRemoveResponse"
+ examples:
+ BulkRemoveResponse:
+ $ref: "#/components/examples/BulkRemoveResponse"
+ "400":
+ $ref: "./openapi.yaml#/components/responses/BadRequestErrorResponse"
+ "404":
+ description: Not Found - The specified metalake does not exist
+ content:
+ application/vnd.gravitino.v1+json:
+ schema:
+ $ref: "./openapi.yaml#/components/schemas/ErrorModel"
+ examples:
+ NoSuchMetalakeException:
+ $ref:
"./metalakes.yaml#/components/examples/NoSuchMetalakeException"
+ "5xx":
+ $ref: "./openapi.yaml#/components/responses/ServerErrorResponse"
+
+components:
+
+ schemas:
+ BulkUserAddRequest:
+ type: object
+ required:
+ - users
+ properties:
+ users:
+ type: array
+ minItems: 1
+ description: The maximum length is controlled by
`gravitino.server.bulk.maxItems`, which defaults to 100.
+ items:
+ $ref: "./users.yaml#/components/schemas/UserAddRequest"
+
+ BulkRemoveRequest:
+ type: object
+ required:
+ - names
+ properties:
+ names:
+ type: array
+ minItems: 1
+ description: The maximum length is controlled by
`gravitino.server.bulk.maxItems`, which defaults to 100.
+ items:
+ type: string
+
+ BulkError:
+ type: object
+ required:
+ - index
+ - code
+ - type
+ - message
+ properties:
+ index:
+ type: integer
+ format: int32
+ description: The zero-based index of the failed request item
+ name:
+ type: string
+ description: The name of the failed request item
+ code:
+ type: integer
+ format: int32
+ description: Gravitino error code of the failed item
+ type:
+ type: string
+ description: Error type of the failed item
+ message:
+ type: string
+ description: Error message of the failed item
+
+ BulkSummary:
+ type: object
+ required:
+ - total
+ - succeeded
+ - failed
+ properties:
+ total:
+ type: integer
+ format: int32
+ succeeded:
+ type: integer
+ format: int32
+ failed:
+ type: integer
+ format: int32
+
+ BulkUserResponse:
+ type: object
+ properties:
+ code:
+ type: integer
+ format: int32
+ description: Status code of the response
+ enum:
+ - 0
+ users:
+ type: array
+ description: Successfully added users
+ items:
+ $ref: "./users.yaml#/components/schemas/User"
+ errors:
+ type: array
+ description: Item-level errors
+ items:
+ $ref: "#/components/schemas/BulkError"
+ summary:
+ $ref: "#/components/schemas/BulkSummary"
+
+ BulkRemoveResponse:
+ type: object
+ properties:
+ code:
+ type: integer
+ format: int32
+ description: Status code of the response
+ enum:
+ - 0
+ names:
+ type: array
+ description: Successfully removed names
+ items:
+ type: string
+ errors:
+ type: array
+ description: Item-level errors
+ items:
+ $ref: "#/components/schemas/BulkError"
+ summary:
+ $ref: "#/components/schemas/BulkSummary"
+
+ examples:
+ BulkUserAddRequest:
+ value: {
+ "users": [
+ {"name": "alice", "externalId": "ext-alice", "enabled": true},
+ {"name": "bob", "enabled": true}
+ ]
+ }
+
+ BulkRemoveRequest:
+ value: {
+ "names": ["alice", "bob", "ghost"]
+ }
+
+ BulkUserResponse:
+ value: {
+ "code": 0,
+ "users": [
+ {
+ "name": "alice",
+ "externalId": "ext-alice",
+ "enabled": true,
+ "roles": [],
+ "audit": {
+ "creator": "gravitino",
+ "createTime": "2026-07-28T10:00:00Z"
+ }
+ }
+ ],
+ "errors": [
+ {
+ "index": 1,
+ "name": "bob",
+ "code": 1004,
+ "type": "UserAlreadyExistsException",
+ "message": "User already exists: bob"
+ }
+ ],
+ "summary": {"total": 2, "succeeded": 1, "failed": 1}
+ }
+
+ BulkRemoveResponse:
+ value: {
+ "code": 0,
+ "names": ["alice", "bob"],
+ "errors": [
+ {
+ "index": 2,
+ "name": "ghost",
+ "code": 1003,
+ "type": "NoSuchUserException",
+ "message": "User does not exist: ghost"
+ }
+ ],
+ "summary": {"total": 3, "succeeded": 2, "failed": 1}
+ }
diff --git a/docs/open-api/openapi.yaml b/docs/open-api/openapi.yaml
index ac73f3fa0c..76d32c69fb 100644
--- a/docs/open-api/openapi.yaml
+++ b/docs/open-api/openapi.yaml
@@ -201,6 +201,12 @@ paths:
/metalakes/{metalake}/roles/{role}:
$ref: "./roles.yaml#/paths/~1metalakes~1%7Bmetalake%7D~1roles~1%7Brole%7D"
+ /bulk/metalakes/{metalake}/users/add:
+ $ref: "./bulk.yaml#/paths/~1bulk~1metalakes~1%7Bmetalake%7D~1users~1add"
+
+ /bulk/metalakes/{metalake}/users/remove:
+ $ref: "./bulk.yaml#/paths/~1bulk~1metalakes~1%7Bmetalake%7D~1users~1remove"
+
/metalakes/{metalake}/owners/{metadataObjectType}/{metadataObjectFullName}:
$ref:
"./owners.yaml#/paths/~1metalakes~1%7Bmetalake%7D~1owners~1%7BmetadataObjectType%7D~1%7BmetadataObjectFullName%7D"
diff --git a/docs/security/access-control.md b/docs/security/access-control.md
index 6883089813..969d8a359c 100755
--- a/docs/security/access-control.md
+++ b/docs/security/access-control.md
@@ -270,6 +270,42 @@ object: the owner of the table or view, plus
`CREATE_TABLE` or `CREATE_VIEW` on
| Job template | `REGISTER_JOB_TEMPLATE` | `USE_JOB_TEMPLATE`
| Owner | Run a job: `RUN_JOB` and `USE_JOB_TEMPLATE` |
| Job | | Owner
| Owner | |
+Bulk user access-control APIs use the same privileges as the matching
single-user operations. These
+bulk operations are authorized once before processing the request. Bulk user
requests report
+item-level failures in `errors`.
+
+| API | Required privilege
|
+|----------------------------------------------------|-------------------------------------------|
+| `POST /api/bulk/metalakes/{metalake}/users/add` | `OWNER` of the metalake
or `MANAGE_USERS` |
+| `POST /api/bulk/metalakes/{metalake}/users/remove` | `OWNER` of the metalake
or `MANAGE_USERS` |
+
+For example, add users in bulk:
+
+```shell
+curl -X POST "http://localhost:8090/api/bulk/metalakes/{metalake}/users/add"; \
+ -H "Authorization: Bearer $MANAGER_TOKEN" \
+ -H "Accept: application/vnd.gravitino.v1+json" \
+ -H "Content-Type: application/json" \
+ -d '{
+ "users": [
+ {"name": "analyst"},
+ {"name": "developer", "externalId": "[email protected]", "enabled":
true}
+ ]
+}'
+```
+
+Remove users in bulk:
+
+```shell
+curl -X POST
"http://localhost:8090/api/bulk/metalakes/{metalake}/users/remove"; \
+ -H "Authorization: Bearer $MANAGER_TOKEN" \
+ -H "Accept: application/vnd.gravitino.v1+json" \
+ -H "Content-Type: application/json" \
+ -d '{
+ "names": ["analyst", "developer"]
+}'
+```
+
Granting or revoking a privilege on an object takes `MANAGE_GRANTS` on that
object or an ancestor.
Granting or revoking a role, and overriding a role's privileges, takes
`MANAGE_GRANTS` on the
metalake. Setting an owner takes ownership.
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
index e3e057ffc4..c23818cc17 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
@@ -54,6 +54,7 @@ import
org.apache.gravitino.server.authorization.annotations.ExpressionCondition
import org.apache.gravitino.server.web.Utils;
import
org.apache.gravitino.server.web.filter.authorization.AuthorizationExecutor;
import
org.apache.gravitino.server.web.filter.authorization.AuthorizeExecutorFactory;
+import org.apache.gravitino.server.web.rest.BulkOperations;
import org.apache.gravitino.server.web.rest.CatalogOperations;
import org.apache.gravitino.server.web.rest.FilesetOperations;
import org.apache.gravitino.server.web.rest.FunctionOperations;
@@ -101,6 +102,7 @@ public class GravitinoInterceptionService implements
InterceptionService {
FunctionOperations.class.getName(),
TopicOperations.class.getName(),
FilesetOperations.class.getName(),
+ BulkOperations.class.getName(),
UserOperations.class.getName(),
GroupOperations.class.getName(),
PermissionOperations.class.getName(),
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/rest/BulkOperations.java
b/server/src/main/java/org/apache/gravitino/server/web/rest/BulkOperations.java
new file mode 100644
index 0000000000..0cd73dd5fe
--- /dev/null
+++
b/server/src/main/java/org/apache/gravitino/server/web/rest/BulkOperations.java
@@ -0,0 +1,181 @@
+/*
+ * 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.server.web.rest;
+
+import com.codahale.metrics.annotation.ResponseMetered;
+import com.codahale.metrics.annotation.Timed;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.Response;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.MetadataObject;
+import org.apache.gravitino.MetadataObjects;
+import org.apache.gravitino.authorization.AccessControlDispatcher;
+import org.apache.gravitino.authorization.Owner;
+import org.apache.gravitino.authorization.OwnerDispatcher;
+import org.apache.gravitino.authorization.User;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.BulkManager;
+import org.apache.gravitino.bulk.UserAdd;
+import org.apache.gravitino.dto.authorization.UserDTO;
+import org.apache.gravitino.dto.requests.BulkRemoveRequest;
+import org.apache.gravitino.dto.requests.BulkUserAddRequest;
+import org.apache.gravitino.dto.responses.BulkError;
+import org.apache.gravitino.dto.responses.BulkRemoveResponse;
+import org.apache.gravitino.dto.responses.BulkSummary;
+import org.apache.gravitino.dto.responses.BulkUserResponse;
+import org.apache.gravitino.dto.util.DTOConverters;
+import org.apache.gravitino.metalake.MetalakeManager;
+import org.apache.gravitino.metrics.MetricNames;
+import org.apache.gravitino.server.authorization.NameBindings;
+import
org.apache.gravitino.server.authorization.annotations.AuthorizationExpression;
+import
org.apache.gravitino.server.authorization.annotations.AuthorizationMetadata;
+import org.apache.gravitino.server.web.Utils;
+
+/** Provides best-effort bulk APIs for metalake access-control entities. */
[email protected]
+@Path("/bulk/metalakes/{metalake}")
+public class BulkOperations {
+
+ private static final String USERS_FIELD_NAME = "users";
+ private static final String NAMES_FIELD_NAME = "names";
+
+ private final BulkManager bulkManager;
+ private final AccessControlDispatcher accessControlDispatcher;
+ private final OwnerDispatcher ownerDispatcher;
+
+ @Context private HttpServletRequest httpRequest;
+
+ /** Creates a new bulk operations resource. */
+ public BulkOperations() {
+ this.bulkManager = GravitinoEnv.getInstance().bulkManager();
+ this.accessControlDispatcher =
GravitinoEnv.getInstance().accessControlDispatcher();
+ this.ownerDispatcher = GravitinoEnv.getInstance().ownerDispatcher();
+ }
+
+ /**
+ * Adds users in bulk.
+ *
+ * @param metalake The metalake name.
+ * @param request The bulk user add request.
+ * @return The bulk user response.
+ */
+ @POST
+ @Path("users/add")
+ @Produces("application/vnd.gravitino.v1+json")
+ @Timed(name = "bulk-add-user." + MetricNames.HTTP_PROCESS_DURATION, absolute
= true)
+ @ResponseMetered(name = "bulk-add-user", absolute = true)
+ @AuthorizationExpression(expression = "METALAKE::OWNER ||
METALAKE::MANAGE_USERS")
+ public Response addUsers(
+ @PathParam("metalake") @AuthorizationMetadata(type =
Entity.EntityType.METALAKE)
+ String metalake,
+ BulkUserAddRequest request) {
+ try {
+ return Utils.doAs(
+ httpRequest,
+ () -> {
+ request.validate();
+ bulkManager.checkBulkSize(USERS_FIELD_NAME,
request.getUsers().length);
+ MetalakeManager.checkMetalakeInUse(metalake);
+ List<BulkItemResult<User>> results =
+ accessControlDispatcher.addUsers(
+ metalake,
+ Arrays.stream(request.getUsers())
+ .map(
+ user ->
+ new UserAdd(
+ user.getName(), user.getExternalId(),
user.getEnabled()))
+ .collect(Collectors.toList()));
+ UserDTO[] users =
+ results.stream()
+ .filter(BulkItemResult::succeeded)
+ .map(result -> DTOConverters.toDTO(result.value().get()))
+ .toArray(UserDTO[]::new);
+ BulkError[] errors =
+ results.stream()
+ .filter(result -> !result.succeeded())
+ .map(bulkManager::toBulkError)
+ .toArray(BulkError[]::new);
+ return Utils.ok(
+ new BulkUserResponse(
+ users, errors, new BulkSummary(results.size(),
users.length, errors.length)));
+ });
+ } catch (Exception e) {
+ return ExceptionHandlers.handleUserException(OperationType.ADD, "",
metalake, e);
+ }
+ }
+
+ /**
+ * Removes users in bulk.
+ *
+ * @param metalake The metalake name.
+ * @param request The bulk remove request.
+ * @return The bulk remove response.
+ */
+ @POST
+ @Path("users/remove")
+ @Produces("application/vnd.gravitino.v1+json")
+ @Timed(name = "bulk-remove-user." + MetricNames.HTTP_PROCESS_DURATION,
absolute = true)
+ @ResponseMetered(name = "bulk-remove-user", absolute = true)
+ @AuthorizationExpression(expression = "METALAKE::OWNER ||
METALAKE::MANAGE_USERS")
+ public Response removeUsers(
+ @PathParam("metalake") @AuthorizationMetadata(type =
Entity.EntityType.METALAKE)
+ String metalake,
+ BulkRemoveRequest request) {
+ try {
+ return Utils.doAs(
+ httpRequest,
+ () -> {
+ request.validate();
+ bulkManager.checkBulkSize(NAMES_FIELD_NAME,
request.getNames().length);
+ MetalakeManager.checkMetalakeInUse(metalake);
+ Optional<Owner> metalakeOwner =
+ ownerDispatcher.getOwner(
+ metalake, MetadataObjects.of(null, metalake,
MetadataObject.Type.METALAKE));
+ List<BulkItemResult<String>> results =
+ accessControlDispatcher.removeUsers(
+ metalake, Arrays.asList(request.getNames()),
metalakeOwner);
+ String[] names =
+ results.stream()
+ .filter(BulkItemResult::succeeded)
+ .map(BulkItemResult::name)
+ .toArray(String[]::new);
+ BulkError[] errors =
+ results.stream()
+ .filter(result -> !result.succeeded())
+ .map(bulkManager::toBulkError)
+ .toArray(BulkError[]::new);
+ return Utils.ok(
+ new BulkRemoveResponse(
+ names, errors, new BulkSummary(results.size(),
names.length, errors.length)));
+ });
+ } catch (Exception e) {
+ return ExceptionHandlers.handleUserException(OperationType.REMOVE, "",
metalake, e);
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/rest/TestBulkOperations.java
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestBulkOperations.java
new file mode 100644
index 0000000000..b55f599758
--- /dev/null
+++
b/server/src/test/java/org/apache/gravitino/server/web/rest/TestBulkOperations.java
@@ -0,0 +1,253 @@
+/*
+ * 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.server.web.rest;
+
+import static org.apache.gravitino.Configs.TREE_LOCK_CLEAN_INTERVAL;
+import static org.apache.gravitino.Configs.TREE_LOCK_MAX_NODE_IN_MEMORY;
+import static org.apache.gravitino.Configs.TREE_LOCK_MIN_NODE_IN_MEMORY;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.RETURNS_DEFAULTS;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+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.MediaType;
+import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.EntityStore;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.authorization.AccessControlManager;
+import org.apache.gravitino.authorization.OwnerDispatcher;
+import org.apache.gravitino.authorization.User;
+import org.apache.gravitino.bulk.BulkItemResult;
+import org.apache.gravitino.bulk.BulkManager;
+import org.apache.gravitino.bulk.UserAdd;
+import org.apache.gravitino.config.ConfigEntry;
+import org.apache.gravitino.connector.PropertiesMetadata;
+import org.apache.gravitino.dto.requests.BulkRemoveRequest;
+import org.apache.gravitino.dto.requests.BulkUserAddRequest;
+import org.apache.gravitino.dto.requests.UserAddRequest;
+import org.apache.gravitino.dto.responses.BulkRemoveResponse;
+import org.apache.gravitino.dto.responses.BulkUserResponse;
+import org.apache.gravitino.dto.responses.ErrorConstants;
+import org.apache.gravitino.dto.responses.ErrorResponse;
+import org.apache.gravitino.exceptions.NoSuchUserException;
+import org.apache.gravitino.exceptions.UserAlreadyExistsException;
+import org.apache.gravitino.lock.LockManager;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.UserEntity;
+import org.apache.gravitino.rest.RESTUtils;
+import org.glassfish.hk2.utilities.binding.AbstractBinder;
+import org.glassfish.jersey.server.ResourceConfig;
+import org.glassfish.jersey.test.TestProperties;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+public class TestBulkOperations extends BaseOperationsTest {
+
+ private static final AccessControlManager manager =
mock(AccessControlManager.class);
+ private static final EntityStore entityStore = mock(EntityStore.class);
+ private static final OwnerDispatcher ownerDispatcher =
mock(OwnerDispatcher.class);
+ private static BulkOperations bulkOperations;
+
+ private static class MockServletRequestFactory extends
ServletRequestFactoryBase {
+ @Override
+ public HttpServletRequest get() {
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ when(request.getRemoteUser()).thenReturn(null);
+ return request;
+ }
+ }
+
+ @BeforeAll
+ public static void setup() throws IllegalAccessException {
+ Config config =
+ mock(
+ Config.class,
+ invocation -> {
+ if ("get".equals(invocation.getMethod().getName())
+ && invocation.getArguments().length == 1
+ && invocation.getArgument(0) instanceof ConfigEntry) {
+ ConfigEntry<?> entry = invocation.getArgument(0);
+ return entry.getDefaultValue();
+ }
+ return RETURNS_DEFAULTS.answer(invocation);
+ });
+ doReturn(100000L).when(config).get(TREE_LOCK_MAX_NODE_IN_MEMORY);
+ doReturn(1000L).when(config).get(TREE_LOCK_MIN_NODE_IN_MEMORY);
+ doReturn(36000L).when(config).get(TREE_LOCK_CLEAN_INTERVAL);
+ doReturn(2).when(config).get(org.apache.gravitino.Configs.BULK_MAX_ITEMS);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new
LockManager(config), true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(),
"accessControlDispatcher", manager, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "ownerDispatcher",
ownerDispatcher, true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "bulkManager", new
BulkManager(config), true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "entityStore",
entityStore, true);
+ bulkOperations = new BulkOperations();
+ }
+
+ @BeforeEach
+ public void resetMocks() throws IOException {
+ Mockito.reset(manager, entityStore, ownerDispatcher);
+ BaseMetalake metalake = mock(BaseMetalake.class);
+ PropertiesMetadata propertiesMetadata = mock(PropertiesMetadata.class);
+ when(propertiesMetadata.getOrDefault(any(), any())).thenReturn(true);
+ when(metalake.propertiesMetadata()).thenReturn(propertiesMetadata);
+ when(entityStore.get(any(), any(), any())).thenReturn(metalake);
+ when(ownerDispatcher.getOwner(any(), any())).thenReturn(Optional.empty());
+ }
+
+ @Override
+ protected Application configure() {
+ try {
+ forceSet(
+ TestProperties.CONTAINER_PORT,
String.valueOf(RESTUtils.findAvailablePort(2000, 3000)));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ ResourceConfig resourceConfig = new ResourceConfig();
+ resourceConfig.register(bulkOperations);
+ resourceConfig.register(
+ new AbstractBinder() {
+ @Override
+ protected void configure() {
+
bindFactory(MockServletRequestFactory.class).to(HttpServletRequest.class);
+ }
+ });
+
+ return resourceConfig;
+ }
+
+ @Test
+ public void testBulkAddUsersBestEffort() {
+ User user1 = buildUser("user1");
+ when(manager.addUsers(any(), any()))
+ .thenReturn(
+ Arrays.asList(
+ BulkItemResult.success(0, "user1", user1),
+ BulkItemResult.failure(
+ 1, "user2", new UserAlreadyExistsException("User already
exists: user2"))));
+
+ BulkUserAddRequest request =
+ new BulkUserAddRequest(
+ new UserAddRequest[] {
+ new UserAddRequest("user1", "ext-user1", false), new
UserAddRequest("user2")
+ });
+ Response response =
+ target("/bulk/metalakes/metalake1/users/add")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
response.getStatus());
+ BulkUserResponse bulkResponse =
response.readEntity(BulkUserResponse.class);
+ Assertions.assertEquals(1, bulkResponse.getUsers().length);
+ Assertions.assertEquals("user1", bulkResponse.getUsers()[0].name());
+ Assertions.assertEquals(1, bulkResponse.getErrors().length);
+ Assertions.assertEquals(1, bulkResponse.getErrors()[0].getIndex());
+ Assertions.assertEquals("user2", bulkResponse.getErrors()[0].getName());
+ Assertions.assertEquals(
+ ErrorConstants.ALREADY_EXISTS_CODE,
bulkResponse.getErrors()[0].getCode());
+ Assertions.assertEquals(2, bulkResponse.getSummary().getTotal());
+ Assertions.assertEquals(1, bulkResponse.getSummary().getSucceeded());
+ Assertions.assertEquals(1, bulkResponse.getSummary().getFailed());
+
+ ArgumentCaptor<List<UserAdd>> usersCaptor =
ArgumentCaptor.forClass(List.class);
+ Mockito.verify(manager).addUsers(eq("metalake1"), usersCaptor.capture());
+ Assertions.assertEquals("user1", usersCaptor.getValue().get(0).name());
+ Assertions.assertEquals("ext-user1",
usersCaptor.getValue().get(0).externalId());
+ Assertions.assertEquals(false, usersCaptor.getValue().get(0).enabled());
+ }
+
+ @Test
+ public void testBulkRemoveUsersBestEffort() {
+ when(manager.removeUsers(any(), any(), any()))
+ .thenReturn(
+ Arrays.asList(
+ BulkItemResult.success(0, "user1"),
+ BulkItemResult.failure(
+ 1, "ghost", new NoSuchUserException("User does not exist:
ghost"))));
+
+ BulkRemoveRequest request = new BulkRemoveRequest(new String[] {"user1",
"ghost"});
+ Response response =
+ target("/bulk/metalakes/metalake1/users/remove")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(Entity.entity(request, MediaType.APPLICATION_JSON_TYPE));
+
+ Assertions.assertEquals(Response.Status.OK.getStatusCode(),
response.getStatus());
+ BulkRemoveResponse bulkResponse =
response.readEntity(BulkRemoveResponse.class);
+ Assertions.assertArrayEquals(new String[] {"user1"},
bulkResponse.getNames());
+ Assertions.assertEquals(1, bulkResponse.getErrors().length);
+ Assertions.assertEquals("ghost", bulkResponse.getErrors()[0].getName());
+ Assertions.assertEquals(ErrorConstants.NOT_FOUND_CODE,
bulkResponse.getErrors()[0].getCode());
+ }
+
+ @Test
+ public void testBulkRejectsEmptyAndExceededRequest() {
+ Response emptyResponse =
+ target("/bulk/metalakes/metalake1/users/add")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(
+ Entity.entity(
+ new BulkUserAddRequest(new UserAddRequest[] {}),
+ MediaType.APPLICATION_JSON_TYPE));
+ Assertions.assertEquals(Response.Status.BAD_REQUEST.getStatusCode(),
emptyResponse.getStatus());
+
+ BulkRemoveRequest exceededRequest =
+ new BulkRemoveRequest(new String[] {"user1", "user2", "user3"});
+ Response exceededResponse =
+ target("/bulk/metalakes/metalake1/users/remove")
+ .request(MediaType.APPLICATION_JSON_TYPE)
+ .accept("application/vnd.gravitino.v1+json")
+ .post(Entity.entity(exceededRequest,
MediaType.APPLICATION_JSON_TYPE));
+ Assertions.assertEquals(
+ Response.Status.BAD_REQUEST.getStatusCode(),
exceededResponse.getStatus());
+ ErrorResponse errorResponse =
exceededResponse.readEntity(ErrorResponse.class);
+ Assertions.assertEquals(ErrorConstants.ILLEGAL_ARGUMENTS_CODE,
errorResponse.getCode());
+ }
+
+ private User buildUser(String user) {
+ return UserEntity.builder()
+ .withId(1L)
+ .withName(user)
+ .withRoleNames(Collections.emptyList())
+ .withAuditInfo(
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build())
+ .build();
+ }
+}
