Skip to content
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.

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

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

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
3 changes: 3 additions & 0 deletions conf/standalone.conf
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3571,9 +3571,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();
}
}

Expand Down
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
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,8 @@ public interface AutoTopicCreationOverride {
String getTopicType();

Integer getDefaultNumPartitions();

Integer getSystemTopicDefaultNumPartitions();

interface Builder {
Builder allowAutoTopicCreation(boolean allowTopicCreation);
Expand All @@ -37,6 +39,8 @@ interface Builder {

Builder defaultNumPartitions(Integer defaultNumPartition);

Builder systemTopicDefaultNumPartitions(Integer systemTopicDefaultNumPartitions);

AutoTopicCreationOverride build();
}

Expand Down
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.");
}
} 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);
}
}
}
Loading