You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by yu...@apache.org on 2026/08/27 08:38:57 UTC
(pulsar) branch master updated: [feat][broker] PIP-470: Close inactive topics without deleting their data (#25574)
This is an automated email from the ASF dual-hosted git repository.
poorbarcode pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 3bf3ec2237ff [feat][broker] PIP-470: Close inactive topics without deleting their data (#25574)
3bf3ec2237ff is described below
commit 3bf3ec2237ff4cb97dc3cf2247a6d4de2765cdc2
Author: Penghui Li <pe...@apache.org>
AuthorDate: Thu Aug 27 01:38:50 2026 -0700
[feat][broker] PIP-470: Close inactive topics without deleting their data (#25574)
Co-authored-by: Claude Opus 4.7 (1M context) <no...@anthropic.com>
---
pip/pip-470.md | 378 +++++++++++++++++++++
.../apache/pulsar/broker/ServiceConfiguration.java | 22 ++
.../org/apache/pulsar/broker/PulsarService.java | 17 +
.../pulsar/broker/service/AbstractTopic.java | 4 +
.../pulsar/broker/service/BrokerService.java | 5 +-
.../service/nonpersistent/NonPersistentTopic.java | 29 +-
.../broker/service/persistent/PersistentTopic.java | 36 +-
.../broker/service/InactiveTopicCloseTest.java | 228 +++++++++++++
8 files changed, 713 insertions(+), 6 deletions(-)
diff --git a/pip/pip-470.md b/pip/pip-470.md
new file mode 100644
index 000000000000..11361db59ee9
--- /dev/null
+++ b/pip/pip-470.md
@@ -0,0 +1,378 @@
+# PIP-470: Close Inactive Topics Without Deleting Them
+
+# Background Knowledge
+
+Pulsar has a long-standing inactive-topic GC mechanism controlled by the broker
+configuration `brokerDeleteInactiveTopicsEnabled`. When enabled, the broker
+periodically iterates over loaded topics, detects topics with no active
+producers/consumers (and optionally no subscriptions or no backlog), and
+**deletes** them — both the in-memory topic instance and the persistent data
+in BookKeeper, plus topic-level metadata.
+
+The detection logic lives in `PersistentTopic.checkGC()` /
+`NonPersistentTopic.checkGC()`. The policy object is
+`InactiveTopicPolicies`, which has two modes:
+
+- `delete_when_no_subscriptions`
+- `delete_when_subscriptions_caught_up`
+
+Both modes lead to permanent deletion of the topic and its data.
+
+Each loaded `PersistentTopic` pins a non-trivial amount of broker memory:
+a managed-ledger with its cache, subscription/cursor state, rate limiters,
+dispatchers, schema references, and — notably — per-topic metric series.
+Deployments with millions of low-traffic topics routinely hit memory pressure
+and metrics cardinality pressure even when most topics are idle most of the
+time.
+
+Pulsar also supports topic **unload** via the admin API
+(`pulsar-admin topics unload <topic>`) and namespace-bundle unload. Unload
+evicts the topic from the broker's topic map and releases all broker-side
+resources, while the ledgers in BookKeeper are preserved. The next
+produce/consume call transparently reloads the topic. There is currently no
+automatic mechanism that unloads idle topics based on inactivity.
+
+# Motivation
+
+Operators with very large topic counts (commonly tens of thousands to millions
+per cluster, with a long tail of low-traffic topics) face two problems today:
+
+1. **Broker memory pressure.** Every loaded topic holds a managed-ledger,
+ cursors, rate limiters, dispatchers, and metric series. An idle topic that
+ hasn't been produced/consumed for hours still occupies all of that memory.
+2. **Metrics cardinality.** Per-topic metrics grow linearly with the number
+ of loaded topics. Scrape endpoints get large and the monitoring system
+ gets expensive.
+
+The natural remedy today is `brokerDeleteInactiveTopicsEnabled`, but that
+**deletes the data**. Many users explicitly do not want deletion — they want
+to keep the data around for later consumers, replay, compliance retention, etc.
+Their only options today are:
+
+- Leave all idle topics loaded and accept the memory/metrics cost.
+- Build an external cron job that polls `topics stats` for zero
+ producers/consumers and calls `pulsar-admin topics unload` per topic.
+
+The external-cron approach works but is awkward: it reimplements the existing
+broker inactivity detection, has to run with admin credentials, adds a moving
+part to the deployment, and lacks integration with the existing
+`InactiveTopicPolicies` configuration surface.
+
+This PIP adds a first-class broker option that performs the same idle-topic
+detection the broker already does for delete-GC, but performs a **close**
+(unload) instead of a delete when the topic is determined to be inactive.
+
+# Goals
+
+## In Scope
+
+- A new broker configuration `brokerCloseInactiveTopicsEnabled` (default
+ `false`) that, when enabled, causes the inactivity monitor to **close**
+ (unload) topics determined to be inactive rather than delete them.
+- Mutual exclusion with `brokerDeleteInactiveTopicsEnabled`: only one of the
+ two may be enabled at a time. Broker start fails fast if both are set.
+- Restriction to `delete_when_no_subscriptions`: close mode requires
+ `brokerDeleteInactiveTopicsMode` to be `delete_when_no_subscriptions`. Broker
+ start fails fast on any other value, and a namespace- or topic-level policy
+ that overrides the mode at runtime disables close for that topic rather than
+ closing it. See *Why close mode requires `delete_when_no_subscriptions`*.
+- Precedence over policy-level delete: while `brokerCloseInactiveTopicsEnabled`
+ is on, it overrides any namespace- or topic-level
+ `inactive_topic_policies.deleteWhileInactive`. Inactive topics are closed,
+ never deleted, so the data-preservation guarantee holds broker-wide rather
+ than only for namespaces that happen to leave the policy unset.
+- Reuse of the existing inactivity detection: same
+ `brokerDeleteInactiveTopicsMode`, `brokerDeleteInactiveTopicsFrequencySeconds`,
+ and `brokerDeleteInactiveTopicsMaxInactiveDurationSeconds` settings determine
+ when a topic is considered inactive. No new detection surface is introduced
+ for v1.
+- Data preservation guarantee: only the in-memory topic instance is released.
+ BookKeeper ledgers, topic metadata, subscriptions, cursors, and topic-level
+ policies are all retained. The next producer/consumer reconnect transparently
+ reloads the topic.
+- Coverage for both `PersistentTopic` and `NonPersistentTopic`.
+
+## Out of Scope
+
+- A per-topic or per-namespace override for close-on-inactive, and therefore
+ any mix of close for some namespaces and delete for others. The v1 knob is
+ broker-level only and, while on, it suppresses policy-level delete entirely;
+ a follow-up PIP can extend `InactiveTopicPolicies` with an action field
+ (`delete` vs `close`) if operators want per-namespace control.
+- Supporting `delete_when_subscriptions_caught_up` under close mode. Making it
+ useful would require a new detection input (e.g. "caught up *and* no
+ connected consumers"), which is deliberately deferred.
+- Changing the existing `InactiveTopicDeleteMode` enum or
+ `InactiveTopicPolicies` schema.
+- A new admin API. Manual `pulsar-admin topics unload` remains available for
+ ad-hoc use.
+- Partitioned-topic metadata handling changes. Because nothing is deleted,
+ `brokerDeleteInactivePartitionedTopicMetadataEnabled` has no interaction
+ with the new close mode.
+- Changing the detection cadence or detection semantics beyond reusing
+ what's there.
+
+# High Level Design
+
+The existing inactivity monitor in `BrokerService.startInactivityMonitor()`
+schedules `checkGC()` on all loaded topics when
+`brokerDeleteInactiveTopicsEnabled` is true. With this PIP, the monitor also
+runs when `brokerCloseInactiveTopicsEnabled` is true.
+
+`PersistentTopic.checkGC()` and `NonPersistentTopic.checkGC()` gain a branch:
+if close mode is enabled, after the same isActive / duration /
+replication-producer checks pass, the topic calls its own
+`close(disconnectClients=true, closeWithoutWaitingClientDisconnect=false)`
+path. That path already:
+
+- closes subscriptions and the managed ledger (without deleting ledgers),
+- removes the topic entry from `BrokerService.topics` via `disposeTopic` →
+ `removeTopicFromCache`.
+
+`close()` is *also* capable of disconnecting producers and consumers, but on
+this path there is nothing left to disconnect. `isActive()` returns true while
+`hasProducersActive()` or `hasActiveReplicators()` holds, and under the only
+supported mode (`delete_when_no_subscriptions`) it returns true while any
+subscription exists — and a connected consumer implies a subscription. So by
+the time GC reaches `close()`, every producer, consumer and replicator is
+already gone, and the disconnect loops are no-ops. `disconnectClients=true` is
+still the right argument: it selects the `waitDisconnectClients` close type,
+which closes the managed ledger only after all resources are released, and it
+covers the window between the inactivity check and the close.
+
+Data on BookKeeper is untouched. Subsequent lookups reload the topic from
+metadata and BookKeeper exactly as they do after a manual unload or a broker
+restart.
+
+The **retention-window check** (`shouldTopicBeRetained`) is intentionally
+bypassed for the close branch — that check exists to avoid deleting data
+that's still within the retention window, which is moot when we are not
+deleting anything.
+
+# Detailed Design
+
+## Public-facing Changes
+
+### Configuration
+
+One new dynamic broker configuration in `ServiceConfiguration`:
+
+```java
+@FieldContext(
+ category = CATEGORY_POLICIES,
+ dynamic = true,
+ doc = "Enable closing (unloading from broker memory) of inactive topics without deleting their data. ..."
+)
+private boolean brokerCloseInactiveTopicsEnabled = false;
+```
+
+Behavior:
+
+- Default: `false` (no behavior change for existing deployments).
+- Mutually exclusive with `brokerDeleteInactiveTopicsEnabled`. If both are
+ `true` at broker startup, the broker fails fast with an
+ `IllegalArgumentException`.
+- Requires `brokerDeleteInactiveTopicsMode` to be `delete_when_no_subscriptions`.
+ Any other value at broker startup is also rejected with an
+ `IllegalArgumentException`.
+- Takes precedence over `inactive_topic_policies.deleteWhileInactive` at the
+ namespace and topic level: while it is on, `checkGC` never takes the delete
+ branch, whatever the policy says.
+- Dynamic: can be changed at runtime via the existing dynamic-config
+ mechanism. Dynamic transitions that would produce a conflicting state
+ (both flags simultaneously true) are rejected by the same validation path.
+ A runtime change of the *mode* is not rejected — it is handled per topic in
+ `checkGC`, which simply skips the close for topics whose effective mode is
+ not `delete_when_no_subscriptions`.
+
+Reused configuration (no changes):
+
+- `brokerDeleteInactiveTopicsMode` — selects `delete_when_no_subscriptions`
+ vs `delete_when_subscriptions_caught_up` semantics for the inactivity check.
+ Close mode supports only `delete_when_no_subscriptions`.
+ The name is retained for v1 to avoid churn; a rename to a neutral name
+ (e.g. `inactiveTopicsMode`) could be proposed separately.
+- `brokerDeleteInactiveTopicsFrequencySeconds` — how often the monitor runs.
+- `brokerDeleteInactiveTopicsMaxInactiveDurationSeconds` — inactivity
+ threshold before the action fires.
+
+### Public API / Binary / Wire
+
+None. No changes to the Pulsar admin API, client protocol, or topic
+policies schema. The `InactiveTopicPolicies` schema is untouched; only its
+*effect* changes while close mode is enabled, where `deleteWhileInactive` is
+overridden by the broker-level close switch and the delete mode other than
+`delete_when_no_subscriptions` disables inactive-topic GC for that topic.
+
+### CLI
+
+None.
+
+### Metrics
+
+None added. Existing per-topic metrics naturally shrink when topics are
+closed. A follow-up could add a broker-level counter
+`pulsar_broker_topic_inactive_close_total` but it is not required for v1.
+
+## Design & Implementation Details
+
+### Validation
+
+`PulsarService.start()` already performs static config validation. We add:
+
+```java
+if (config.isBrokerDeleteInactiveTopicsEnabled() && config.isBrokerCloseInactiveTopicsEnabled()) {
+ throw new IllegalArgumentException(
+ "brokerDeleteInactiveTopicsEnabled and brokerCloseInactiveTopicsEnabled are mutually "
+ + "exclusive. Enable at most one of them.");
+}
+
+if (config.isBrokerCloseInactiveTopicsEnabled()
+ && config.getBrokerDeleteInactiveTopicsMode() != InactiveTopicDeleteMode.delete_when_no_subscriptions) {
+ throw new IllegalArgumentException(
+ "brokerCloseInactiveTopicsEnabled only supports brokerDeleteInactiveTopicsMode="
+ + "delete_when_no_subscriptions. ...");
+}
+```
+
+Startup validation can only cover the broker-level value. The effective mode is
+resolved per topic from `InactiveTopicPolicies` (topic → namespace → broker), so
+a namespace or topic policy can still set `delete_when_subscriptions_caught_up`
+on a running broker; `checkGC` guards against that at runtime.
+
+### Why close mode requires `delete_when_no_subscriptions`
+
+Under `delete_when_subscriptions_caught_up`, `isActive()` only checks
+`hasBacklogs()` — a topic whose subscriptions are all caught up is "inactive"
+**even while consumers are connected**. Deleting such a topic is a deliberate,
+terminal act, so nothing comes back. Closing it is not: the close disconnects
+those consumers, they immediately reconnect, the reconnect reloads the topic,
+and one inactivity window later GC closes it again. The result is a permanent
+unload/reload loop that frees no memory and adds churn for every idle-but-
+connected consumer in the cluster.
+
+Under `delete_when_no_subscriptions` the problem cannot arise: any subscription
+keeps the topic active, and a connected consumer always implies a subscription,
+so a topic is only ever closed once every client has genuinely gone away.
+
+### Scheduler
+
+`BrokerService.startInactivityMonitor()` is updated to start the monitor
+when **either** flag is enabled:
+
+```java
+if (config.isBrokerDeleteInactiveTopicsEnabled() || config.isBrokerCloseInactiveTopicsEnabled()) {
+ int interval = config.getBrokerDeleteInactiveTopicsFrequencySeconds();
+ inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> checkGC(), interval, interval,
+ TimeUnit.SECONDS);
+}
+```
+
+### Helper on AbstractTopic
+
+```java
+public boolean isCloseWhileInactive() {
+ return brokerService.pulsar().getConfiguration().isBrokerCloseInactiveTopicsEnabled();
+}
+```
+
+Kept broker-level only in v1; future work may read from a per-topic policy.
+
+### PersistentTopic.checkGC
+
+The method now branches on action:
+
+1. Compute `closeEnabled = isCloseWhileInactive()` and
+ `deleteEnabled = !closeEnabled && isDeleteWhileInactive()`. If neither is
+ true, return. Note the order: close is evaluated **first**, so the
+ broker-level close switch suppresses a policy-level `deleteWhileInactive`
+ instead of losing to it.
+2. **New for close mode:** if the effective `deleteMode` is not
+ `delete_when_no_subscriptions`, log at debug and return without acting.
+ This catches namespace/topic policies that override the mode after the
+ startup validation has passed.
+3. Detect inactivity using the existing `isActive(deleteMode)` and the
+ inactivity-duration check.
+4. **New for close mode:** skip the `shouldTopicBeRetained()` guard, since
+ retention only guards against data loss, which is irrelevant here.
+5. Close replication producers first (same as today). If remote producers
+ are connected, abort with `TopicBusyException`.
+6. If close mode is active, call `close(true, false)`. That path closes
+ subscriptions and the managed ledger and removes the topic from the broker
+ cache via `disposeTopic`. Log success; on `TopicBusyException`, log at debug
+ (topic became active again and will be reloaded).
+7. Otherwise proceed with the existing delete branch unchanged.
+
+### NonPersistentTopic.checkGC
+
+Parallel change: same `closeEnabled`/`deleteEnabled` gate in the same order; if
+close mode is active, call `close(true, false)` and skip
+`tryToDeletePartitionedMetadata`.
+
+No delete-mode guard is needed here. `NonPersistentTopic.isActive()` is
+`!subscriptions.isEmpty() || hasLocalProducers()` — it ignores `deleteMode`
+entirely and therefore always behaves like `delete_when_no_subscriptions`, so
+the unload/reload loop described above cannot occur for non-persistent topics.
+
+### Race: topic reloaded before close completes
+
+A client can connect between the inactivity check and the call to `close()`.
+This is already handled by `close()`'s fencing logic: `fenceTopicToCloseOrDelete`
+rejects new client operations during close. If a lookup races in and sees
+the topic is gone from the cache it will re-create a fresh `PersistentTopic`
+instance from metadata, which is exactly the desired behavior.
+
+### Race: close completes and topic is immediately reloaded
+
+Expected. The new `PersistentTopic` instance has its own `lastActive`
+initialized to now; another full inactivity window must elapse before it
+becomes a close candidate again. Activity produced after reload resets the
+clock as usual.
+
+### Interaction with unload-based load balancing
+
+The existing bundle-unload mechanism and this new inactive-close mechanism
+are independent. A bundle unload closes every topic in the bundle
+regardless of activity; the new mechanism closes only inactive topics,
+cluster-wide. They compose without conflict.
+
+### Interaction with `brokerDeleteInactivePartitionedTopicMetadataEnabled`
+
+Because nothing is deleted, partitioned-topic metadata cleanup does not run
+under close mode. The metadata (partition count, topic policies) is
+preserved.
+
+### Backwards compatibility
+
+The new configuration defaults to `false`, so existing deployments see no
+behavioral change. Every behavior described here — the precedence over
+`inactive_topic_policies.deleteWhileInactive` and the delete-mode restriction —
+is reachable only once an operator sets `brokerCloseInactiveTopicsEnabled=true`.
+The delete path is byte-for-byte unchanged when the flag is off.
+
+## Security Considerations
+
+None. This PIP introduces no new network-accessible endpoints, no new
+credentials, and no new authorization surface. Inactivity detection and
+close use the same internal code paths that the broker already executes for
+the delete variant and for admin-driven unload.
+
+# Monitoring
+
+Operators can watch for the log line
+`Topic closed successfully due to inactivity` to observe the feature
+firing. The existing `pulsar_topics_count` metric and per-topic metric
+series naturally shrink as idle topics are closed. Producers/consumers
+reconnecting to a closed topic exercise the same load-topic path that's
+already exercised at broker startup.
+
+# General Notes
+
+This PIP is intentionally minimal — one new flag, one branch in `checkGC`.
+Everything else is reused: the inactivity detection, the close code path,
+the dispose-from-cache machinery, and the admin-driven unload semantics.
+
+# Links
+
+* Mailing List discussion thread: TBD
+* Mailing List voting thread: TBD
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index fca90fc588d7..9a31de009605 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -831,6 +831,28 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private Integer brokerDeleteInactiveTopicsMaxInactiveDurationSeconds = null;
+ @FieldContext(
+ category = CATEGORY_POLICIES,
+ dynamic = true,
+ doc = "Enable closing (unloading from broker memory) of inactive topics without deleting their data.\n"
+ + "When a topic is deemed inactive (no producers and no subscriptions), the broker will close the topic\n"
+ + "instance, releasing in-memory resources such as the managed ledger cache, subscription state, and\n"
+ + "per-topic metrics. The topic data in BookKeeper is preserved; clients will transparently reload the\n"
+ + "topic on the next produce/consume.\n"
+ + "This option is mutually exclusive with 'brokerDeleteInactiveTopicsEnabled': only one of the two may\n"
+ + "be enabled at a time. It also requires 'brokerDeleteInactiveTopicsMode' to be\n"
+ + "'delete_when_no_subscriptions'; with 'delete_when_subscriptions_caught_up' a topic is inactive as\n"
+ + "soon as its subscriptions are caught up even while consumers are connected, so closing it would only\n"
+ + "disconnect those consumers and immediately reload the topic. The broker fails to start on either\n"
+ + "unsupported combination.\n"
+ + "While enabled, this broker-level setting takes precedence over any namespace- or topic-level\n"
+ + "'inactive_topic_policies.deleteWhileInactive': inactive topics are closed, never deleted.\n"
+ + "The inactivity detection reuses 'brokerDeleteInactiveTopicsMode',\n"
+ + "'brokerDeleteInactiveTopicsFrequencySeconds', and\n"
+ + "'brokerDeleteInactiveTopicsMaxInactiveDurationSeconds'."
+ )
+ private boolean brokerCloseInactiveTopicsEnabled = false;
+
@FieldContext(
category = CATEGORY_POLICIES,
dynamic = true,
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
index 526f5cce5570..a2d9f22da299 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
@@ -169,6 +169,7 @@ import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterDataImpl;
+import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
import org.apache.pulsar.common.protocol.schema.SchemaStorage;
import org.apache.pulsar.common.util.FutureUtil;
@@ -900,6 +901,22 @@ public class PulsarService implements AutoCloseable, ShutdownService {
config.getDefaultRetentionTimeInMinutes() * 60));
}
+ if (config.isBrokerDeleteInactiveTopicsEnabled() && config.isBrokerCloseInactiveTopicsEnabled()) {
+ throw new IllegalArgumentException(
+ "brokerDeleteInactiveTopicsEnabled and brokerCloseInactiveTopicsEnabled are mutually "
+ + "exclusive. Enable at most one of them.");
+ }
+
+ if (config.isBrokerCloseInactiveTopicsEnabled()
+ && config.getBrokerDeleteInactiveTopicsMode()
+ != InactiveTopicDeleteMode.delete_when_no_subscriptions) {
+ throw new IllegalArgumentException(
+ "brokerCloseInactiveTopicsEnabled only supports brokerDeleteInactiveTopicsMode="
+ + "delete_when_no_subscriptions. Under delete_when_subscriptions_caught_up a topic "
+ + "whose subscriptions are caught up is inactive even while consumers are still "
+ + "connected, so closing it would repeatedly unload and reload the topic.");
+ }
+
openTelemetryTopicStats = new OpenTelemetryTopicStats(this);
openTelemetryConsumerStats = new OpenTelemetryConsumerStats(this);
openTelemetryProducerStats = new OpenTelemetryProducerStats(this);
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 3975b180cf6c..a77543da5f16 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
@@ -1386,6 +1386,10 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener {
return topicPolicies.getInactiveTopicPolicies().get().isDeleteWhileInactive();
}
+ public boolean isCloseWhileInactive() {
+ return brokerService.pulsar().getConfiguration().isBrokerCloseInactiveTopicsEnabled();
+ }
+
public boolean deletePartitionedTopicMetadataWhileInactive() {
return brokerService.pulsar().getConfiguration().isBrokerDeleteInactivePartitionedTopicMetadataEnabled();
}
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 458e8777bdc5..2ed9ec422eca 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
@@ -827,8 +827,9 @@ public class BrokerService implements Closeable {
}
protected void startInactivityMonitor() {
- if (pulsar().getConfiguration().isBrokerDeleteInactiveTopicsEnabled()) {
- int interval = pulsar().getConfiguration().getBrokerDeleteInactiveTopicsFrequencySeconds();
+ ServiceConfiguration config = pulsar().getConfiguration();
+ if (config.isBrokerDeleteInactiveTopicsEnabled() || config.isBrokerCloseInactiveTopicsEnabled()) {
+ int interval = config.getBrokerDeleteInactiveTopicsFrequencySeconds();
inactivityMonitor.scheduleAtFixedRateNonConcurrently(() -> checkGC(), interval, interval,
TimeUnit.SECONDS);
if (pulsar().getConfig().getBrokerReplicationInactiveThresholdSeconds() > 0) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index 8377162a7fea..fc8bd5b5e496 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -1091,7 +1091,14 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
@Override
public void checkGC() {
- if (!isDeleteWhileInactive()) {
+ // Close-on-inactive is a broker-level switch and takes precedence over any namespace- or topic-level
+ // `deleteWhileInactive` policy, so enabling it can never fall through to the delete path.
+ // No delete-mode guard is needed here: isActive() only reports inactivity when there is neither a
+ // subscription nor a local producer, so a non-persistent topic always follows
+ // delete_when_no_subscriptions semantics regardless of the configured mode.
+ boolean closeEnabled = isCloseWhileInactive();
+ boolean deleteEnabled = !closeEnabled && isDeleteWhileInactive();
+ if (!deleteEnabled && !closeEnabled) {
// This topic is not included in GC
return;
}
@@ -1102,12 +1109,30 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
if (System.nanoTime() - lastActive > TimeUnit.SECONDS.toNanos(maxInactiveDurationInSec)) {
// Close repl producers first.
- // Once all repl producers are closed, we can delete the topic,
+ // Once all repl producers are closed, we can delete/close the topic,
// provided no remote producers connected to the broker.
log.debug()
.attr("maxInactiveDurationInSec", maxInactiveDurationInSec)
.log("Topic inactive for seconds, closing repl producers.");
+ if (closeEnabled) {
+ stopReplProducers().thenCompose(v -> close(true, false))
+ .thenRun(() -> log.info("Topic closed successfully due to inactivity"))
+ .exceptionally(e -> {
+ Throwable throwable = e.getCause();
+ if (throwable instanceof TopicBusyException) {
+ log.debug()
+ .exceptionMessage(throwable)
+ .log("Did not close busy topic");
+ replicators.forEach((region, replicator) -> replicator.startProducer());
+ } else {
+ log.warn().exception(e).log("Inactive topic close failed");
+ }
+ return null;
+ });
+ return;
+ }
+
stopReplProducers().thenCompose(v -> delete(true, false))
.thenCompose(__ -> tryToDeletePartitionedMetadata())
.thenRun(() -> log.info("Topic deleted successfully due to inactivity"))
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index d2691c3d9c7b..1b835682859a 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -3621,22 +3621,54 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal
// broker quietly tearing down a segment the controller still considers live.
return;
}
- if (!isDeleteWhileInactive()) {
+ // Close-on-inactive is a broker-level switch and takes precedence over any namespace- or topic-level
+ // `deleteWhileInactive` policy: an operator who enabled it asked for the topic data to be kept, so GC
+ // must never fall back to the delete path while it is on.
+ boolean closeEnabled = isCloseWhileInactive();
+ boolean deleteEnabled = !closeEnabled && isDeleteWhileInactive();
+ if (!deleteEnabled && !closeEnabled) {
// This topic is not included in GC
return;
}
InactiveTopicDeleteMode deleteMode =
topicPolicies.getInactiveTopicPolicies().get().getInactiveTopicDeleteMode();
+ if (closeEnabled && deleteMode != InactiveTopicDeleteMode.delete_when_no_subscriptions) {
+ // Close mode only supports delete_when_no_subscriptions. Under delete_when_subscriptions_caught_up a
+ // topic counts as inactive as soon as its subscriptions are caught up, even while consumers are still
+ // connected; closing it would only disconnect those consumers, which immediately reload the topic,
+ // turning GC into an unload/reload loop. The broker refuses to start on that combination, so reaching
+ // here means a namespace- or topic-level policy overrode the mode at runtime.
+ log.debug()
+ .attr("deleteMode", deleteMode.name())
+ .log("Skipping inactive-topic close, only delete_when_no_subscriptions is supported");
+ return;
+ }
int maxInactiveDurationInSec = topicPolicies.getInactiveTopicPolicies().get().getMaxInactiveDurationSeconds();
if (isActive(deleteMode)) {
lastActive = System.nanoTime();
} else if (System.nanoTime() - lastActive < SECONDS.toNanos(maxInactiveDurationInSec)) {
// Gc interval did not expire yet
return;
- } else if (shouldTopicBeRetained()) {
+ } else if (deleteEnabled && shouldTopicBeRetained()) {
// Topic activity is still within the retention period
return;
} else {
+ if (closeEnabled) {
+ close(true, false)
+ .thenRun(() -> log.info("Topic closed successfully due to inactivity"))
+ .exceptionally(e -> {
+ if (e.getCause() instanceof TopicBusyException) {
+ log.debug()
+ .exceptionMessage(e.getCause())
+ .log("Did not close busy topic");
+ } else {
+ log.warn().exception(e).log("Inactive topic close failed");
+ }
+ return null;
+ });
+ return;
+ }
+
delete(deleteMode == InactiveTopicDeleteMode.delete_when_no_subscriptions,
deleteMode == InactiveTopicDeleteMode.delete_when_subscriptions_caught_up, false)
.thenCompose((res) -> tryToDeletePartitionedMetadata())
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicCloseTest.java
new file mode 100644
index 000000000000..9db8663cfc96
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/InactiveTopicCloseTest.java
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.common.policies.data.InactiveTopicDeleteMode;
+import org.apache.pulsar.common.policies.data.InactiveTopicPolicies;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker")
+public class InactiveTopicCloseTest extends BrokerTestBase {
+
+ private static final String NAMESPACE = "prop/ns-abc";
+
+ @BeforeMethod
+ protected void setup() throws Exception {
+ //No-op
+ }
+
+ @AfterMethod(alwaysRun = true)
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ /** Enable close mode with the only supported delete mode and a 1s inactivity window. */
+ private void setupCloseMode(InactiveTopicDeleteMode mode) throws Exception {
+ conf.setBrokerDeleteInactiveTopicsEnabled(false);
+ conf.setBrokerCloseInactiveTopicsEnabled(true);
+ conf.setBrokerDeleteInactiveTopicsMode(mode);
+ conf.setBrokerDeleteInactiveTopicsFrequencySeconds(1);
+ conf.setBrokerDeleteInactiveTopicsMaxInactiveDurationSeconds(1);
+ super.baseSetup();
+ }
+
+ private void assertStartupFailsWith(String expectedMessageFragment) throws Exception {
+ try {
+ super.baseSetup();
+ fail("expected broker startup to fail with: " + expectedMessageFragment);
+ } catch (Exception e) {
+ Throwable cause = e;
+ while (cause != null) {
+ if (cause instanceof IllegalArgumentException
+ && cause.getMessage() != null
+ && cause.getMessage().contains(expectedMessageFragment)) {
+ return;
+ }
+ cause = cause.getCause();
+ }
+ fail("expected IllegalArgumentException containing '" + expectedMessageFragment + "', got: " + e);
+ }
+ }
+
+ @Test
+ public void testCloseInactiveTopicKeepsDataAndEvictsFromCache() throws Exception {
+ setupCloseMode(InactiveTopicDeleteMode.delete_when_no_subscriptions);
+
+ final String topic = "persistent://" + NAMESPACE + "/testCloseInactive";
+
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscribe();
+ producer.send("hello".getBytes());
+ consumer.close();
+ producer.close();
+
+ // Drop the only subscription so the topic qualifies as inactive in delete_when_no_subscriptions mode.
+ admin.topics().deleteSubscription(topic, "sub");
+
+ // Topic should be evicted from broker cache (closed) but still present in metadata.
+ Awaitility.await().untilAsserted(() ->
+ assertFalse(pulsar.getBrokerService().getTopicReference(topic).isPresent()));
+ assertTrue(admin.topics().getList(NAMESPACE).contains(topic));
+
+ assertMessagePreserved(topic, "sub2", "hello");
+ }
+
+ @Test
+ public void testMutualExclusionWithDeleteInactive() throws Exception {
+ conf.setBrokerDeleteInactiveTopicsEnabled(true);
+ conf.setBrokerCloseInactiveTopicsEnabled(true);
+ assertStartupFailsWith("mutually exclusive");
+ }
+
+ /**
+ * Close mode only supports delete_when_no_subscriptions: under delete_when_subscriptions_caught_up a topic
+ * with connected but caught-up consumers is inactive, so closing it would unload and reload it forever.
+ */
+ @Test
+ public void testCloseModeRejectsCaughtUpDeleteMode() throws Exception {
+ conf.setBrokerDeleteInactiveTopicsEnabled(false);
+ conf.setBrokerCloseInactiveTopicsEnabled(true);
+ conf.setBrokerDeleteInactiveTopicsMode(InactiveTopicDeleteMode.delete_when_subscriptions_caught_up);
+ assertStartupFailsWith("only supports brokerDeleteInactiveTopicsMode");
+ }
+
+ @Test
+ public void testActiveTopicIsNotClosed() throws Exception {
+ setupCloseMode(InactiveTopicDeleteMode.delete_when_no_subscriptions);
+
+ final String topic = "persistent://" + NAMESPACE + "/testActiveNotClosed";
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscribe();
+ try {
+ // Wait past the inactivity window; topic must remain loaded because a subscription exists.
+ Thread.sleep(3000);
+ assertTrue(pulsar.getBrokerService().getTopicReference(topic).isPresent());
+ } finally {
+ consumer.close();
+ producer.close();
+ }
+ }
+
+ /**
+ * The broker-level close switch must win over a namespace-level deleteWhileInactive policy, otherwise
+ * enabling close mode would still silently delete data in namespaces that set that policy.
+ */
+ @Test
+ public void testCloseWinsOverNamespaceDeleteWhileInactivePolicy() throws Exception {
+ setupCloseMode(InactiveTopicDeleteMode.delete_when_no_subscriptions);
+ admin.namespaces().setInactiveTopicPolicies(NAMESPACE, new InactiveTopicPolicies(
+ InactiveTopicDeleteMode.delete_when_no_subscriptions, 1, true));
+
+ final String topic = "persistent://" + NAMESPACE + "/testCloseWinsOverDeletePolicy";
+
+ Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create();
+ producer.send("hello".getBytes());
+ producer.close();
+
+ // No subscription was ever created, so the topic is inactive right away.
+ Awaitility.await().untilAsserted(() ->
+ assertFalse(pulsar.getBrokerService().getTopicReference(topic).isPresent()));
+
+ // Closed, not deleted: still in metadata and the data is still readable.
+ assertTrue(admin.topics().getList(NAMESPACE).contains(topic),
+ "topic was deleted despite brokerCloseInactiveTopicsEnabled");
+ assertMessagePreserved(topic, "sub", "hello");
+ }
+
+ /**
+ * Startup validation only covers the broker-level mode. A namespace policy can still switch the effective
+ * mode to delete_when_subscriptions_caught_up at runtime, and checkGC must then skip the topic entirely
+ * rather than close it out from under its connected consumers.
+ */
+ @Test
+ public void testCloseSkippedWhenNamespacePolicyOverridesDeleteMode() throws Exception {
+ setupCloseMode(InactiveTopicDeleteMode.delete_when_no_subscriptions);
+ admin.namespaces().setInactiveTopicPolicies(NAMESPACE, new InactiveTopicPolicies(
+ InactiveTopicDeleteMode.delete_when_subscriptions_caught_up, 1, false));
+
+ final String topic = "persistent://" + NAMESPACE + "/testCaughtUpModeSkipsClose";
+
+ // A consumer with no backlog and no producer is "inactive" under delete_when_subscriptions_caught_up.
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName("sub")
+ .subscribe();
+ try {
+ PersistentTopic persistentTopic =
+ (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic).orElseThrow();
+ // Precondition: the namespace policy really did override the mode, and under that mode the topic
+ // looks inactive - i.e. without the guard in checkGC it would be closed out from under the consumer.
+ assertEquals(persistentTopic.getInactiveTopicPolicies().getInactiveTopicDeleteMode(),
+ InactiveTopicDeleteMode.delete_when_subscriptions_caught_up);
+ assertFalse(persistentTopic.isActive(InactiveTopicDeleteMode.delete_when_subscriptions_caught_up),
+ "precondition: topic must look inactive under the overridden mode");
+
+ Thread.sleep(3000);
+ // Identity, not mere presence: a closed topic is reloaded within milliseconds by the consumer
+ // reconnecting, so "a topic is cached" would still hold in the unload/reload loop this guards
+ // against. Only the very same instance proves checkGC left the topic alone.
+ assertSame(pulsar.getBrokerService().getTopicReference(topic).orElseThrow(), persistentTopic,
+ "topic was closed and reloaded under an unsupported delete mode");
+ assertTrue(admin.topics().getList(NAMESPACE).contains(topic));
+ } finally {
+ consumer.close();
+ }
+ }
+
+ private void assertMessagePreserved(String topic, String subscription, String expected) throws Exception {
+ Consumer<byte[]> reReader = pulsarClient.newConsumer()
+ .topic(topic)
+ .subscriptionName(subscription)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+ try {
+ Message<byte[]> msg = reReader.receive(10, TimeUnit.SECONDS);
+ assertNotNull(msg);
+ assertEquals(new String(msg.getValue()), expected);
+ } finally {
+ reReader.close();
+ }
+ }
+}