[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsOwnership #24780

BewareMyPower · 2025-09-25T04:56:23Z

Motivation

I'm working on optimizing the long topic loading path recently and I found these two methods just work the same. The ownership validation does not have a single source of truth, which makes code hard to read.

BrokerService#checkTopicNsOwnership

This method returns a void future that fails if the topic is not owned by the current broker. It's called by:

TransactionMetadataStoreService#handleTcClientConnect
AbstractTopic#addProducer
BrokerService#createNonPersistentTopic
BrokerService#loadOrCreatePersistentTopic
BrokerService#createPendingLoadTopic
NonPersistentTopic#checkTopicNsOwnership
PersistentTopic#internalSubscribe

NamespaceService#isServiceUnitActiveAsync

This method returns a boolean future that returns false if the topic is not owned by the current broker. It's only used in BrokerService#checkOwnershipAndCreatePersistentTopic.
The implementations of these two methods are slightly different, which might also mislead users that they are different.

Here is how isServiceUnitActiveAsync handles topics' bundle:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

Lines 1249 to 1253 in 1b74fe0

    
           Optional<CompletableFuture<OwnedBundle>> optionalFuture = ownershipCache.getOwnedBundleAsync(bundle); 
        
           if (optionalFuture.isEmpty()) { 
        
               return CompletableFuture.completedFuture(false); 
        
           } 
        
           return optionalFuture.get().thenApply(ob -> ob != null && ob.isActive());

However, the logic is duplicated with

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/OwnershipCache.java

Lines 147 to 153 in 1b74fe0

    
           public CompletableFuture<Boolean> checkOwnershipAsync(NamespaceBundle bundle) { 
        
               Optional<CompletableFuture<OwnedBundle>> ownedBundleFuture = getOwnedBundleAsync(bundle); 
        
               if (!ownedBundleFuture.isPresent()) { 
        
                   return CompletableFuture.completedFuture(false); 
        
               } 
        
               return ownedBundleFuture.get() 
        
                       .thenApply(bd -> bd != null && bd.isActive());

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

Lines 1280 to 1286 in 1b74fe0

    
           public CompletableFuture<Boolean> checkBundleOwnership(TopicName topicName, NamespaceBundle bundle) { 
        
               if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) { 
        
                   // TODO: Add unit tests cover it. 
        
                   return loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle); 
        
               } else { 
        
                   return ownershipCache.checkOwnershipAsync(bundle); 
        
               }

which is called by checkTopicNsOwnership:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Line 2376 in 1b74fe0

    
           namespaceService.checkBundleOwnership(topicName, bundle).thenCompose(ownedByThisInstance -> {

The only place that calls isServiceUnitActiveAsync is here:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Line 1739 in 1b74fe0

pulsar.getNamespaceService().isServiceUnitActiveAsync(topicName)

when it returns false, it just completes another future exceptionally:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Lines 1761 to 1764 in 1b74fe0

    
           String msg = String.format("Namespace is being unloaded, cannot add topic %s", topic); 
        
           log.warn(msg); 
        
           pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture)); 
        
           topicFuture.completeExceptionally(new ServiceUnitNotReadyException(msg));

where the error message does not have bundle info included like checkTopicNsOwnership.

Modifications

Use checkTopicNsOwnership instead of isServiceUnitActiveAsync in BrokerService#checkOwnershipAndCreatePersistentTopic.

Remove the isServiceUnitActiveAsync and isServiceUnitActive methods from NamespaceService. The same function can be implemented easily by composing existing methods like:

    public static CompletableFuture<Boolean> isServiceUnitActiveAsync(NamespaceService namespaceService,
                                                                      TopicName topicName) {
        return namespaceService.getBundleAsync(topicName).thenCompose(bundle ->
                namespaceService.checkBundleOwnership(topicName, bundle));
    }

Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository:

…wnership

lhotari

LGTM, thanks for taking actions to cleanup Pulsar code!

codecov-commenter · 2025-09-25T08:59:54Z

Codecov Report

❌ Patch coverage is 80.00000% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.15%. Comparing base (7c8d3c9) to head (03ee056).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
...rg/apache/pulsar/broker/service/BrokerService.java	80.00%	3 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #24780      +/-   ##
============================================
- Coverage     74.23%   74.15%   -0.09%     
- Complexity    33271    33601     +330     
============================================
  Files          1904     1905       +1     
  Lines        148507   148571      +64     
  Branches      17214    17226      +12     
============================================
- Hits         110246   110166      -80     
- Misses        29484    29588     +104     
- Partials       8777     8817      +40

Flag	Coverage Δ
inttests	`26.43% <53.33%> (-0.04%)`	⬇️
systests	`22.79% <53.33%> (-0.03%)`	⬇️
unittests	`73.66% <80.00%> (-0.10%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...ache/pulsar/broker/namespace/NamespaceService.java	`74.35% <ø> (+0.30%)`	⬆️
...rg/apache/pulsar/broker/service/BrokerService.java	`83.02% <80.00%> (-0.49%)`	⬇️

... and 85 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

…wnership (#24780) (cherry picked from commit 46a76e9)

…wnership (apache#24780)

…wnership (apache#24780) (cherry picked from commit 46a76e9) (cherry picked from commit 05d88f7)

BewareMyPower self-assigned this Sep 25, 2025

BewareMyPower added area/broker type/refactor Code or documentation refactors. e.g. refactor code structure or methods to improve code readability release/4.1.2 release/4.0.8 labels Sep 25, 2025

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Sep 25, 2025

BewareMyPower requested review from Demogorgon314, Technoboy-, codelipenghui, lhotari, nodece, poorbarcode and shibd September 25, 2025 04:57

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsO…

03ee056

…wnership

BewareMyPower force-pushed the bewaremypower/ownership-check-cleanup branch from da543d0 to 03ee056 Compare September 25, 2025 05:02

Demogorgon314 approved these changes Sep 25, 2025

View reviewed changes

Technoboy- added the ready-to-test label Sep 25, 2025

Technoboy- added this to the 4.2.0 milestone Sep 25, 2025

Technoboy- closed this Sep 25, 2025

Technoboy- reopened this Sep 25, 2025

BewareMyPower mentioned this pull request Sep 25, 2025

[improve][broker] Improve the topic loading observability and performance #24577

Closed

4 tasks

lhotari approved these changes Sep 25, 2025

View reviewed changes

nodece approved these changes Sep 25, 2025

View reviewed changes

BewareMyPower merged commit 46a76e9 into apache:master Sep 25, 2025
70 of 73 checks passed

BewareMyPower deleted the bewaremypower/ownership-check-cleanup branch September 25, 2025 09:02

lhotari added the cherry-picked/branch-4.1 label Oct 4, 2025

lhotari pushed a commit that referenced this pull request Oct 4, 2025

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsO…

71903c4

…wnership (#24780) (cherry picked from commit 46a76e9)

lhotari pushed a commit that referenced this pull request Oct 8, 2025

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsO…

05d88f7

…wnership (#24780) (cherry picked from commit 46a76e9)

lhotari added the cherry-picked/branch-4.0 label Oct 8, 2025

walkinggo pushed a commit to walkinggo/pulsar that referenced this pull request Oct 8, 2025

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsO…

415876d

…wnership (apache#24780)

manas-ctds pushed a commit to datastax/pulsar that referenced this pull request Oct 15, 2025

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsO…

64cf82c

…wnership (apache#24780) (cherry picked from commit 46a76e9) (cherry picked from commit 05d88f7)

srinath-ctds pushed a commit to datastax/pulsar that referenced this pull request Nov 6, 2025

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsO…

e4c4a35

…wnership (apache#24780) (cherry picked from commit 46a76e9) (cherry picked from commit 05d88f7)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsOwnership #24780

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsOwnership #24780

Uh oh!

BewareMyPower commented Sep 25, 2025

Uh oh!

lhotari left a comment

Uh oh!

codecov-commenter commented Sep 25, 2025

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

	Optional<CompletableFuture<OwnedBundle>> optionalFuture = ownershipCache.getOwnedBundleAsync(bundle);
	if (optionalFuture.isEmpty()) {
	return CompletableFuture.completedFuture(false);
	}
	return optionalFuture.get().thenApply(ob -> ob != null && ob.isActive());

	public CompletableFuture<Boolean> checkOwnershipAsync(NamespaceBundle bundle) {
	Optional<CompletableFuture<OwnedBundle>> ownedBundleFuture = getOwnedBundleAsync(bundle);
	if (!ownedBundleFuture.isPresent()) {
	return CompletableFuture.completedFuture(false);
	}
	return ownedBundleFuture.get()
	.thenApply(bd -> bd != null && bd.isActive());

	public CompletableFuture<Boolean> checkBundleOwnership(TopicName topicName, NamespaceBundle bundle) {
	if (ExtensibleLoadManagerImpl.isLoadManagerExtensionEnabled(pulsar)) {
	// TODO: Add unit tests cover it.
	return loadManager.get().checkOwnershipAsync(Optional.of(topicName), bundle);
	} else {
	return ownershipCache.checkOwnershipAsync(bundle);
	}

	String msg = String.format("Namespace is being unloaded, cannot add topic %s", topic);
	log.warn(msg);
	pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
	topicFuture.completeExceptionally(new ServiceUnitNotReadyException(msg));

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsOwnership #24780

[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsOwnership #24780

Uh oh!

Conversation

BewareMyPower commented Sep 25, 2025

Motivation

Modifications

Verifying this change

Documentation

Matching PR in forked repository

Uh oh!

lhotari left a comment

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Sep 25, 2025

Codecov Report

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants