Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
zjxxzjwang wants to merge 7 commits into
base: master
Choose a base branch
from
+39 −3
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
594ec35
[improve][broker]Optimize the partition size of the system theme to r…
Dec 26, 2024
d04fe96
[improve][broker]Optimize the partition size of the system theme to r…
Dec 26, 2024
586d1fc
[improve][broker]Optimize the partition size of the system theme to r…
Dec 31, 2024
5b52438
[improve][broker]Optimize the partition size of the system theme to r…
Jan 6, 2025
a1d41fa
[improve][broker]Optimize the partition size of the system theme to r…
Jan 7, 2025
3984650
fix checkstyle
dao-jun Feb 24, 2025
9155035
Merge branch 'refs/heads/master' into master-optimizeTheNumberOfParti…
zjxxzjwang Feb 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions conf/broker.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,6 +212,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
Expand Down
3 changes: 3 additions & 0 deletions conf/standalone.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1322,6 +1322,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
Expand Down
6 changes: 6 additions & 0 deletions pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2343,6 +2343,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;
Copy link
Member

@lhotari lhotari Feb 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would be useful to support existing way where the default number of partitions is determined by defaultNumPartitions, for example setting this value to -1.
For backwards compatibility the default value should be -1 when this change is introduced. We could make the default 1 in master branch for the feature release of Pulsar.

@FieldContext(
category = CATEGORY_STORAGE_ML,
doc = "The class of the managed ledger storage"
Expand Down
6 changes: 4 additions & 2 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3729,9 +3729,11 @@ public boolean isDefaultTopicTypePartitioned(final TopicName topicName, final Op
public int getDefaultNumPartitions(final TopicName topicName, final Optional<Policies> policies) {
AutoTopicCreationOverride autoTopicCreationOverride = getAutoTopicCreationOverride(topicName, policies);
if (autoTopicCreationOverride != null) {
return autoTopicCreationOverride.getDefaultNumPartitions();
return isSystemTopic(topicName) ? autoTopicCreationOverride.getSystemTopicDefaultNumPartitions()
: autoTopicCreationOverride.getDefaultNumPartitions();
} else {
return pulsar.getConfiguration().getDefaultNumPartitions();
return isSystemTopic(topicName) ? pulsar.getConfiguration().getSystemTopicDefaultNumPartitions()
: pulsar.getConfiguration().getDefaultNumPartitions();
Comment on lines +3735 to +3736
Copy link
Member

@lhotari lhotari Feb 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this should check that systemTopicDefaultNumPartitions > 0.

}
}

Expand Down
1 change: 1 addition & 0 deletions ...ar-broker/src/test/java/org/apache/pulsar/broker/systopic/PartitionedSystemTopicTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
4 changes: 4 additions & 0 deletions ...n-api/src/main/java/org/apache/pulsar/common/policies/data/AutoTopicCreationOverride.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,13 +30,17 @@ public interface AutoTopicCreationOverride {

Integer getDefaultNumPartitions();

Integer getSystemTopicDefaultNumPartitions();

interface Builder {
Builder allowAutoTopicCreation(boolean allowTopicCreation);

Builder topicType(String topicType);

Builder defaultNumPartitions(Integer defaultNumPartition);

Builder systemTopicDefaultNumPartitions(Integer systemTopicDefaultNumPartitions);

AutoTopicCreationOverride build();
}

Expand Down
19 changes: 18 additions & 1 deletion .../main/java/org/apache/pulsar/common/policies/data/impl/AutoTopicCreationOverrideImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ public final class AutoTopicCreationOverrideImpl implements AutoTopicCreationOve
private boolean allowAutoTopicCreation;
private String topicType;
private Integer defaultNumPartitions;
private Integer systemTopicDefaultNumPartitions;

public static ValidateResult validateOverride(AutoTopicCreationOverride override) {
if (override == null) {
Expand All @@ -51,6 +52,14 @@ public static ValidateResult validateOverride(AutoTopicCreationOverride override
if (override.getDefaultNumPartitions() <= 0) {
return ValidateResult.fail("[defaultNumPartitions] cannot be less than 1 for partition type.");
}
if (override.getSystemTopicDefaultNumPartitions() == null) {
return ValidateResult.fail(
"[systemTopicDefaultNumPartitions] cannot be null when the type is partitioned.");
}
if (override.getSystemTopicDefaultNumPartitions() <= 0) {
return ValidateResult.fail(
"[systemTopicDefaultNumPartitions] cannot be less than 1 for partition type.");
}
Comment on lines +55 to +62
Copy link
Member

@lhotari lhotari Feb 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This wouldn't be backwards compatible for existing deployments of Pulsar. Please handle backwards compatibility in some way.

Copy link
Member

@lhotari lhotari Feb 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One possible way would be to use defaultNumPartitions in the case that systemTopicDefaultNumPartitions is null or <= 0

} else if (TopicType.NON_PARTITIONED.toString().equals(override.getTopicType())) {
if (override.getDefaultNumPartitions() != null) {
return ValidateResult.fail("[defaultNumPartitions] is not allowed to be"
Expand All @@ -69,6 +78,7 @@ public static class AutoTopicCreationOverrideImplBuilder implements AutoTopicCre
private boolean allowAutoTopicCreation;
private String topicType;
private Integer defaultNumPartitions;
private Integer systemTopicDefaultNumPartitions;

public AutoTopicCreationOverrideImplBuilder allowAutoTopicCreation(boolean allowAutoTopicCreation) {
this.allowAutoTopicCreation = allowAutoTopicCreation;
Expand All @@ -85,8 +95,15 @@ public AutoTopicCreationOverrideImplBuilder defaultNumPartitions(Integer default
return this;
}

public AutoTopicCreationOverrideImplBuilder systemTopicDefaultNumPartitions(
Integer systemTopicDefaultNumPartitions) {
this.systemTopicDefaultNumPartitions = systemTopicDefaultNumPartitions;
return this;
}

public AutoTopicCreationOverrideImpl build() {
return new AutoTopicCreationOverrideImpl(allowAutoTopicCreation, topicType, defaultNumPartitions);
return new AutoTopicCreationOverrideImpl(allowAutoTopicCreation, topicType, defaultNumPartitions,
systemTopicDefaultNumPartitions);
}
}
}