[pulsar-admin] Add remove-subscription-types-enabled command for namespace

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.admin.v2;
 
 import static org.apache.pulsar.common.policies.data.PoliciesUtil.getBundles;
+import com.google.common.collect.Sets;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiParam;
@@ -1615,6 +1616,18 @@ public void setSubscriptionTypesEnabled(
         internalSetSubscriptionTypesEnabled(subscriptionTypesEnabled);
     }
 
+    @DELETE
+    @Path("/{tenant}/{namespace}/subscriptionTypesEnabled")
+    @ApiOperation(value = " Remove subscription types enabled on a namespace.")
+    @ApiResponses(value = {
+            @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Tenant or Namespace does not exist"),
+            @ApiResponse(code = 409, message = "Concurrent modification")})
+    public void removeSubscriptionTypesEnabled(@PathParam("tenant") String tenant,
+                                               @PathParam("namespace") String namespace) {
+            validateNamespaceName(tenant, namespace);
+            internalSetSubscriptionTypesEnabled(Sets.newHashSet());
+    }
 
     @GET
     @Path("/{tenant}/{namespace}/schemaValidationEnforced")

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java

@@ -1698,8 +1698,8 @@ public void testSubscriptionTypesEnabled() throws PulsarAdminException, PulsarCl
         }
 
         // clear all namespace subType enabled, add failover to broker.conf and sub with shared will fail
-        subscriptionTypes.clear();
-        admin.namespaces().setSubscriptionTypesEnabled(namespace, subscriptionTypes);
+        admin.namespaces().removeSubscriptionTypesEnabled(namespace);
+        assertEquals(admin.namespaces().getSubscriptionTypesEnabled(namespace), Sets.newHashSet());
         consumerBuilder.subscriptionType(SubscriptionType.Shared);
         HashSet<String> subscriptions = new HashSet<>();
         subscriptions.add("Failover");

pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Namespaces.java

@@ -1396,6 +1396,31 @@ CompletableFuture<Void> setSubscriptionTypesEnabledAsync(String namespace,
      */
     CompletableFuture<Set<SubscriptionType>> getSubscriptionTypesEnabledAsync(String namespace);
 
+    /**
+     * Removes the subscriptionTypesEnabled policy for a given namespace.
+     *
+     * @param namespace
+     *            Namespace name
+     *
+     * @throws NotAuthorizedException
+     *             Don't have admin permission
+     * @throws NotFoundException
+     *             Namespace does not exist
+     * @throws PulsarAdminException
+     *             Unexpected error
+     * @return
+     */
+    void removeSubscriptionTypesEnabled(String namespace) throws PulsarAdminException;
+
+    /**
+     * Removes the subscriptionTypesEnabled policy for a given namespace.
+     *
+     * @param namespace
+     *            Namespace name
+     * @return
+     */
+    CompletableFuture<Void> removeSubscriptionTypesEnabledAsync(String namespace);
+
     /**
      * Removes the autoSubscriptionCreation policy for a given namespace.
      * <p/>

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/NamespacesImpl.java

@@ -1098,6 +1098,27 @@ public void failed(Throwable throwable) {
         return future;
     }
 
+    @Override
+    public void removeSubscriptionTypesEnabled(String namespace) throws PulsarAdminException {
+        try {
+            removeSubscriptionTypesEnabledAsync(namespace).get(this.readTimeoutMs, TimeUnit.MILLISECONDS);
+        } catch (ExecutionException e) {
+            throw (PulsarAdminException) e.getCause();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new PulsarAdminException(e);
+        } catch (TimeoutException e) {
+            throw new PulsarAdminException.TimeoutException(e);
+        }
+    }
+
+    @Override
+    public CompletableFuture<Void> removeSubscriptionTypesEnabledAsync(String namespace) {
+        NamespaceName ns = NamespaceName.get(namespace);
+        WebTarget path = namespacePath(ns, "subscriptionTypesEnabled");
+        return asyncDeleteRequest(path);
+    }
+
     @Override
     public void removeAutoSubscriptionCreation(String namespace) throws PulsarAdminException {
         try {

pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java

@@ -368,6 +368,9 @@ public void namespaces() throws Exception {
         namespaces.run(split("get-subscription-types-enabled myprop/clust/ns1"));
         verify(mockNamespaces).getSubscriptionTypesEnabled("myprop/clust/ns1");
 
+        namespaces.run(split("remove-subscription-types-enabled myprop/clust/ns1"));
+        verify(mockNamespaces).removeSubscriptionTypesEnabled("myprop/clust/ns1");
+
         namespaces.run(split("get-schema-validation-enforce myprop/clust/ns1 -ap"));
         verify(mockNamespaces).getSchemaValidationEnforced("myprop/clust/ns1", true);
 

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdNamespaces.java

@@ -342,6 +342,18 @@ void run() throws PulsarAdminException {
         }
     }
 
+    @Parameters(commandDescription = "Remove subscription types enabled for a namespace")
+    private class RemoveSubscriptionTypesEnabled extends CliCommand {
+        @Parameter(description = "tenant/namespace", required = true)
+        private java.util.List<String> params;
+
+        @Override
+        void run() throws PulsarAdminException {
+            String namespace = validateNamespace(params);
+            getAdmin().namespaces().removeSubscriptionTypesEnabled(namespace);
+        }
+    }
+
     @Parameters(commandDescription = "Set Message TTL for a namespace")
     private class SetMessageTTL extends CliCommand {
         @Parameter(description = "tenant/namespace", required = true)
@@ -2336,6 +2348,7 @@ public CmdNamespaces(Supplier<PulsarAdmin> admin) {
 
         jcommander.addCommand("set-subscription-types-enabled", new SetSubscriptionTypesEnabled());
         jcommander.addCommand("get-subscription-types-enabled", new GetSubscriptionTypesEnabled());
+        jcommander.addCommand("remove-subscription-types-enabled", new RemoveSubscriptionTypesEnabled());
 
         jcommander.addCommand("get-backlog-quotas", new GetBacklogQuotaMap());
         jcommander.addCommand("set-backlog-quota", new SetBacklogQuota());