[improve][broker] Add dedicated partition configuration for system topics to reduce resource usage

conf/broker.conf

@@ -204,6 +204,9 @@ allowAutoSubscriptionCreation=true
 # The number of partitioned topics that is allowed to be automatically created if allowAutoTopicCreationType is partitioned.
 defaultNumPartitions=1
 
+# Default number of partitions for the system topics
+systemTopicDefaultNumPartitions=1
+
 # Enable the deletion of inactive topics. This parameter need to cooperate with the allowAutoTopicCreation parameter.
 # If brokerDeleteInactiveTopicsEnabled is set to true, we should ensure that allowAutoTopicCreation is also set to true.
 brokerDeleteInactiveTopicsEnabled=true

conf/standalone.conf

@@ -1187,6 +1187,9 @@ allowAutoSubscriptionCreation=true
 # The number of partitioned topics that is allowed to be automatically created if allowAutoTopicCreationType is partitioned.
 defaultNumPartitions=1
 
+# Default number of partitions for the system topics
+systemTopicDefaultNumPartitions=1
+
 ### --- Transaction config variables --- ###
 # Enable transaction coordinator in broker
 transactionCoordinatorEnabled=false

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -2167,6 +2167,12 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
                     + " if allowAutoTopicCreationType is partitioned."
     )
     private int defaultNumPartitions = 1;
+    @FieldContext(
+            category = CATEGORY_STORAGE_ML,
+            dynamic = true,
+            doc = "Default number of partitions for the system topics."
+    )
+    private int systemTopicDefaultNumPartitions = 1;
     @FieldContext(
         category = CATEGORY_STORAGE_ML,
         doc = "The class of the managed ledger storage"

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -3291,12 +3291,14 @@ private CompletableFuture<PartitionedTopicMetadata> createDefaultPartitionedTopi
                                                                                         Optional<Policies> policies) {
         final int defaultNumPartitions = pulsar.getBrokerService().getDefaultNumPartitions(topicName, policies);
         final int maxPartitions = pulsar().getConfig().getMaxNumPartitionsPerPartitionedTopic();
+        final int systemTopicDefaultNumPartitions = pulsar.getConfiguration().getSystemTopicDefaultNumPartitions();
         checkArgument(defaultNumPartitions > 0,
                 "Default number of partitions should be more than 0");
         checkArgument(maxPartitions <= 0 || defaultNumPartitions <= maxPartitions,
                 "Number of partitions should be less than or equal to " + maxPartitions);
 
-        PartitionedTopicMetadata configMetadata = new PartitionedTopicMetadata(defaultNumPartitions);
+        PartitionedTopicMetadata configMetadata = new PartitionedTopicMetadata(
+                isSystemTopic(topicName) ? systemTopicDefaultNumPartitions : defaultNumPartitions);
 
         return checkMaxTopicsPerNamespace(topicName, defaultNumPartitions)
                 .thenCompose(__ -> {

pulsar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java

@@ -80,6 +80,7 @@ protected void setup() throws Exception {
         conf.setAllowAutoTopicCreation(false);
         conf.setAllowAutoTopicCreationType(TopicType.PARTITIONED);
         conf.setDefaultNumPartitions(PARTITIONS);
+        conf.setSystemTopicDefaultNumPartitions(PARTITIONS);
         conf.setManagedLedgerMaxEntriesPerLedger(1);
         conf.setBrokerDeleteInactiveTopicsEnabled(false);
         conf.setTransactionCoordinatorEnabled(true);