[fix][broker]After the broker is restarted, the cache dynamic configuration is invalid

[lordcheng10]

Motivation

We update some cache dynamic configurations, such as: managedLedgerCacheSizeMB=100, but after we restarted the broker, we found that the dynamically modified cache configuration on zookeeper was invalid, and the value configured in broker.conf took effect.

After restarting, the loading process of config is as follows:

1. Load the broker.conf file and initialize the ServiceConfiguration object;

2. Use the ServiceConfiguration object to build the ManagedLedgerStorage object. When building the managedLedgerClientFactory object, the cache configuration in managedLedgerFactoryConfig is initialized through the conf object:
   

   pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

   Lines 742 to 748 in 4b757cf

   	managedLedgerClientFactory = ManagedLedgerStorage.create(
   	config, localMetadataStore,
   	bkClientFactory, ioEventLoopGroup
   	);
   	this.brokerService = newBrokerService(this);

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/ManagedLedgerClientFactory.java

Lines 59 to 60 in abff91f

	managedLedgerFactoryConfig.setMaxCacheSize(conf.getManagedLedgerCacheSizeMB() * 1024L * 1024L);
	managedLedgerFactoryConfig.setCacheEvictionWatermark(conf.getManagedLedgerCacheEvictionWatermark());

3. When creating the BrokerService object, it will read the dynamic configuration on zookeeper, update it to conf, and then register the listener for the relevant configuration update, but the execution of these listeners is not triggered:
   

   pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

   Lines 746 to 748 in 4b757cf

   	this.brokerService = newBrokerService(this);

   
   

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

   Lines 2217 to 2250 in 4b757cf

   	// (2) update ServiceConfiguration value by reading zk-configuration-map
   	updateDynamicServiceConfiguration();
   	// (3) Listener Registration
   	// add listener on "maxConcurrentLookupRequest" value change
   	registerConfigurationListener("maxConcurrentLookupRequest",
   	(maxConcurrentLookupRequest) -> lookupRequestSemaphore.set(
   	new Semaphore((int) maxConcurrentLookupRequest, false)));
   	// add listener on "maxConcurrentTopicLoadRequest" value change
   	registerConfigurationListener("maxConcurrentTopicLoadRequest",
   	(maxConcurrentTopicLoadRequest) -> topicLoadRequestSemaphore.set(
   	new Semaphore((int) maxConcurrentTopicLoadRequest, false)));
   	registerConfigurationListener("loadManagerClassName", className -> {
   	pulsar.getExecutor().execute(() -> {
   	try {
   	final LoadManager newLoadManager = LoadManager.create(pulsar);
   	log.info("Created load manager: {}", className);
   	pulsar.getLoadManager().get().stop();
   	newLoadManager.start();
   	pulsar.getLoadManager().set(newLoadManager);
   	} catch (Exception ex) {
   	log.warn("Failed to change load manager", ex);
   	}
   	});
   	});
   	// add listener to notify broker managedLedgerCacheSizeMB dynamic config
   	registerConfigurationListener("managedLedgerCacheSizeMB", (managedLedgerCacheSizeMB) -> {
   	managedLedgerFactory.getEntryCacheManager()
   	.updateCacheSizeAndThreshold(((int) managedLedgerCacheSizeMB) * 1024L * 1024L);
   	});
   	// add listener to notify broker managedLedgerCacheEvictionWatermark dynamic config

Therefore, after the broker is restarted, the cache-related configuration used is still the configuration in the broker.conf file.

Solution:
The configuration is loaded in the following order:

1. Register and configure the listener;
2. Read the dynamic configuration on zookeeper and trigger the corresponding listener;

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)

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[codelipenghui]

Could you please help add tests? We are able to restart the broker during the test.

[HQebupt]

Good catch 👍

[lordcheng10]

Could you please help add tests? We are able to restart the broker during the test.

@codelipenghui @HQebupt @Technoboy- Fixed, PTAL,thanks!

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[eolivelli]

Lgtm

This looks like a regression.
We should pick this into 2.11
@codelipenghui @zymap @lhotari

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[Jason918]

LGTM

[Jason918]

@lordcheng10 Please solve CI failure.

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

@lordcheng10 Please solve CI failure.

OK

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

/pulsarbot run-failure-checks

[lordcheng10]

@lordcheng10 Please solve CI failure.

Fixed,PTAL,thanks! @Jason918

[lordcheng10]

Could you please help add tests? We are able to restart the broker during the test.

@codelipenghui Fixed, PTAL,thanks!

[mattisonchao]

Hi @lordcheng10
It looks like we got many conflicts when cherry-picking it to branch-2.9.
Would you mind pushing a PR to branch-2.9? (To avoid cherry-picking involving bugs)