[improve][broker]Optimize the partition size of the system theme to r…

…educe resource consumption

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 theme
+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
@@ -243,7 +246,7 @@ messageExpiryCheckIntervalInMinutes=5
 # How long to delay rewinding cursor and dispatching messages when active consumer is changed
 activeConsumerFailoverDelayTimeMillis=1000
 
-# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. 
+# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type.
 # For non-partitioned topics, consistent hashing is used by default.
 activeConsumerFailoverConsistentHashing=false
 

conf/standalone.conf

@@ -152,7 +152,7 @@ maxMessageSizeCheckIntervalInSeconds=60
 # How long to delay rewinding cursor and dispatching messages when active consumer is changed
 activeConsumerFailoverDelayTimeMillis=1000
 
-# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. 
+# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type.
 # For non-partitioned topics, consistent hashing is used by default.
 activeConsumerFailoverConsistentHashing=false
 
@@ -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 theme
+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 theme."
+    )
+    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(__ -> {