[improve][PIP] PIP-383: Support granting/revoking permissions for multiple topics (#23355)

Author: Technoboy-

pip/pip-383.md

@@ -0,0 +1,144 @@
+# PIP-383: Support granting/revoking permissions for multiple topics
+
+## Background
+
+In AuthorizationProvider, the authorization interface `grantPermissionAsync(TopicName topicName, Set<AuthAction> actions, String role, String authDataJson)` currently only supports granting permissions to a single topic at a time.
+When multiple topics need to be authorized under a namespace, the client makes the calls to the authorization interface concurrently. 
+Since the permissions information is stored in the namespace-level policies, and multiple topics may be on different brokers, concurrent authorization modification will cause concurrent modification exceptions. 
+Therefore, supporting granting permissions for multiple topics is very beneficial.
+
+
+## Motivation
+
+Supporting granting/revoking permissions for multiple topics, 
+add `grantPermissionAsync(List<GrantTopicPermissionOptions> options)` and `revokePermissionAsync(List<RevokeTopicPermissionOptions> options)` in AuthorizationProvider.
+
+## Goals
+
+### In Scope
+
+- Add `grantPermissionAsync(List<GrantTopicPermissionOptions> options)` in AuthorizationProvider.
+- Add `revokePermissionAsync(List<GrantTopicPermissionOptions> options)` in AuthorizationProvider.
+
+## High-Level Design
+
+### Design & Implementation Details
+
+Add default method implementation in AuthorizationProvider
+```java
+
+public interface AuthorizationProvider extends Closeable {
+
+    default CompletableFuture<Void> grantPermissionAsync(List<GrantTopicPermissionOptions> options) {
+        return FutureUtil.failedFuture(new IllegalStateException(
+                String.format("grantPermissionAsync is not supported by the Authorization")));
+    }
+
+    default CompletableFuture<Void> revokePermissionAsync(List<RevokeTopicPermissionOptions> options) {
+        return FutureUtil.failedFuture(new IllegalStateException(
+                String.format("revokePermissionAsync is not supported by the Authorization")));
+    }
+}
+```
+
+```
+@Data
+@Builder
+public class GrantTopicPermissionOptions {
+
+    private final String topic;
+    
+    private final String role;
+
+    private final Set<AuthAction> actions;
+}
+
+@Data
+@Builder
+public class RevokeTopicPermissionOptions {
+
+    private final String topic;
+
+    private final String role;
+}
+```
+
+Add namespace admin API.
+
+```java
+public interface Namespaces {
+    
+    CompletableFuture<Void> grantPermissionOnTopicsAsync(List<GrantTopicPermissionOptions> options);
+
+    void grantPermissionOnTopics(List<GrantTopicPermissionOptions> options) throws PulsarAdminException;
+
+    CompletableFuture<Void> revokePermissionOnTopicsAsync(List<RevokeTopicPermissionOptions> options);
+
+    void revokePermissionOnTopics(List<RevokeTopicPermissionOptions> options) throws PulsarAdminException;
+}
+```
+
+Add namespace rest implementation in broker side.
+```java
+@POST
+@Path("/grantPermissions")
+public void grantPermissionOnTopics(@Suspended final AsyncResponse asyncResponse,
+                             List<GrantTopicPermissionOptions> options) {
+    internalGrantPermissionsAsync(options)
+            .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+            .exceptionally(ex -> {
+                log.error("[{}] Failed to grant permissions {}",
+                        clientAppId(), options, ex);
+                resumeAsyncResponseExceptionally(asyncResponse, ex);
+                return null;
+            });
+}
+
+@POST
+@Path("/revokePermissions")
+public void revokePermissionOnTopics(@Suspended final AsyncResponse asyncResponse,
+                             List<RevokeTopicPermissionOptions> options) {
+    internalRevokePermissionsAsync(options)
+            .thenAccept(__ -> asyncResponse.resume(Response.noContent().build()))
+            .exceptionally(ex -> {
+                log.error("[{}] Failed to revoke permissions {}",
+                        clientAppId(), options, ex);
+                resumeAsyncResponseExceptionally(asyncResponse, ex);
+                return null;
+            });
+}
+```
+
+so user can grant/revoke permissions to multi-topics like :
+```java
+public class TestAuthorization {
+    
+    @Test
+    public void testGrantPermission() {
+        // grant permission for multi-topics
+        List<GrantPermissionOptions> grantPermissions = new ArrayList<>();
+        grantPermissions.add(GrantPermissionOptions.builder().topic("topic1").role("role1").actions(Set.of(AuthAction.produce)).build());
+        grantPermissions.add(GrantPermissionOptions.builder().topic("topic2").role("role2").actions(Set.of(AuthAction.consume)).build());
+        admin.namespaces().grantPermissionOnTopics(grantPermissions);
+        // revoke permission topics
+        List<RevokePermissionOptions> revokePermissions = new ArrayList<>();
+        revokePermissions.add(RevokePermissionOptions.builder().topic("topic1").role("role1").build());
+        revokePermissions.add(RevokePermissionOptions.builder().topic("topic2").role("role2").build());
+        admin.namespaces().revokePermissionOnTopics(revokePermissions);
+    }
+}
+
+```
+
+## Backward & Forward Compatibility
+
+
+
+## Alternatives
+
+## General Notes
+
+## Links
+
+* Mailing List discussion thread:  https://lists.apache.org/thread/6n2jdl9bsf1f6xz2orygz3kvxmy11ykh
+* Mailing List voting thread: https://lists.apache.org/thread/qbyvs75r0d64h6jk8w1swr782l85b77h