[fix][broker] Throw AlreadyClosedException while request to a closed metadata store

[codelipenghui]

Motivation

After the metadata store is closed, the metadata store will fail all the pending requests, but the new request will be blocked forever for the batched-supported metadata store.

The root cause is we are using a separate thread to flush the requests to the metadata services. The thread will be shutdown after the metadata store has been closed. But the new requests can still add to the pending requests and will never be complete.

So this PR is to prevent the new request to a closed metadata store. The caller will get AlreadyClosedException if the requested metadata store is closed

It will also fix some flaky tests that stuck at the broker shutdown. Here is an example

2022-12-20T11:45:12.6521329Z "main" #1 prio=5 os_prio=0 cpu=134299.52ms elapsed=3504.19s tid=0x00007fa37c02ab80 nid=0xb20 waiting on condition  [0x00007fa383aed000]
2022-12-20T11:45:12.6521804Z    java.lang.Thread.State: WAITING (parking)
2022-12-20T11:45:12.6522810Z 	at jdk.internal.misc.Unsafe.park(java.base@17.0.5/Native Method)
2022-12-20T11:45:12.6523655Z 	- parking to wait for  <0x0000100010a0a8d8> (a java.util.concurrent.CompletableFuture$Signaller)
2022-12-20T11:45:12.6525332Z 	at java.util.concurrent.locks.LockSupport.park(java.base@17.0.5/LockSupport.java:211)
2022-12-20T11:45:12.6525816Z 	at java.util.concurrent.CompletableFuture$Signaller.block(java.base@17.0.5/CompletableFuture.java:1864)
2022-12-20T11:45:12.6526317Z 	at java.util.concurrent.ForkJoinPool.unmanagedBlock(java.base@17.0.5/ForkJoinPool.java:3463)
2022-12-20T11:45:12.6526984Z 	at java.util.concurrent.ForkJoinPool.managedBlock(java.base@17.0.5/ForkJoinPool.java:3434)
2022-12-20T11:45:12.6527493Z 	at java.util.concurrent.CompletableFuture.waitingGet(java.base@17.0.5/CompletableFuture.java:1898)
2022-12-20T11:45:12.6527987Z 	at java.util.concurrent.CompletableFuture.join(java.base@17.0.5/CompletableFuture.java:2117)
2022-12-20T11:45:12.6528626Z 	at org.apache.pulsar.metadata.coordination.impl.LockManagerImpl.close(LockManagerImpl.java:163)
2022-12-20T11:45:12.6529250Z 	at org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl.stop(ModularLoadManagerImpl.java:982)
2022-12-20T11:45:12.6529917Z 	at org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerWrapper.stop(ModularLoadManagerWrapper.java:118)
2022-12-20T11:45:12.6530566Z 	at org.apache.pulsar.functions.worker.PulsarFunctionTlsTest.tearDown(PulsarFunctionTlsTest.java:194)
2022-12-20T11:45:12.6531116Z 	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(java.base@17.0.5/Native Method)
2022-12-20T11:45:12.6531636Z 	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(java.base@17.0.5/NativeMethodAccessorImpl.java:77)
2022-12-20T11:45:12.6532225Z 	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(java.base@17.0.5/DelegatingMethodAccessorImpl.java:43)
2022-12-20T11:45:12.6532707Z 	at java.lang.reflect.Method.invoke(java.base@17.0.5/Method.java:568)
2022-12-20T11:45:12.6533198Z 	at org.testng.internal.invokers.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:139)
2022-12-20T11:45:12.6534052Z 	at org.testng.internal.invokers.MethodInvocationHelper.invokeMethodConsideringTimeout(MethodInvocationHelper.java:69)
2022-12-20T11:45:12.6535785Z 	at org.testng.internal.invokers.ConfigInvoker.invokeConfigurationMethod(ConfigInvoker.java:361)
2022-12-20T11:45:12.6551326Z 	at org.testng.internal.invokers.ConfigInvoker.invokeConfigurations(ConfigInvoker.java:296)
2022-12-20T11:45:12.6552044Z 	at org.testng.internal.invokers.TestInvoker.runConfigMethods(TestInvoker.java:823)
2022-12-20T11:45:12.6553196Z 	at org.testng.internal.invokers.TestInvoker.runAfterConfigurations(TestInvoker.java:792)
2022-12-20T11:45:12.6554483Z 	at org.testng.internal.invokers.TestInvoker.invokeMethod(TestInvoker.java:768)
2022-12-20T11:45:12.6556779Z 	at org.testng.internal.invokers.TestInvoker.invokeTestMethod(TestInvoker.java:221)
2022-12-20T11:45:12.6557545Z 	at org.testng.internal.invokers.MethodRunner.runInSequence(MethodRunner.java:50)
2022-12-20T11:45:12.6559390Z 	at org.testng.internal.invokers.TestInvoker$MethodInvocationAgent.invoke(TestInvoker.java:969)
2022-12-20T11:45:12.6561582Z 	at org.testng.internal.invokers.TestInvoker.invokeTestMethods(TestInvoker.java:194)
2022-12-20T11:45:12.6563985Z 	at org.testng.internal.invokers.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:148)
2022-12-20T11:45:12.6575879Z 	at org.testng.internal.invokers.TestMethodWorker.run(TestMethodWorker.java:128)
2022-12-20T11:45:12.6580949Z 	at org.testng.TestRunner$$Lambda$245/0x0000000800dc1238.accept(Unknown Source)
2022-12-20T11:45:12.6583537Z 	at java.util.ArrayList.forEach(java.base@17.0.5/ArrayList.java:1511)
2022-12-20T11:45:12.6587266Z 	at org.testng.TestRunner.privateRun(TestRunner.java:829)
2022-12-20T11:45:12.6592843Z 	at org.testng.TestRunner.run(TestRunner.java:602)
2022-12-20T11:45:12.6595493Z 	at org.testng.SuiteRunner.runTest(SuiteRunner.java:437)
2022-12-20T11:45:12.6600393Z 	at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:431)
2022-12-20T11:45:12.6602541Z 	at org.testng.SuiteRunner.privateRun(SuiteRunner.java:391)
2022-12-20T11:45:12.6602915Z 	at org.testng.SuiteRunner.run(SuiteRunner.java:330)
2022-12-20T11:45:12.6603303Z 	at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:52)
2022-12-20T11:45:12.6603715Z 	at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:95)
2022-12-20T11:45:12.6604088Z 	at org.testng.TestNG.runSuitesSequentially(TestNG.java:1256)
2022-12-20T11:45:12.6604477Z 	at org.testng.TestNG.runSuitesLocally(TestNG.java:1176)
2022-12-20T11:45:12.6604818Z 	at org.testng.TestNG.runSuites(TestNG.java:1099)
2022-12-20T11:45:12.6605333Z 	at org.testng.TestNG.run(TestNG.java:1067)
2022-12-20T11:45:12.6605934Z 	at org.apache.maven.surefire.testng.TestNGExecutor.run(TestNGExecutor.java:135)
2022-12-20T11:45:12.6606730Z 	at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.executeSingleClass(TestNGDirectoryTestSuite.java:112)
2022-12-20T11:45:12.6607859Z 	at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.executeLazy(TestNGDirectoryTestSuite.java:123)
2022-12-20T11:45:12.6608512Z 	at org.apache.maven.surefire.testng.TestNGDirectoryTestSuite.execute(TestNGDirectoryTestSuite.java:90)
2022-12-20T11:45:12.6609083Z 	at org.apache.maven.surefire.testng.TestNGProvider.invoke(TestNGProvider.java:146)
2022-12-20T11:45:12.6609829Z 	at org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
2022-12-20T11:45:12.6610433Z 	at org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
2022-12-20T11:45:12.6610950Z 	at org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
2022-12-20T11:45:12.6611422Z 	at org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

