You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by pe...@apache.org on 2026/10/14 20:19:07 UTC
(pulsar) branch branch-4.0 updated: [fix][broker] Run the inactive topic GC when it is enabled by dynamic configuration (#26894)
This is an automated email from the ASF dual-hosted git repository.
codelipenghui pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 7d0a18cf955b [fix][broker] Run the inactive topic GC when it is enabled by dynamic configuration (#26894)
7d0a18cf955b is described below
commit 7d0a18cf955b83afa344dba553a51a9aaabaa2e3
Author: Penghui Li <pe...@apache.org>
AuthorDate: Tue Oct 13 21:00:24 2026 -0700
[fix][broker] Run the inactive topic GC when it is enabled by dynamic configuration (#26894)
(cherry picked from commit 312b527f0c6a2ee44898f66c32ee0607c7e2f236)
---
.../pulsar/broker/service/AbstractTopic.java | 15 ++++++---
.../pulsar/broker/service/BrokerService.java | 36 +++++++++++++++++-----
.../broker/service/InactiveTopicDeleteTest.java | 17 ++++++++++
3 files changed, 56 insertions(+), 12 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
index 55f692ea7682..821849605127 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java
@@ -467,11 +467,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
private void updateTopicPolicyByBrokerConfig() {
ServiceConfiguration config = brokerService.pulsar().getConfiguration();
- topicPolicies.getInactiveTopicPolicies().updateBrokerValue(new InactiveTopicPolicies(
- config.getBrokerDeleteInactiveTopicsMode(),
- config.getBrokerDeleteInactiveTopicsMaxInactiveDurationSeconds(),
- config.isBrokerDeleteInactiveTopicsEnabled()));
-
+ updateBrokerInactiveTopicPolicies();
updateBrokerSubscriptionTypesEnabled();
topicPolicies.getMaxSubscriptionsPerTopic().updateBrokerValue(config.getMaxSubscriptionsPerTopic());
topicPolicies.getMaxProducersPerTopic().updateBrokerValue(config.getMaxProducersPerTopic());
@@ -1410,6 +1406,15 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
this.topicPublishRateLimiter.update(publishRate);
}
+ // brokerDeleteInactiveTopicsEnabled is dynamic and can be updated online.
+ public void updateBrokerInactiveTopicPolicies() {
+ ServiceConfiguration config = brokerService.pulsar().getConfiguration();
+ topicPolicies.getInactiveTopicPolicies().updateBrokerValue(new InactiveTopicPolicies(
+ config.getBrokerDeleteInactiveTopicsMode(),
+ config.getBrokerDeleteInactiveTopicsMaxInactiveDurationSeconds(),
+ config.isBrokerDeleteInactiveTopicsEnabled()));
+ }
+
// subscriptionTypesEnabled is dynamic and can be updated online.
public void updateBrokerSubscriptionTypesEnabled() {
topicPolicies.getSubscriptionTypesEnabled().updateBrokerValue(
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index c8693c5ec21b..f487997f13e7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -718,14 +718,20 @@ public class BrokerService implements Closeable {
}
protected void startInactivityMonitor() {
- if (pulsar().getConfiguration().isBrokerDeleteInactiveTopicsEnabled()) {
- int interval = pulsar().getConfiguration().getBrokerDeleteInactiveTopicsFrequencySeconds();
- inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> checkGC(), interval, interval,
- TimeUnit.SECONDS);
- if (pulsar().getConfig().getBrokerReplicationInactiveThresholdSeconds() > 0) {
- inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> checkInactiveReplication(), interval,
- interval, TimeUnit.SECONDS);
+ ServiceConfiguration config = pulsar().getConfiguration();
+ // Scheduled even when disabled: brokerDeleteInactiveTopicsEnabled is dynamic, so it is checked on every run.
+ int gcInterval = config.getBrokerDeleteInactiveTopicsFrequencySeconds();
+ inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> {
+ if (isInactiveTopicGCEnabled()) {
+ checkGC();
}
+ }, gcInterval, gcInterval, TimeUnit.SECONDS);
+ if (pulsar().getConfig().getBrokerReplicationInactiveThresholdSeconds() > 0) {
+ inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> {
+ if (isInactiveTopicGCEnabled()) {
+ checkInactiveReplication();
+ }
+ }, gcInterval, gcInterval, TimeUnit.SECONDS);
}
// Deduplication info checker
@@ -753,6 +759,10 @@ public class BrokerService implements Closeable {
}
}
+ private boolean isInactiveTopicGCEnabled() {
+ return pulsar().getConfiguration().isBrokerDeleteInactiveTopicsEnabled();
+ }
+
protected void startMessageExpiryMonitor() {
int interval = pulsar().getConfiguration().getMessageExpiryCheckIntervalInMinutes();
messageExpiryMonitor.scheduleAtFixedRateNonConcurrently(this::checkMessageExpiry, interval, interval,
@@ -3154,6 +3164,10 @@ public class BrokerService implements Closeable {
// add listener to notify topic subscriptionTypesEnabled changed.
registerConfigurationListener("subscriptionTypesEnabled", this::updateBrokerSubscriptionTypesEnabled);
+ // add listener to update the broker-level inactive topic policies of the loaded topics
+ registerConfigurationListener("brokerDeleteInactiveTopicsEnabled",
+ (brokerDeleteInactiveTopicsEnabled) -> updateBrokerInactiveTopicPolicies());
+
// add listener to notify partitioned topic defaultNumPartitions changed
registerConfigurationListener("defaultNumPartitions", defaultNumPartitions -> {
this.updateDefaultNumPartitions((int) defaultNumPartitions);
@@ -3276,6 +3290,14 @@ public class BrokerService implements Closeable {
});
}
+ private void updateBrokerInactiveTopicPolicies() {
+ this.pulsar().getExecutor().execute(() -> forEachTopic(topic -> {
+ if (topic instanceof AbstractTopic abstractTopic) {
+ abstractTopic.updateBrokerInactiveTopicPolicies();
+ }
+ }));
+ }
+
private void updateSubscriptionMessageDispatchRate() {
this.pulsar().getExecutor().execute(() -> {
// update message-rate for each topic subscription
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
index 33e38d97a904..abb4697eae99 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicDeleteTest.java
@@ -644,6 +644,23 @@ public class InactiveTopicDeleteTest extends BrokerTestBase {
});
}
+ @Test
+ public void testEnableBrokerDeleteInactiveTopicsDynamically() throws Exception {
+ conf.setBrokerDeleteInactiveTopicsEnabled(false);
+ conf.setBrokerDeleteInactiveTopicsMode(InactiveTopicDeleteMode.delete_when_no_subscriptions);
+ conf.setBrokerDeleteInactiveTopicsFrequencySeconds(1);
+ conf.setBrokerDeleteInactiveTopicsMaxInactiveDurationSeconds(1);
+ super.baseSetup();
+
+ // Loaded while deletion is disabled, so the topic itself must pick up the change.
+ final String topic = "persistent://prop/ns-abc/testEnableBrokerDeleteInactiveTopicsDynamically";
+ pulsarClient.newProducer().topic(topic).create().close();
+ assertTrue(pulsar.getBrokerService().getTopicReference(topic).isPresent());
+
+ admin.brokers().updateDynamicConfiguration("brokerDeleteInactiveTopicsEnabled", "true");
+ Awaitility.await().untilAsserted(() -> assertFalse(admin.topics().getList("prop/ns-abc").contains(topic)));
+ }
+
@Test
public void testDynamicConfigurationBrokerDeleteInactiveTopicsFrequencySeconds() throws Exception {
conf.setBrokerDeleteInactiveTopicsFrequencySeconds(30);