[improve][broker] Replace isServiceUnitActiveAsync with checkTopicNsOwnership

[BewareMyPower]

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.

1. 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

2. 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:

[lhotari]

LGTM, thanks for taking actions to cleanup Pulsar code!

[codecov-commenter]

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.