[fix][admin] Allow defaultNumPartitions for non-partitioned autoTopicCreation override

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

@@ -2566,6 +2566,54 @@ public void testAutoTopicCreationOverrideWithMaxNumPartitionsLimit() throws Exce
             assertTrue(ex.getCause() instanceof NotAcceptableException);
         }
     }
+
+    @Test
+    public void testAutoTopicCreationOverrideNonPartitionedDefaultNumPartitionsIgnored() throws Exception {
+        String tenantName = newUniqueName("prop-xyz2");
+        String namespaceName = tenantName + "/ns-" + System.currentTimeMillis();
+        TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
+        admin.tenants().createTenant(tenantName, tenantInfo);
+        admin.namespaces().createNamespace(namespaceName, Set.of("test"));
+
+        AutoTopicCreationOverride overridePolicy = AutoTopicCreationOverride.builder()
+                .allowAutoTopicCreation(true)
+                .topicType(TopicType.NON_PARTITIONED.toString())
+                .defaultNumPartitions(1)
+                .build();
+        admin.namespaces().setAutoTopicCreation(namespaceName, overridePolicy);
+
+        AutoTopicCreationOverride storedPolicy = admin.namespaces().getAutoTopicCreation(namespaceName);
+        assertTrue(storedPolicy.isAllowAutoTopicCreation());
+        assertEquals(storedPolicy.getTopicType(), TopicType.NON_PARTITIONED.toString());
+        assertEquals(storedPolicy.getDefaultNumPartitions(), Integer.valueOf(1));
+
+        String topicName = "persistent://" + namespaceName + "/auto-topic-" + UUID.randomUUID();
+        pulsarClient.newProducer().topic(topicName).create().close();
+
+        assertEquals(admin.topics().getPartitionedTopicMetadata(topicName).partitions, 0);
+    }
+
+    @Test
+    public void testAutoTopicCreationOverrideNonPartitionedDefaultNumPartitionsRejected() throws Exception {
+        String tenantName = newUniqueName("prop-xyz2");
+        String namespaceName = tenantName + "/ns-" + System.currentTimeMillis();
+        TenantInfoImpl tenantInfo = new TenantInfoImpl(Set.of("role1", "role2"), Set.of("test"));
+        admin.tenants().createTenant(tenantName, tenantInfo);
+        admin.namespaces().createNamespace(namespaceName, Set.of("test"));
+
+        AutoTopicCreationOverride overridePolicy = AutoTopicCreationOverride.builder()
+                .allowAutoTopicCreation(true)
+                .topicType(TopicType.NON_PARTITIONED.toString())
+                .defaultNumPartitions(5)
+                .build();
+        try {
+            admin.namespaces().setAutoTopicCreation(namespaceName, overridePolicy);
+            fail("Should have failed");
+        } catch (PulsarAdminException e) {
+            assertEquals(e.getStatusCode(), Status.PRECONDITION_FAILED.getStatusCode());
+        }
+    }
+
     @Test
     public void testMaxTopicsPerNamespace() throws Exception {
         restartClusterAfterTest();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SetReplicationClustersValidationTest.java

@@ -245,7 +245,7 @@ public void testSetReplicationClustersWithMismatchedDefaultNumPartitions() throw
             admin1.namespaces().setAutoTopicCreation(namespace, policy1);
             fail("Expected behaviour: Pulsar does not allow setting non-partitioned and a certain partition counts");
         } catch (Exception ex) {
-            assertTrue(ex.getMessage().contains("is not allowed to be set when the type is non-partition"));
+            assertTrue(ex.getMessage().contains("must be null, 0 or 1 when the type is non-partitioned"));
         }
     }
 

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/impl/AutoTopicCreationOverrideImpl.java

@@ -44,6 +44,7 @@ public static ValidateResult validateOverride(AutoTopicCreationOverride override
             if (!TopicType.isValidTopicType(override.getTopicType())) {
                 return ValidateResult.fail(String.format("Unknown topic type [%s]", override.getTopicType()));
             }
+
             if (TopicType.PARTITIONED.toString().equals(override.getTopicType())) {
                 if (override.getDefaultNumPartitions() == null) {
                     return ValidateResult.fail("[defaultNumPartitions] cannot be null when the type is partitioned.");
@@ -52,9 +53,10 @@ public static ValidateResult validateOverride(AutoTopicCreationOverride override
                     return ValidateResult.fail("[defaultNumPartitions] cannot be less than 1 for partition type.");
                 }
             } else if (TopicType.NON_PARTITIONED.toString().equals(override.getTopicType())) {
-                if (override.getDefaultNumPartitions() != null) {
-                    return ValidateResult.fail("[defaultNumPartitions] is not allowed to be"
-                            + " set when the type is non-partition.");
+                Integer p = override.getDefaultNumPartitions();
+                if (p != null && p != 0 && p != 1) {
+                    return ValidateResult.fail(
+                            "[defaultNumPartitions] must be null, 0 or 1 when the type is non-partitioned.");
                 }
             }
         }

pulsar-common/src/test/java/org/apache/pulsar/common/policies/data/AutoTopicCreationOverrideTest.java

@@ -73,12 +73,34 @@ public void testNumPartitionsNotSet() {
     }
 
     @Test
-    public void testNumPartitionsOnNonPartitioned() {
+    public void testNumPartitionsOnNonPartitionedZeroAllowed() {
+        AutoTopicCreationOverride override = AutoTopicCreationOverride.builder()
+                .allowAutoTopicCreation(true)
+                .topicType(TopicType.NON_PARTITIONED.toString())
+                .defaultNumPartitions(0)
+                .build();
+        assertTrue(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
+    }
+
+    @Test
+    public void testNumPartitionsOnNonPartitionedOneAllowed() {
+        AutoTopicCreationOverride override = AutoTopicCreationOverride.builder()
+                .allowAutoTopicCreation(true)
+                .topicType(TopicType.NON_PARTITIONED.toString())
+                .defaultNumPartitions(1)
+                .build();
+        assertTrue(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
+    }
+
+
+    @Test
+    public void testNumPartitionsOnNonPartitionedTooHighRejected() {
         AutoTopicCreationOverride override = AutoTopicCreationOverride.builder()
                 .allowAutoTopicCreation(true)
                 .topicType(TopicType.NON_PARTITIONED.toString())
                 .defaultNumPartitions(2)
                 .build();
         assertFalse(AutoTopicCreationOverrideImpl.validateOverride(override).isSuccess());
     }
+
 }