[fix][broker] Fix unloadNamespaceBundlesGracefully can be stuck with extensible load manager

[BewareMyPower]

Motivation

I observed an issue that broker was stuck at close for a long time. It's stuck at BrokerService#unloadNamespaceBundlesGracefully, which calls disableBroker once and unloadNamespaceBundleAsync for all owned namespace bundles synchronously. Most issues happen when the broker is the last broker.

Issue 1: Free events won't be sent in overrideOwnership

In overrideOwnership, if no broker is available, a Free event will be created.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

Line 1363 in 4ce0c75

}).orElseGet(() -> new ServiceUnitStateData(Free, null,

However, since the dstBroker and sourceBroker fields are null in the Free event, an exception will be thrown so that the Free event won't be created and sent.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateData.java

Line 37 in 4ce0c75

if (StringUtils.isBlank(dstBroker) && StringUtils.isBlank(sourceBroker)) {

Issue 2: Free events could be skipped due to the same version id

The Free event is created in overrideOwnership based on a previous event on the same bundle from the table view. However, there might be inflight events that are not in the table view yet. In ServiceUnitStateDataConflictResolver#shouldKeepLeft, if the version id is the same, the Free event will be skipped. Then, if the last event is the Owned event whose target broker is the current broker in close, waitForCleanups will wait until the timeout exceeds.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/channel/ServiceUnitStateChannelImpl.java

Lines 1394 to 1396 in 4ce0c75

	if (data.state() == Owned && broker.equals(data.dstBroker())) {
	cleaned = false;
	break;

Issue 3: __change_events topic prevents waitForCleanups from finishing

The __change_events topic's reader, which is managed by the system topic based topic policies service, will try acquire the ownership. So that in waitForCleanups, there will always be a Owned event for this topic's bundle. If the target broker is the broker itself, waitForCleanups will never have a chance to exit until the timeout exceeds.

Issue 4: unloadNamespaceBundleAsync will be stuck at getOwnershipAsync if there is no available broker

Broker unregisters itself in disableBroker, if it's the last broker, then no brokers will be available after that. However, unloadNamespaceBundleAsync needs to publish a Unloaded event in unloadAsync.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java

Lines 700 to 701 in 4ce0c75

	CompletableFuture<Void> future = serviceUnitStateChannel.publishUnloadEventAsync(unload);
	return unloadManager.waitAsync(future, unload.serviceUnit(), unloadDecision, timeout, timeoutUnit)

Modifications

• issue 1: change the constructor of ServiceUnitStateData to all null dstBroker and sourceBroker for Free events.
• issue 2: call TableView#refreshAsync to refresh the entry set in ServiceUnitStateTableViewImpl#flush and call it before overrideOwnership. Besides, skip the version id check for the Free event.
• issue 3: close the topic policies service at the beginning of PulsarService#closeAsync.
• issue 4: close the extensible load manager if there is no available broker after unregister

Add testLookup to cover the changes above.

Documentation

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

Matching PR in forked repository

PR in forked repository:

[lhotari]

LGTM

[BewareMyPower]

This fix might change some behaviors, I will try to fix the root cause