[fix][broker]Fix ConcurrentModificationException when ModularLoadManagerImpl start

[AnonHxy]

Fixes #16952

Motivation

• Fix ConcurrentModificationException when ModularLoadManagerImpl start. The root cause is that ModularLoadManagerImpl#knownBrokers can be modified in multi-thread through ModularLoadManagerImpl#updateAll
  • Thread-1 is line290:
    

    pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

    Lines 282 to 292 in 5c09d53

    	public void handleDataNotification(Notification t) {
    	if (t.getPath().startsWith(LoadManager.LOADBALANCE_BROKERS_ROOT)) {
    	brokersData.listLocks(LoadManager.LOADBALANCE_BROKERS_ROOT)
    	.thenAccept(brokers -> {
    	reapDeadBrokerPreallocations(brokers);
    	});
    	try {
    	scheduler.execute(ModularLoadManagerImpl.this::updateAll);
    	} catch (RejectedExecutionException e) {
    	// Executor is shutting down

  • Thread-2 is line969
    

    pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

    Lines 938 to 970 in 5c09d53

    	public void start() throws PulsarServerException {
    	try {
    	// At this point, the ports will be updated with the real port number that the server was assigned
    	Map<String, String> protocolData = pulsar.getProtocolDataToAdvertise();
    	lastData = new LocalBrokerData(pulsar.getSafeWebServiceAddress(), pulsar.getWebServiceAddressTls(),
    	pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls(), pulsar.getAdvertisedListeners());
    	lastData.setProtocols(protocolData);
    	// configure broker-topic mode
    	lastData.setPersistentTopicsEnabled(pulsar.getConfiguration().isEnablePersistentTopics());
    	lastData.setNonPersistentTopicsEnabled(pulsar.getConfiguration().isEnableNonPersistentTopics());
    	localData = new LocalBrokerData(pulsar.getSafeWebServiceAddress(), pulsar.getWebServiceAddressTls(),
    	pulsar.getBrokerServiceUrl(), pulsar.getBrokerServiceUrlTls(), pulsar.getAdvertisedListeners());
    	localData.setProtocols(protocolData);
    	localData.setBrokerVersionString(pulsar.getBrokerVersion());
    	// configure broker-topic mode
    	localData.setPersistentTopicsEnabled(pulsar.getConfiguration().isEnablePersistentTopics());
    	localData.setNonPersistentTopicsEnabled(pulsar.getConfiguration().isEnableNonPersistentTopics());
    	String lookupServiceAddress = pulsar.getAdvertisedAddress() + ":"
    	+ (conf.getWebServicePort().isPresent() ? conf.getWebServicePort().get()
    	: conf.getWebServicePortTls().get());
    	brokerZnodePath = LoadManager.LOADBALANCE_BROKERS_ROOT + "/" + lookupServiceAddress;
    	final String timeAverageZPath = TIME_AVERAGE_BROKER_ZPATH + "/" + lookupServiceAddress;
    	updateLocalBrokerData();
    	brokerDataLock = brokersData.acquireLock(brokerZnodePath, localData).join();
    	timeAverageBrokerDataCache.readModifyUpdateOrCreate(timeAverageZPath,
    	__ -> new TimeAverageBrokerData()).join();
    	updateAll();
    	} catch (Exception e) {

Modifications

• Changing ModularLoadManagerImpl#knownBrokers to ConcurrentHashMap.newKeySet()

Verifying this change

• Make sure that the change passes the CI checks.

Documentation

Check the box below or label this PR directly.

Need to update docs?

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  (Please explain why)

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

[AnonHxy]

/pulsarbot run-failure-checks

[poorbarcode]

Maybe we can move this line below to method start and add some code comments, this could solves the concurrency problem.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/impl/ModularLoadManagerImpl.java

Line 249 in 7650612

pulsar.getLocalMetadataStore().registerListener(this::handleDataNotification);

[AnonHxy]

handleDataNotification also has other actions, so we'd better keep it in init method I think @poorbarcode

[mattisonchao]

@poorbarcode Could you do the final review?

[poorbarcode]

@poorbarcode Could you do the final review?

OK. Approved