[fix][broker] Fix ExtensibleLoadManager flaky test

[Demogorgon314]

Fixes #20618 #20157

Motivation

jstack:

"metadata-store-9-1" #26 prio=5 os_prio=0 cpu=227.69ms elapsed=109.96s tid=0x000056447f5a14c0 nid=0xec waiting on condition  [0x00007f9db07d0000]
   java.lang.Thread.State: TIMED_WAITING (parking)
at jdk.internal.misc.Unsafe.park(java.base@17.0.7/Native Method)
- parking to wait for  <0x00001000048e8830> (a java.util.concurrent.CompletableFuture$Signaller)
at java.util.concurrent.locks.LockSupport.parkNanos(java.base@17.0.7/LockSupport.java:252)
at java.util.concurrent.CompletableFuture$Signaller.block(java.base@17.0.7/CompletableFuture.java:1866)
at java.util.concurrent.ForkJoinPool.unmanagedBlock(java.base@17.0.7/ForkJoinPool.java:3463)
at java.util.concurrent.ForkJoinPool.managedBlock(java.base@17.0.7/ForkJoinPool.java:3434)
at java.util.concurrent.CompletableFuture.timedGet(java.base@17.0.7/CompletableFuture.java:1939)
at java.util.concurrent.CompletableFuture.get(java.base@17.0.7/CompletableFuture.java:2095)
at org.apache.pulsar.broker.loadbalance.impl.LoadManagerShared.filterAntiAffinityGroupOwnedBrokers(LoadManagerShared.java:403)
at org.apache.pulsar.broker.loadbalance.extensions.policies.AntiAffinityGroupPolicyHelper.filter(AntiAffinityGroupPolicyHelper.java:46)
at org.apache.pulsar.broker.loadbalance.extensions.filter.AntiAffinityGroupPolicyFilter.filter(AntiAffinityGroupPolicyFilter.java:43)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.lambda$selectAsync$16(ExtensibleLoadManagerImpl.java:494)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl$$Lambda$933/0x000000080187dfb8.apply(Unknown Source)
at java.util.concurrent.CompletableFuture.uniComposeStage(java.base@17.0.7/CompletableFuture.java:1187)
at java.util.concurrent.CompletableFuture.thenCompose(java.base@17.0.7/CompletableFuture.java:2309)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.selectAsync(ExtensibleLoadManagerImpl.java:480)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.selectAsync(ExtensibleLoadManagerImpl.java:473)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.lambda$getOwnerAsync$9(ExtensibleLoadManagerImpl.java:396)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl$$Lambda$834/0x0000000801738890.apply(Unknown Source)
at java.util.concurrent.CompletableFuture.uniComposeStage(java.base@17.0.7/CompletableFuture.java:1187)
at java.util.concurrent.CompletableFuture.thenCompose(java.base@17.0.7/CompletableFuture.java:2309)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.getOwnerAsync(ExtensibleLoadManagerImpl.java:388)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.lambda$assign$7(ExtensibleLoadManagerImpl.java:380)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl$$Lambda$656/0x00000008016a3b20.apply(Unknown Source)
at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.put(ConcurrentOpenHashMap.java:409)
at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.computeIfAbsent(ConcurrentOpenHashMap.java:243)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.dedupeLookupRequest(ExtensibleLoadManagerImpl.java:461)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerImpl.assign(ExtensibleLoadManagerImpl.java:374)
at org.apache.pulsar.broker.loadbalance.extensions.ExtensibleLoadManagerWrapper.findBrokerServiceUrl(ExtensibleLoadManagerWrapper.java:66)
at org.apache.pulsar.broker.namespace.NamespaceService.lambda$getBrokerServiceUrlAsync$0(NamespaceService.java:198)
at org.apache.pulsar.broker.namespace.NamespaceService$$Lambda$741/0x00000008016fd8d8.apply(Unknown Source)
at java.util.concurrent.CompletableFuture.uniComposeStage(java.base@17.0.7/CompletableFuture.java:1187)
at java.util.concurrent.CompletableFuture.thenCompose(java.base@17.0.7/CompletableFuture.java:2309)
at org.apache.pulsar.broker.namespace.NamespaceService.lambda$getBrokerServiceUrlAsync$1(NamespaceService.java:191)
at org.apache.pulsar.broker.namespace.NamespaceService$$Lambda$740/0x00000008016fd690.apply(Unknown Source)
at java.util.concurrent.CompletableFuture.uniComposeStage(java.base@17.0.7/CompletableFuture.java:1187)
at java.util.concurrent.CompletableFuture.thenCompose(java.base@17.0.7/CompletableFuture.java:2309)
at org.apache.pulsar.broker.namespace.NamespaceService.getBrokerServiceUrlAsync(NamespaceService.java:189)
at org.apache.pulsar.broker.lookup.TopicLookupBase.lambda$internalLookupTopicAsync$6(TopicLookupBase.java:85)
at org.apache.pulsar.broker.lookup.TopicLookupBase$$Lambda$927/0x000000080187d228.apply(Unknown Source)
at java.util.concurrent.CompletableFuture$UniCompose.tryFire(java.base@17.0.7/CompletableFuture.java:1150)
at java.util.concurrent.CompletableFuture.postComplete(java.base@17.0.7/CompletableFuture.java:510)
at java.util.concurrent.CompletableFuture.complete(java.base@17.0.7/CompletableFuture.java:2147)
at org.apache.pulsar.metadata.impl.ZKMetadataStore.lambda$existsFromStore$9(ZKMetadataStore.java:349)
at org.apache.pulsar.metadata.impl.ZKMetadataStore$$Lambda$279/0x000000080138a950.run(Unknown Source)
at java.util.concurrent.Executors$RunnableAdapter.call(java.base@17.0.7/Executors.java:539)
at java.util.concurrent.FutureTask.run(java.base@17.0.7/FutureTask.java:264)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(java.base@17.0.7/ScheduledThreadPoolExecutor.java:304)
at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.7/ThreadPoolExecutor.java:1136)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.7/ThreadPoolExecutor.java:635)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(java.base@17.0.7/Thread.java:833)

Modifications

Documentation

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

[Demogorgon314]

Waiting for the Demogorgon314#14 test pass.