Documentation

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

Matching PR in forked repository

PR in forked repository: codelipenghui#20

[nodece]

/pulsarbot rerun-failure-checks

[codecov-commenter]

Codecov Report

Merging #19055 (4369fc3) into master (41edd2e) will increase coverage by 0.08%.
The diff coverage is 33.33%.

@@             Coverage Diff              @@
##             master   #19055      +/-   ##
============================================
+ Coverage     47.06%   47.14%   +0.08%     
- Complexity    10620    10622       +2     
============================================
  Files           709      710       +1     
  Lines         69423    69428       +5     
  Branches       7448     7450       +2     
============================================
+ Hits          32677    32735      +58     
+ Misses        33102    33026      -76     
- Partials       3644     3667      +23     

Flag	Coverage Δ
unittests	47.14% <33.33%> (+0.08%)	⬆️

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

Impacted Files	Coverage Δ
...n/java/org/apache/pulsar/broker/PulsarService.java	56.91% <33.33%> (-1.14%)	⬇️
...rvice/schema/KeyValueSchemaCompatibilityCheck.java	21.62% <0.00%> (-45.95%)	⬇️
...ker/loadbalance/impl/LeastLongTermMessageRate.java	84.44% <0.00%> (-8.89%)	⬇️
...ker/resourcegroup/ResourceGroupPublishLimiter.java	31.03% <0.00%> (-8.05%)	⬇️
...ker/resourcegroup/ResourceGroupConfigListener.java	57.53% <0.00%> (-6.85%)	⬇️
...balance/impl/SimpleResourceAllocationPolicies.java	48.57% <0.00%> (-5.72%)	⬇️
...sar/broker/service/schema/SchemaRegistryStats.java	71.25% <0.00%> (-3.75%)	⬇️
.../service/SystemTopicBasedTopicPoliciesService.java	69.81% <0.00%> (-3.36%)	⬇️
...sar/broker/loadbalance/impl/LoadManagerShared.java	40.35% <0.00%> (-3.08%)	⬇️
...rg/apache/pulsar/broker/service/AbstractTopic.java	62.06% <0.00%> (-3.00%)	⬇️
... and 34 more