[fix][broker] Fix tenant creation and update with null value (apache#24209)

Signed-off-by: Zixuan Liu <nodeces@gmail.com>
(cherry picked from commit c59412e)
(cherry picked from commit b4ba347)

Author: nodece

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

@@ -273,15 +273,25 @@ protected CompletableFuture<Void> internalDeleteTenantAsyncForcefully(String ten
     }
 
     private CompletableFuture<Void> validateClustersAsync(TenantInfo info) {
-        // empty cluster shouldn't be allowed
-        if (info == null || info.getAllowedClusters().stream().filter(c -> !StringUtils.isBlank(c))
-                .collect(Collectors.toSet()).isEmpty()
-                || info.getAllowedClusters().stream().anyMatch(ac -> StringUtils.isBlank(ac))) {
-            log.warn("[{}] Failed to validate due to clusters are empty", clientAppId());
-            return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, "Clusters can not be empty"));
+        if (info == null) {
+            return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, "TenantInfo is null"));
         }
+
+        Set<String> allowedClusters = info.getAllowedClusters();
+        if (allowedClusters == null) {
+            return FutureUtil.failedFuture(new RestException(Status.PRECONDITION_FAILED, "Clusters cannot be null"));
+        }
+
+        Set<String> cleanedClusters = allowedClusters.stream()
+                .filter(c -> !StringUtils.isBlank(c))
+                .collect(Collectors.toSet());
+        if (cleanedClusters.isEmpty() || allowedClusters.stream().anyMatch(StringUtils::isBlank)) {
+            log.warn("[{}] Validation failed: allowed clusters are empty or contain blanks", clientAppId());
+            return FutureUtil.failedFuture(
+                    new RestException(Status.PRECONDITION_FAILED, "Clusters cannot be empty or blank"));
+        }
+
         return clusterResources().listAsync().thenAccept(availableClusters -> {
-            Set<String> allowedClusters = info.getAllowedClusters();
             List<String> nonexistentClusters = allowedClusters.stream()
                     .filter(cluster -> !(availableClusters.contains(cluster) || GLOBAL_CLUSTER.equals(cluster)))
                     .collect(Collectors.toList());

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

@@ -29,6 +29,7 @@
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -73,4 +74,49 @@ public void testDeleteNonExistTenant() {
         String tenant = "test-non-exist-tenant-" + UUID.randomUUID();
         assertThrows(PulsarAdminException.NotFoundException.class, () -> admin.tenants().deleteTenant(tenant));
     }
+
+    @Test
+    public void testCreateTenantWithNull() {
+        String tenant = "test-create-tenant-with-null-value-" + UUID.randomUUID();
+        // Put doesn't allow null value
+        assertThrows(PulsarAdminException.class,
+                () -> admin.tenants().createTenant(tenant, null));
+    }
+
+    @Test
+    public void testCreateTenantWithInvalidCluster() {
+        String tenant = "test-create-tenant-with-invalid-cluster-" + UUID.randomUUID();
+        // clusters is empty
+        assertThrows(PulsarAdminException.PreconditionFailedException.class,
+                () -> admin.tenants().createTenant(tenant, TenantInfo.builder().build()));
+
+        // clusters is null
+        assertThrows(PulsarAdminException.PreconditionFailedException.class,
+                () -> {
+                    TenantInfoImpl tenantInfo = new TenantInfoImpl();
+                    tenantInfo.setAdminRoles(null);
+                    tenantInfo.setAllowedClusters(null);
+                    admin.tenants().createTenant(tenant, tenantInfo);
+                });
+    }
+
+    @Test
+    public void testUpdateTenantWithInvalidCluster() throws PulsarAdminException {
+        String tenant = "test-update-tenant-with-invalid-cluster-" + UUID.randomUUID();
+        admin.tenants().createTenant(tenant,
+                TenantInfo.builder().allowedClusters(Collections.singleton(CLUSTER)).build());
+
+        // clusters is empty
+        assertThrows(PulsarAdminException.PreconditionFailedException.class,
+                () -> admin.tenants().updateTenant(tenant, TenantInfo.builder().build()));
+
+        // clusters is null
+        assertThrows(PulsarAdminException.PreconditionFailedException.class,
+                () -> {
+                    TenantInfoImpl tenantInfo = new TenantInfoImpl();
+                    tenantInfo.setAdminRoles(null);
+                    tenantInfo.setAllowedClusters(null);
+                    admin.tenants().updateTenant(tenant, tenantInfo);
+                });
+    }
 }