This is an automated email from the ASF dual-hosted git repository. lhotari pushed a commit to branch branch-4.0 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit f3818ec533e37fee3351e693e1a9ceabe5c16875 Author: Lari Hotari <[email protected]> AuthorDate: Thu Jul 2 21:16:19 2026 +0300 [improve][broker] Load topic policies on non-persistent topic load and gate the policy replay (#26134) (cherry picked from commit 0629dc5add92c638aa37a01eacb7933723452a5c) --- .../apache/pulsar/broker/ServiceConfiguration.java | 9 ++ .../pulsar/broker/service/AbstractTopic.java | 107 +++++++++++++++++++-- .../SystemTopicBasedTopicPoliciesService.java | 39 ++++++-- .../broker/service/TopicPolicyListenerWrapper.java | 68 +++++++++++-- .../service/nonpersistent/NonPersistentTopic.java | 25 ++--- .../broker/service/persistent/PersistentTopic.java | 57 +---------- .../admin/MetadataStoreTopicPoliciesTest.java | 8 ++ .../pulsar/broker/admin/TopicPoliciesTest.java | 67 +++++++++++++ .../SystemTopicBasedTopicPoliciesServiceTest.java | 64 ++++++++++++ .../service/TopicPolicyListenerWrapperTest.java | 102 ++++++++++++++++++-- .../service/persistent/PersistentTopicTest.java | 7 ++ 11 files changed, 451 insertions(+), 102 deletions(-) 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 22d8e353bc5..4f3455ec82b 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 @@ -1729,6 +1729,15 @@ public class ServiceConfiguration implements PulsarConfiguration { + "please enable the system topic first.") private boolean topicLevelPoliciesEnabled = true; + @FieldContext( + category = CATEGORY_SERVER, + doc = "When enabled, all registered topic-policy listeners in a namespace are re-notified with the current" + + " topic policies after the namespace's topic-policy cache finishes its initial load. Topics load" + + " and apply their own policies when they are loaded, so this broadcast is normally redundant; it" + + " is only needed for custom plugins that register TopicPolicyListeners and depend on it for" + + " backwards compatibility. Disabled by default.") + private boolean topicPolicyListenerReplayEnabled = false; + @FieldContext( category = CATEGORY_SERVER, doc = """ 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 b08f0c5cd10..55f692ea768 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 @@ -31,6 +31,7 @@ import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.EnumSet; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -48,6 +49,7 @@ import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicLongFieldUpdater; import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Function; import java.util.function.ToLongFunction; import lombok.Getter; import lombok.Setter; @@ -58,6 +60,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl; import org.apache.pulsar.broker.resourcegroup.ResourceGroup; import org.apache.pulsar.broker.resourcegroup.ResourceGroupPublishLimiter; import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException; @@ -114,6 +117,10 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener { protected final BrokerService brokerService; + // Wraps this topic as a TopicPolicyListener so topic-policy updates received while the initial policy is still + // loading are buffered and applied in order once initTopicPolicy() completes initialization. + protected final TopicPolicyListenerWrapper topicPolicyListener = new TopicPolicyListenerWrapper(this); + // Prefix for replication cursors protected final String replicatorPrefix; @@ -173,7 +180,17 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener { AtomicLongFieldUpdater.newUpdater(AbstractTopic.class, "usageCount"); private volatile long usageCount = 0; - private Map<String/*subscription*/, SubscriptionPolicies> subscriptionPolicies = Collections.emptyMap(); + // Effective per-subscription policies, merged from the local and global topic policies with local precedence. + // Unlike the PolicyHierarchyValue-backed fields, subscriptionPolicies is a plain map. It is kept as the merge of + // the two scopes below (rather than assigned directly) because the local-before-global initialization order + // (see TopicPolicyListenerWrapper) would otherwise let the global map -- empty by default in TopicPolicies -- + // overwrite and clear the local per-subscription policies that were applied just before it. + // subscriptionPolicies is volatile because it is read on dispatch threads (getSubscriptionDispatchRate) while it + // is updated on the policy-update thread. localSubscriptionPolicies/globalSubscriptionPolicies are only ever read + // and written on the (single) policy-update thread, so they don't need to be volatile. + private volatile Map<String/*subscription*/, SubscriptionPolicies> subscriptionPolicies = Collections.emptyMap(); + private Map<String/*subscription*/, SubscriptionPolicies> localSubscriptionPolicies = Collections.emptyMap(); + private Map<String/*subscription*/, SubscriptionPolicies> globalSubscriptionPolicies = Collections.emptyMap(); protected final LongAdder msgOutFromRemovedSubscriptions = new LongAdder(); protected final LongAdder bytesOutFromRemovedSubscriptions = new LongAdder(); @@ -310,11 +327,36 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener { topicPolicies.getEntryFilters().updateTopicValue(data.getEntryFilters(), isGlobalPolicies); topicPolicies.getDispatcherPauseOnAckStatePersistentEnabled() .updateTopicValue(data.getDispatcherPauseOnAckStatePersistentEnabled(), isGlobalPolicies); - this.subscriptionPolicies = data.getSubscriptionPolicies(); + + // Merge instead of assigning directly: keep the local and global per-subscription policies separately and + // recompute the effective map with local precedence, so applying the (default-empty) global map does not + // clear the local per-subscription policies during the local-before-global initialization. + if (isGlobalPolicies) { + globalSubscriptionPolicies = data.getSubscriptionPolicies(); + } else { + localSubscriptionPolicies = data.getSubscriptionPolicies(); + } + subscriptionPolicies = mergeSubscriptionPolicies(globalSubscriptionPolicies, localSubscriptionPolicies); updateEntryFilters(); } + // Merges the global and local per-subscription policies with local precedence: a subscription present in the + // local policies keeps its local value; otherwise the global value (if any) is used. + private static Map<String, SubscriptionPolicies> mergeSubscriptionPolicies( + Map<String, SubscriptionPolicies> globalSubscriptionPolicies, + Map<String, SubscriptionPolicies> localSubscriptionPolicies) { + if (globalSubscriptionPolicies.isEmpty()) { + return localSubscriptionPolicies; + } + if (localSubscriptionPolicies.isEmpty()) { + return globalSubscriptionPolicies; + } + Map<String, SubscriptionPolicies> merged = new HashMap<>(globalSubscriptionPolicies); + merged.putAll(localSubscriptionPolicies); + return merged; + } + protected void updateTopicPolicyByNamespacePolicy(Policies namespacePolicies) { if (log.isDebugEnabled()) { log.debug("[{}]updateTopicPolicyByNamespacePolicy,data={}", topic, namespacePolicies); @@ -550,12 +592,7 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener { } protected TopicPolicyListener getTopicPolicyListener() { - return this; - } - - protected void registerTopicPolicyListener() { - brokerService.getPulsar().getTopicPoliciesService() - .registerListenerAsync(TopicName.getPartitionedTopicName(topic), getTopicPolicyListener()); + return topicPolicyListener; } protected void unregisterTopicPolicyListener() { @@ -563,6 +600,60 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener { .unregisterListener(TopicName.getPartitionedTopicName(topic), getTopicPolicyListener()); } + /** + * Registers the topic-policy listener and applies the topic's initial policies (global and local) to this topic. + * Shared by {@link org.apache.pulsar.broker.service.persistent.PersistentTopic} and + * {@link org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic} so both load their own policies on + * topic load, which removes the need to broadcast every topic's policy when a namespace's policy cache finishes + * loading (see {@code topicPolicyListenerReplayEnabled}). + * + * <p>Each call re-initializes the listener wrapper and, whatever the outcome, always completes its initialization + * afterwards, so the wrapper never stays in the buffering phase (dropping updates) even if policy loading fails. + * This makes the method safe to run again (e.g. a future retry); runs are expected to be serialized. + */ + protected CompletableFuture<Void> initTopicPolicy() { + final var topicPoliciesService = brokerService.getPulsar().getTopicPoliciesService(); + final var partitionedTopicName = TopicName.getPartitionedTopicName(topic); + + // Begin a fresh initialization phase: updates are buffered until initialization completes below. This resets + // any previous phase so the method can be run again. + topicPolicyListener.startInitialization(); + CompletableFuture<Void> initTopicPolicyFuture = + topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener) + .thenCompose(registered -> { + if (!registered) { + return CompletableFuture.completedFuture(null); + } + if (ExtensibleLoadManagerImpl.isInternalTopic(topic)) { + // Internal topics don't load topic-level policies + return CompletableFuture.completedFuture(null); + } + // future for fetching global topic policies + CompletableFuture<Optional<TopicPolicies>> globalPoliciesFuture = + topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName, + TopicPoliciesService.GetType.GLOBAL_ONLY); + // future for fetching local topic policies + CompletableFuture<Optional<TopicPolicies>> localPoliciesFuture = + topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName, + TopicPoliciesService.GetType.LOCAL_ONLY); + return globalPoliciesFuture.thenCombine(localPoliciesFuture, (global, local) -> { + // finally update the topic policies with the latest value or loaded value + return CompletableFuture.runAsync(() -> + topicPolicyListener.completeInitialization(global.orElse(null), + local.orElse(null)), + getPoliciesNotifyThread()); + }).thenCompose(Function.identity()); + }); + // Whatever the outcome -- success, failure, or the listener not being registered -- make sure the wrapper + // leaves the initialization (buffering) phase, so it forwards any buffered value plus all future live updates + // instead of dropping them. This is a no-op when the loaded policies were already applied above. Return the + // whenComplete stage (not initTopicPolicyFuture) so the returned future completes only after this has run, and + // whenComplete's pass-through semantics carry the original success or failure to the caller's initialize(). + return initTopicPolicyFuture.whenCompleteAsync((v, ex) -> { + topicPolicyListener.completeInitializationUnlessAlreadyCompleted(); + }, getPoliciesNotifyThread()); + } + protected boolean isSameAddressProducersExceeded(Producer producer) { if (isSystemTopic() || producer.isRemote()) { return false; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java index c1cd71f31f2..ef58467c785 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java @@ -864,18 +864,43 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic log.debug("[{}] Reach the end of the system topic.", reader.getSystemTopic().getTopicName()); } - // replay policy message - List<CompletableFuture<Void>> notifyFutures = new ArrayList<>(); - for (Map.Entry<TopicName, TopicPolicies> entry : policiesCache.entrySet()) { - TopicName topicName = entry.getKey(); - TopicPolicies policies = entry.getValue(); - notifyFutures.add(notifyListenersForTopicAsync(topicName, policies)); + // Optionally re-notify the topic-policy listeners for this namespace with the cached policies. + // Topics load their own policies on load (AbstractTopic#initTopicPolicy), so this replay is only + // needed for custom plugins that register TopicPolicyListeners and rely on the broadcast; it is + // off by default (topicPolicyListenerReplayEnabled). + if (pulsarService.getConfiguration().isTopicPolicyListenerReplayEnabled()) { + NamespaceName namespaceObject = reader.getSystemTopic().getTopicName().getNamespaceObject(); + FutureUtil.completeAfter(future, replayTopicPolicyListeners(namespaceObject)); + } else { + future.complete(null); } - FutureUtil.completeAfter(future, FutureUtil.waitForAll(notifyFutures)); } }); } + /** + * Re-notifies the registered topic-policy listeners for every topic in {@code namespace} with the currently + * cached policies (both local and global). Topics apply their own policies on load, so this is only needed for + * custom plugins that register {@link TopicPolicyListener}s and rely on the broadcast when a namespace's policy + * cache finishes loading (see {@code topicPolicyListenerReplayEnabled}). + */ + @VisibleForTesting + CompletableFuture<Void> replayTopicPolicyListeners(NamespaceName namespace) { + List<CompletableFuture<Void>> notifyFutures = new ArrayList<>(); + addNamespacePolicyNotifications(policiesCache, namespace, notifyFutures); + addNamespacePolicyNotifications(globalPoliciesCache, namespace, notifyFutures); + return FutureUtil.waitForAll(notifyFutures); + } + + private void addNamespacePolicyNotifications(Map<TopicName, TopicPolicies> cache, NamespaceName namespace, + List<CompletableFuture<Void>> notifyFutures) { + for (Map.Entry<TopicName, TopicPolicies> entry : cache.entrySet()) { + if (Objects.equals(entry.getKey().getNamespaceObject(), namespace)) { + notifyFutures.add(notifyListenersForTopicAsync(entry.getKey(), entry.getValue())); + } + } + } + // Full teardown of a namespace's topic-policies state: removes and closes the reader, the message-handler // tracker, the cached policies and the init future. Used when the whole namespace is unloaded. @VisibleForTesting diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java index c04d5d30e12..e65acd7741b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapper.java @@ -26,10 +26,15 @@ import org.apache.pulsar.common.policies.data.TopicPolicies; /** * This TopicPolicyListener is used as a wrapper for the real TopicPolicyListener. - * This prevents a race condition in initialization where the topic policy state can change while the - * topic policy state is being applied to the topic in PersistentTopic#initTopicPolicy() method or in - * NonPersistentTopic#initialize method. The impact of the race conditions is that the topic policy state would - * be left in an inconsistent state until another update arrives. This is a rare corner case, but possible. + * This prevents a race condition in initialization where the topic policy state can change while the topic policy + * state is being applied to the topic in AbstractTopic#initTopicPolicy(). The impact of the race condition is that the + * topic policy state would be left inconsistent until another update arrives. This is a rare corner case, but possible. + * + * <p>Updates received while initializing are buffered (only the latest per scope is kept) and applied by + * {@link #completeInitialization}; updates received afterwards are forwarded immediately. The wrapper is reusable so + * that AbstractTopic#initTopicPolicy() can be run again -- for example to retry it, which this enables but does not + * implement. {@link #startInitialization()} begins a new buffering phase and initialization completes at most once per + * phase. Concurrent initialization phases are not supported; {@code initTopicPolicy} runs are serialized by the caller. */ @Slf4j public class TopicPolicyListenerWrapper implements TopicPolicyListener { @@ -42,12 +47,29 @@ public class TopicPolicyListenerWrapper implements TopicPolicyListener { private Optional<TopicPolicies> latestGlobalPolicies; private Optional<TopicPolicies> latestLocalPolicies; private boolean initialized; - private final long createdTimestampNanos = System.nanoTime(); + // Timestamp when the current initialization phase started, set by startInitialization(). Used only to warn if the + // phase takes too long (i.e. completeInitialization was never called after policy loading started). + private long initializationStartedNanos; private static final long INITIALIZATION_WARNING_LOG_INTERVAL_NANOS = TimeUnit.SECONDS.toNanos(30); private int lastIntervalLogged; public TopicPolicyListenerWrapper(TopicPolicyListener realTopicListener) { this.realTopicListener = realTopicListener; + startInitialization(); + } + + /** + * Starts (or restarts) the initialization phase: {@link #onUpdate} buffers updates (keeping only the latest per + * scope) instead of forwarding them, until {@link #completeInitialization} applies them. Called at the start of + * every {@code initTopicPolicy} run so the method can be re-run cleanly (which is what would let a future change + * retry it; no retry is implemented here). Runs before the listener is registered, so no update can arrive before + * the phase (and its warning timer) has started. + */ + public synchronized void startInitialization() { + initialized = false; + latestGlobalPolicies = null; + latestLocalPolicies = null; + initializationStartedNanos = System.nanoTime(); } @Override @@ -77,21 +99,49 @@ public class TopicPolicyListenerWrapper implements TopicPolicyListener { /** * Complete initialization of the TopicPolicyListenerWrapper and emit the latest policies to the real listener. + * * @param loadedGlobalPolicies the loaded global policies - * @param loadedLocalPolicies the loaded local policies + * @param loadedLocalPolicies the loaded local policies */ public synchronized void completeInitialization(TopicPolicies loadedGlobalPolicies, TopicPolicies loadedLocalPolicies) { + // Idempotent: an initialization phase completes at most once. initTopicPolicy runs a terminal + // completeInitializationUnlessAlreadyCompleted() after applying the loaded policies, so a later call must be a + // no-op and must not re-emit policies. A new phase is started explicitly via startInitialization(). + if (initialized) { + return; + } + // The listener might have received a newer value (or a delete) than the loaded one while the loading // was happening; prefer the latest value received during initialization, falling back to the loaded // value only when nothing was received for that scope. - emitInitialPolicies(latestGlobalPolicies, loadedGlobalPolicies); + // + // Emit the local policy before the global policy. A local topic policy takes precedence over a global one, + // so applying the local value first means that by the time the global value is applied the local override is + // already in place and the merged (local-wins) result is what takes effect. Emitting the global value first + // would briefly apply it on its own and let a global-only setting act before the local policy overrides it -- + // e.g. a compaction subscription being created for a global compaction policy even though the local policy + // disables compaction. This does not fully solve such ordering hazards, but it removes them whenever a local + // policy exists. When no local policy exists nothing is emitted for the local scope (see emitInitialPolicies), + // so this ordering does not change behavior for topics that only have a global policy. emitInitialPolicies(latestLocalPolicies, loadedLocalPolicies); + emitInitialPolicies(latestGlobalPolicies, loadedGlobalPolicies); + latestGlobalPolicies = null; latestLocalPolicies = null; initialized = true; } + /** + * Completes initialization with no loaded policies, unless it has already completed. Used as a safety net at the + * end of {@code initTopicPolicy} so the wrapper always leaves the buffering phase -- even when the listener was not + * registered or policy loading failed -- and therefore stops dropping updates: it emits any buffered value and + * forwards all future live updates. A no-op once initialization has completed (e.g. with loaded policies). + */ + public synchronized void completeInitializationUnlessAlreadyCompleted() { + completeInitialization(null, null); + } + private void emitInitialPolicies(Optional<TopicPolicies> latestReceived, TopicPolicies loaded) { if (latestReceived != null) { // A value (or a delete) was received during initialization; it supersedes the loaded value. @@ -104,12 +154,12 @@ public class TopicPolicyListenerWrapper implements TopicPolicyListener { // warn if the initialization takes too long and updates have been received // this helps detect issues where completeInitialization didn't get called after loading policies private void maybeLogWarning() { - long durationNanos = System.nanoTime() - createdTimestampNanos; + long durationNanos = System.nanoTime() - initializationStartedNanos; int warningLogIntervalCount = (int) (durationNanos / INITIALIZATION_WARNING_LOG_INTERVAL_NANOS); if (warningLogIntervalCount > lastIntervalLogged) { log.warn("TopicPolicyUpdate buffered. TopicPolicyListenerWrapper initialization phase took too long. " + "completeInitialization should have been called to complete the phase. " - + "topicPolicyListener={} sinceCreationMs={}", + + "topicPolicyListener={} sinceInitializationStartedMs={}", realTopicListener, TimeUnit.NANOSECONDS.toMillis(durationNanos)); lastIntervalLogged = warningLogIntervalCount; } 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 a3652f2987a..9d03ad5d919 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 @@ -66,7 +66,6 @@ import org.apache.pulsar.broker.service.SubscriptionOption; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.TopicAttributes; import org.apache.pulsar.broker.service.TopicPolicyListener; -import org.apache.pulsar.broker.service.TopicPolicyListenerWrapper; import org.apache.pulsar.broker.service.TransportCnx; import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; import org.apache.pulsar.broker.service.schema.exceptions.NotExistSchemaException; @@ -127,9 +126,6 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol TOPIC_ATTRIBUTES_FIELD_UPDATER = AtomicReferenceFieldUpdater.newUpdater( NonPersistentTopic.class, TopicAttributes.class, "topicAttributes"); - // prevents race conditions in topic policy initialization - private final TopicPolicyListenerWrapper topicPolicyListener = new TopicPolicyListenerWrapper(this); - private static class TopicStats { public double averageMsgSize; public double aggMsgRateIn; @@ -156,7 +152,6 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol public NonPersistentTopic(String topic, BrokerService brokerService) { super(topic, brokerService); this.isFenced = false; - registerTopicPolicyListener(); } private CompletableFuture<Void> updateClusterMigrated() { @@ -183,13 +178,15 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol updateResourceGroupLimiter(policies); return updateClusterMigrated(); }, getPoliciesNotifyThread()) - // Complete the topic-policy listener wrapper so buffered and future topic-level policy - // updates are forwarded to this topic. Without this the wrapper stays uninitialized forever - // and all topic-level policy updates are silently dropped. Unlike PersistentTopic, - // non-persistent topics don't load initial topic policies (matching the previous behavior), - // so the loaded values are passed as null. - .thenRunAsync(() -> topicPolicyListener.completeInitialization(null, null), - getPoliciesNotifyThread()); + // Load the topic's initial policies (global and local) and register the policy listener, so a + // non-persistent topic applies its own policies on load, the same as a persistent topic does. + .thenCompose(ignore -> initTopicPolicy()) + // a failure to load the initial topic policies must not fail topic loading. + .exceptionally(ex -> { + log.warn("[{}] Error loading topic policies during initialization. Ignoring the failure.", + topic, ex); + return null; + }); } @Override @@ -1319,8 +1316,4 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol old -> old != null ? old : new TopicAttributes(TopicName.get(topic))); } - @Override - public TopicPolicyListener getTopicPolicyListener() { - return topicPolicyListener; - } } 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 acdee5f168b..8f9b0ac55a9 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 @@ -59,7 +59,6 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; import java.util.function.BiConsumer; import java.util.function.BiFunction; -import java.util.function.Function; import lombok.Getter; import lombok.Value; import org.apache.bookkeeper.client.BKException.BKNoSuchLedgerExistsException; @@ -135,8 +134,6 @@ import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.broker.service.SubscriptionOption; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.TopicPoliciesService; -import org.apache.pulsar.broker.service.TopicPolicyListener; -import org.apache.pulsar.broker.service.TopicPolicyListenerWrapper; import org.apache.pulsar.broker.service.TransportCnx; import org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage; import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException; @@ -295,9 +292,6 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal @Getter private volatile long lastMaxReadPositionMovedForwardTimestamp = 0; - // prevents race conditions in topic policy initialization - private final TopicPolicyListenerWrapper topicPolicyListener = new TopicPolicyListenerWrapper(this); - @Getter private final ExecutorService orderedExecutor; @@ -506,7 +500,8 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal .thenCompose(ignore -> initTopicPolicy()) .thenCompose(ignore -> removeOrphanReplicationCursors()) .exceptionally(ex -> { - log.warn("[{}] Error getting policies {} and isEncryptionRequired will be set to false", + log.warn("[{}] Error loading topic policies during initialization. Ignoring the failure. " + + "isEncryptionRequired will be set to false. {}", topic, ex.getMessage()); isEncryptionRequired = false; return null; @@ -4679,50 +4674,6 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal }); } - protected CompletableFuture<Void> initTopicPolicy() { - final var topicPoliciesService = brokerService.pulsar().getTopicPoliciesService(); - final var partitionedTopicName = TopicName.getPartitionedTopicName(topic); - - return topicPoliciesService.registerListenerAsync(partitionedTopicName, topicPolicyListener) - .thenCompose(registered -> { - if (!registered) { - return CompletableFuture.completedFuture(null); - } - if (ExtensibleLoadManagerImpl.isInternalTopic(topic)) { - // Internal topics don't load topic-level policies, but the listener wrapper must - // still be initialized so any buffered/future updates are forwarded to the topic - // instead of being silently dropped. - return CompletableFuture.runAsync( - () -> topicPolicyListener.completeInitialization(null, null), - getPoliciesNotifyThread()); - } - // future for fetching global topic policies - CompletableFuture<Optional<TopicPolicies>> globalPoliciesFuture = - topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName, - TopicPoliciesService.GetType.GLOBAL_ONLY); - // future for fetching local topic policies - CompletableFuture<Optional<TopicPolicies>> localPoliciesFuture = - topicPoliciesService.getTopicPoliciesAsync(partitionedTopicName, - TopicPoliciesService.GetType.LOCAL_ONLY); - CompletableFuture<Void> initialPoliciesFuture = - globalPoliciesFuture.thenCombine(localPoliciesFuture, (global, local) -> { - // finally update the topic policies with the latest value or loaded value - return CompletableFuture.runAsync(() -> - topicPolicyListener.completeInitialization(global.orElse(null), - local.orElse(null)), - getPoliciesNotifyThread()); - }).thenCompose(Function.identity()); - return initialPoliciesFuture.exceptionallyCompose(ex -> - // The topic load path logs and continues when initial policy loading fails. Make sure the - // already-registered wrapper is not left buffering future live updates forever. - CompletableFuture.runAsync( - () -> topicPolicyListener.completeInitialization(null, null), - getPoliciesNotifyThread()) - .thenCompose(__ -> FutureUtil.failedFuture( - FutureUtil.unwrapCompletionException(ex)))); - }); - } - @VisibleForTesting public MessageDeduplication getMessageDeduplication() { return messageDeduplication; @@ -4891,8 +4842,4 @@ public class PersistentTopic extends AbstractTopic implements Topic, AddEntryCal return future; } - @Override - public TopicPolicyListener getTopicPolicyListener() { - return topicPolicyListener; - } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/MetadataStoreTopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/MetadataStoreTopicPoliciesTest.java index e7fefa16497..b6927cd5ee8 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/MetadataStoreTopicPoliciesTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/MetadataStoreTopicPoliciesTest.java @@ -44,6 +44,14 @@ public class MetadataStoreTopicPoliciesTest extends TopicPoliciesTest { // Not applicable to MetadataStoreTopicPoliciesService. } + @Test(enabled = false) + @Override + public void testNonPersistentTopicAppliesTopicPolicyOnLoad() throws Exception { + // This test is specific to SystemTopicBasedTopicPoliciesService (casts the service and uses + // getPoliciesCacheInit). The non-persistent load-path behavior itself is backend-agnostic and is + // covered against the default SystemTopicBasedTopicPoliciesService in TopicPoliciesTest. + } + @Test(enabled = false) @Override public void testSystemTopicShouldBeCompacted() throws Exception { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java index 3621c9f21d5..302f060953b 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java @@ -251,6 +251,73 @@ public class TopicPoliciesTest extends MockedPulsarServiceBaseTest { assertEquals(topic1.getHierarchyTopicPolicies().getMaxSubscriptionsPerTopic().get(), Integer.valueOf(10)); } + @Test + public void testNonPersistentTopicAppliesTopicPolicyOnLoad() throws Exception { + // Non-persistent topics now load and apply their own topic policies on load (like persistent topics), so a + // freshly loaded non-persistent topic must already reflect its topic-level policy without waiting for a + // namespace-wide broadcast. Before this change a non-persistent topic never applied its policies on load. + TopicName topicName = TopicName.get( + TopicDomain.non_persistent.value(), + NamespaceName.get(myNamespace), + "test-np-" + UUID.randomUUID() + ); + String topic = topicName.toString(); + + SystemTopicBasedTopicPoliciesService policyService = + (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService(); + + admin.topics().createNonPartitionedTopic(topic); + admin.topicPolicies().setMaxSubscriptionsPerTopicAsync(topic, 10).get(); + + //wait until topic loaded with right policy value. + Awaitility.await().untilAsserted(() -> { + AbstractTopic loaded = (AbstractTopic) pulsar.getBrokerService().getTopic(topic, true).get().get(); + assertEquals(loaded.getHierarchyTopicPolicies().getMaxSubscriptionsPerTopic().get(), Integer.valueOf(10)); + }); + + //unload the topic + pulsar.getNamespaceService().unloadNamespaceBundle(pulsar.getNamespaceService().getBundle(topicName)).get(); + assertFalse(pulsar.getBrokerService().getTopics().containsKey(topic)); + + //re-own the namespace bundle without loading the topic + log.info("lookup={}", admin.lookups().lookupTopic(topic)); + assertTrue(pulsar.getBrokerService().isTopicNsOwnedByBrokerAsync(topicName).join()); + assertFalse(pulsar.getBrokerService().getTopics().containsKey(topic)); + //make sure namespace policy reader is fully started. + Awaitility.await().untilAsserted(() -> + assertTrue(policyService.getPoliciesCacheInit(topicName.getNamespaceObject()).isDone())); + + //load the topic: it must already reflect the topic policy, proving it was applied on load, not via a + //later broadcast. + AbstractTopic loaded = (AbstractTopic) pulsar.getBrokerService().getTopic(topic, true).get().get(); + assertEquals(loaded.getHierarchyTopicPolicies().getMaxSubscriptionsPerTopic().get(), Integer.valueOf(10)); + } + + @Test + public void testGlobalPolicyUpdateDoesNotClearLocalSubscriptionPolicy() throws Exception { + // A global topic policy carries no subscription-level overrides by default (an empty subscriptionPolicies + // map). Because AbstractTopic keeps the local and global per-subscription policies separately and merges them + // with local precedence, applying such a global policy must not clear the local per-subscription dispatch-rate + // policy -- which the previous direct assignment would do under the local-before-global ordering. + final String topic = "persistent://" + myNamespace + "/test-sub-policy-merge-" + UUID.randomUUID(); + final String subName = "sub-1"; + admin.topics().createNonPartitionedTopic(topic); + admin.topics().createSubscription(topic, subName, MessageId.earliest); + + DispatchRate localRate = DispatchRateImpl.builder() + .dispatchThrottlingRateInMsg(100).dispatchThrottlingRateInByte(2048).ratePeriodInSecond(1).build(); + admin.topicPolicies().setSubscriptionDispatchRate(topic, subName, localRate); + + AbstractTopic topicRef = (AbstractTopic) pulsar.getBrokerService().getTopic(topic, false).get().orElseThrow(); + Awaitility.await().untilAsserted(() -> + assertEquals(topicRef.getSubscriptionDispatchRate(subName).getDispatchThrottlingRateInMsg(), 100)); + + // Simulate a global topic-policy update that carries no subscription-level overrides. + topicRef.onUpdate(TopicPolicies.builder().isGlobal(true).build()); + + assertEquals(topicRef.getSubscriptionDispatchRate(subName).getDispatchThrottlingRateInMsg(), 100); + } + @Test public void testSetSizeBasedBacklogQuota() throws Exception { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java index 8ea225ebf62..03081d6d58e 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java @@ -39,6 +39,7 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -47,6 +48,7 @@ import java.util.concurrent.TimeoutException; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.reflect.FieldUtils; +import org.apache.pulsar.broker.ServiceConfiguration; import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest; import org.apache.pulsar.broker.systopic.SystemTopicClient; import org.apache.pulsar.client.admin.PulsarAdminException; @@ -55,11 +57,13 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.events.PulsarEvent; import org.apache.pulsar.common.naming.NamespaceName; +import org.apache.pulsar.common.naming.SystemTopicNames; import org.apache.pulsar.common.naming.TopicName; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.policies.data.TopicPolicies; import org.apache.pulsar.utils.TestLogAppender; +import org.assertj.core.api.Assertions; import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; @@ -842,4 +846,64 @@ public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServic assertSame(reloadInitFuture, spyService.getPoliciesCacheInit(namespace)); Mockito.verify(reloadReader, Mockito.never()).closeAsync(); } + + @Test + public void testReplayTopicPolicyListenersNotifiesOnlyNamespaceScopedLocalAndGlobalPolicies() throws Exception { + SystemTopicBasedTopicPoliciesService service = + (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService(); + final NamespaceName namespaceA = NamespaceName.get(NAMESPACE1); + final NamespaceName namespaceB = NamespaceName.get(NAMESPACE2); + final TopicName localTopicA = TopicName.get("persistent", namespaceA, "replay-local-a"); + final TopicName globalTopicA = TopicName.get("persistent", namespaceA, "replay-global-a"); + final TopicName topicB = TopicName.get("persistent", namespaceB, "replay-b"); + + // Seed the caches: namespace A has one topic with a cached local policy and another with a cached global + // policy; namespace B has a topic with a cached local policy that must not be replayed for namespace A. + service.policiesCache.put(localTopicA, TopicPolicies.builder().isGlobal(false).build()); + service.globalPoliciesCache.put(globalTopicA, TopicPolicies.builder().isGlobal(true).build()); + service.policiesCache.put(topicB, TopicPolicies.builder().isGlobal(false).build()); + + final Map<TopicName, List<TopicPolicies>> received = new ConcurrentHashMap<>(); + for (TopicName topicName : List.of(localTopicA, globalTopicA, topicB)) { + final List<TopicPolicies> updates = new CopyOnWriteArrayList<>(); + received.put(topicName, updates); + service.registerListenerAsync(topicName, updates::add).get(); + } + + service.replayTopicPolicyListeners(namespaceA).get(30, TimeUnit.SECONDS); + + // Only namespace A's topics are notified, once each, and both the local and the global cache are replayed. + Assertions.assertThat(received.get(localTopicA)).hasSize(1); + Assertions.assertThat(received.get(globalTopicA)).hasSize(1); + // Namespace B is left untouched. The pre-fix code iterated the whole cache and replayed every namespace. + Assertions.assertThat(received.get(topicB)).isEmpty(); + } + + @Test + public void testTopicPolicyListenerReplayDisabledByDefault() { + Assertions.assertThat(new ServiceConfiguration().isTopicPolicyListenerReplayEnabled()).isFalse(); + } + + @Test + public void testChangeEventsTopicPolicyLoadDoesNotRecurse() throws Exception { + SystemTopicBasedTopicPoliciesService service = + (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService(); + final String namespaceStr = "system-topic/change-events-recursion"; + admin.namespaces().createNamespace(namespaceStr); + final NamespaceName namespace = NamespaceName.get(namespaceStr); + final TopicName changeEvents = + TopicName.get("persistent", namespace, SystemTopicNames.NAMESPACE_EVENTS_LOCAL_NAME); + + // The __change_events system topic must not load topic-level policies: that would create a policy-cache + // reader on __change_events while __change_events is still loading -- a recursive, deadlocking dependency. + // isSelf() guards getTopicPoliciesAsync so it returns empty for the __change_events topic without ever + // creating a reader. AbstractTopic#initTopicPolicy (now called for persistent AND non-persistent topics) + // relies on this short-circuit when a __change_events topic itself is loaded. + Assertions.assertThat(service.getTopicPoliciesAsync(changeEvents, TopicPoliciesService.GetType.LOCAL_ONLY) + .get(30, TimeUnit.SECONDS)).isEmpty(); + Assertions.assertThat(service.getTopicPoliciesAsync(changeEvents, TopicPoliciesService.GetType.GLOBAL_ONLY) + .get(30, TimeUnit.SECONDS)).isEmpty(); + // No policy-cache reader was created as a side effect, which is what would recurse. + Assertions.assertThat(service.getReaderCaches()).doesNotContainKey(namespace); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java index 0e755359274..9d32342aab5 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/TopicPolicyListenerWrapperTest.java @@ -56,15 +56,15 @@ public class TopicPolicyListenerWrapperTest { assertThat(real.updates).isEmpty(); // On completion, the buffered local value wins over the loaded local value; the loaded global value - // is applied since none was buffered. + // is applied since none was buffered. The local policy is emitted before the global one. TopicPolicies loadedGlobal = globalPolicies(); wrapper.completeInitialization(loadedGlobal, localPolicies()); - assertThat(real.updates).containsExactly(loadedGlobal, bufferedLocal); + assertThat(real.updates).containsExactly(bufferedLocal, loadedGlobal); // After initialization, updates are forwarded immediately. TopicPolicies liveUpdate = localPolicies(); wrapper.onUpdate(liveUpdate); - assertThat(real.updates).containsExactly(loadedGlobal, bufferedLocal, liveUpdate); + assertThat(real.updates).containsExactly(bufferedLocal, loadedGlobal, liveUpdate); } @Test @@ -78,7 +78,7 @@ public class TopicPolicyListenerWrapperTest { wrapper.onUpdate(bufferedLocal); wrapper.completeInitialization(globalPolicies(), localPolicies()); - assertThat(real.updates).containsExactly(bufferedGlobal, bufferedLocal); + assertThat(real.updates).containsExactly(bufferedLocal, bufferedGlobal); } @Test @@ -86,10 +86,12 @@ public class TopicPolicyListenerWrapperTest { RecordingListener real = new RecordingListener(); TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + // The local policy is emitted before the global policy, so a local topic policy takes precedence over a + // global one once both have been applied. TopicPolicies loadedGlobal = globalPolicies(); TopicPolicies loadedLocal = localPolicies(); wrapper.completeInitialization(loadedGlobal, loadedLocal); - assertThat(real.updates).containsExactly(loadedGlobal, loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal, loadedGlobal); } @Test @@ -118,7 +120,93 @@ public class TopicPolicyListenerWrapperTest { wrapper.onUpdate(newerGlobal); wrapper.completeInitialization(globalPolicies(), localPolicies()); - // Global: the newer update wins; Local: the delete (null) wins over the loaded local value. - assertThat(real.updates).containsExactly(newerGlobal, null); + // Local (emitted first): the delete (null) wins over the loaded local value; Global: the newer update wins. + assertThat(real.updates).containsExactly(null, newerGlobal); + } + + @Test + public void shouldNotEmitLocalScopeWhenNoLocalPolicyExists() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + + // With no local policy, only the global policy is emitted; no local onUpdate happens, so the + // local-before-global ordering leaves behavior unchanged for topics that only have a global policy. + TopicPolicies loadedGlobal = globalPolicies(); + wrapper.completeInitialization(loadedGlobal, null); + assertThat(real.updates).containsExactly(loadedGlobal); + } + + @Test + public void shouldIgnoreCompleteInitializationAfterAlreadyCompleted() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + wrapper.startInitialization(); + + TopicPolicies loadedLocal = localPolicies(); + wrapper.completeInitialization(null, loadedLocal); + assertThat(real.updates).containsExactly(loadedLocal); + + // Completing again (e.g. from initTopicPolicy's terminal handler) must be a no-op and must not re-emit. + wrapper.completeInitialization(globalPolicies(), localPolicies()); + wrapper.completeInitializationUnlessAlreadyCompleted(); + assertThat(real.updates).containsExactly(loadedLocal); + } + + @Test + public void shouldEmitBufferedValueAndForwardLiveUpdatesWhenCompletedWithoutLoadedPolicies() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + wrapper.startInitialization(); + + // A policy update arrives while initializing and is buffered. + TopicPolicies buffered = localPolicies(); + wrapper.onUpdate(buffered); + assertThat(real.updates).isEmpty(); + + // initTopicPolicy's terminal handler completes initialization with no loaded policies -- the path taken after + // a policy-load error or when the listener was not registered. The buffered value is emitted and the wrapper + // leaves the buffering phase. + wrapper.completeInitializationUnlessAlreadyCompleted(); + assertThat(real.updates).containsExactly(buffered); + + // Subsequent live updates now flow through instead of being dropped. + TopicPolicies live = globalPolicies(); + wrapper.onUpdate(live); + assertThat(real.updates).containsExactly(buffered, live); + } + + @Test + public void shouldForwardLiveUpdatesAfterCompletingWithNothingBuffered() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + wrapper.startInitialization(); + + // Completed with nothing buffered and no loaded policies (e.g. after a failed load): nothing is emitted, but + // the wrapper still leaves the buffering phase so later live updates are forwarded rather than dropped. + wrapper.completeInitializationUnlessAlreadyCompleted(); + assertThat(real.updates).isEmpty(); + + TopicPolicies live = localPolicies(); + wrapper.onUpdate(live); + assertThat(real.updates).containsExactly(live); + } + + @Test + public void shouldRebufferAndReapplyAfterStartInitializationIsCalledAgain() { + RecordingListener real = new RecordingListener(); + TopicPolicyListenerWrapper wrapper = new TopicPolicyListenerWrapper(real); + wrapper.startInitialization(); + TopicPolicies firstLocal = localPolicies(); + wrapper.completeInitialization(null, firstLocal); + assertThat(real.updates).containsExactly(firstLocal); + + // A new initialization phase (e.g. re-running initTopicPolicy): updates are buffered again until it completes, + // and a value buffered during the phase is applied on completion. + wrapper.startInitialization(); + TopicPolicies bufferedGlobal = globalPolicies(); + wrapper.onUpdate(bufferedGlobal); + assertThat(real.updates).containsExactly(firstLocal); + wrapper.completeInitialization(null, null); + assertThat(real.updates).containsExactly(firstLocal, bufferedGlobal); } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java index 6a91809c4d2..a53e24d1db4 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentTopicTest.java @@ -774,6 +774,13 @@ public class PersistentTopicTest extends BrokerTestBase { public void onUpdate(TopicPolicies policies) { receivedUpdates.add(policies); } + + // initTopicPolicy() moved to AbstractTopic (a different package), so widen it to public here to keep + // this same-package test able to invoke it directly. + @Override + public CompletableFuture<Void> initTopicPolicy() { + return super.initTopicPolicy(); + } } final String topic = "persistent://prop/ns-abc/testTopicPolicyInitFailure-" + UUID.randomUUID();
