[fix][broker] Modify authorization to update topic's properties

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java

@@ -554,6 +554,7 @@ public CompletableFuture<Boolean> allowTopicOperationAsync(TopicName topicName,
                             case OFFLOAD:
                             case UNLOAD:
                             case DELETE_METADATA:
+                            case UPDATE_METADATA:
                             case ADD_BUNDLE_RANGE:
                             case GET_BUNDLE_RANGE:
                             case DELETE_BUNDLE_RANGE:

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -619,7 +619,7 @@ protected CompletableFuture<Void> internalUpdatePropertiesAsync(boolean authorit
             return CompletableFuture.completedFuture(null);
         }
         return validateTopicOwnershipAsync(topicName, authoritative)
-            .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.PRODUCE))
+            .thenCompose(__ -> validateTopicOperationAsync(topicName, TopicOperation.UPDATE_METADATA))
             .thenCompose(__ -> {
                 if (topicName.isPartitioned()) {
                     return internalUpdateNonPartitionedTopicProperties(properties);

pulsar-broker/src/test/java/org/apache/pulsar/client/api/AuthorizationProducerConsumerTest.java

@@ -30,6 +30,7 @@
 import com.google.common.collect.Sets;
 import java.io.IOException;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -548,6 +549,63 @@ public void testSchemaCompatibilityStrategyPermission() throws Exception {
         log.info("-- Exiting {} test --", methodName);
     }
 
+    @Test
+    public void testUpdateTopicPropertiesAuthorization() throws Exception {
+        log.info("-- Starting {} test --", methodName);
+        cleanup();
+        conf.setAuthorizationProvider(PulsarAuthorizationProvider.class.getName());
+        setup();
+
+        final String tenantRole = "tenant-role";
+        final String generalRole = "general-role";
+        final String namespace = "my-property/my-ns-sub-auth";
+        final String topicName = "persistent://" + namespace + "/my-topic";
+        clientAuthProviderSupportedRoles.add(generalRole);
+
+        Authentication superUserAuthentication = new ClientAuthentication("superUser");
+        @Cleanup
+        PulsarAdmin superAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+                .authentication(superUserAuthentication).build());
+
+        Authentication tenantAdminAuthentication = new ClientAuthentication(tenantRole);
+        @Cleanup
+        PulsarAdmin tenantAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+                .authentication(tenantAdminAuthentication).build());
+
+        Authentication generalAuthentication = new ClientAuthentication(generalRole);
+        @Cleanup
+        PulsarAdmin generalAdmin = spy(PulsarAdmin.builder().serviceHttpUrl(brokerUrl.toString())
+                .authentication(generalAuthentication).build());
+
+        superAdmin.clusters().createCluster("test",
+                ClusterData.builder().serviceUrl(brokerUrl.toString()).build());
+        superAdmin.tenants().createTenant("my-property",
+                new TenantInfoImpl(Sets.newHashSet(tenantRole), Sets.newHashSet("test")));
+        superAdmin.namespaces().createNamespace(namespace, Sets.newHashSet("test"));
+        superAdmin.topics().createPartitionedTopic(topicName, 1);
+
+        Map<String, String> topicProperties = new HashMap();
+        topicProperties.put("key1", "value1");
+        // superUser and tenantAdminatrator have authorization to operate topic's properties
+        superAdmin.topics().updateProperties(topicName, topicProperties);
+        Awaitility.await().untilAsserted(() -> assertEquals(
+                superAdmin.topics().getProperties(topicName).get("key1"), "value1"));
+        topicProperties.put("key1", "value2");
+        tenantAdmin.topics().updateProperties(topicName, topicProperties);
+        Awaitility.await().untilAsserted(() -> assertEquals(
+                tenantAdmin.topics().getProperties(topicName).get("key1"), "value2"));
+
+        // generalRole doesn't have authorization to update topic's properties so it will fail
+        try {
+            generalAdmin.topics().updateProperties(topicName, topicProperties);
+            fail("should have failed with authorization exception");
+        } catch (Exception e) {
+            assertTrue(e.getMessage().startsWith(
+                    "Unauthorized to validateTopicOperation for operation [UPDATE_METADATA]"));
+        }
+
+        log.info("-- Exiting {} test --", methodName);
+    }
     @Test
     public void testSubscriptionPrefixAuthorization() throws Exception {
         log.info("-- Starting {} test --", methodName);

pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/TopicOperation.java

@@ -50,6 +50,7 @@ public enum TopicOperation {
     GET_STATS,
     GET_METADATA,
     DELETE_METADATA,
+    UPDATE_METADATA,
     GET_BACKLOG_SIZE,
 
     SET_REPLICATED_SUBSCRIPTION_STATUS,