metadata-store thread should not execute onPoliciesUpdate synchronized when PoliciesUpdate
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
**Describe the bug**
A clear and concise description of what the bug is.
**To Reproduce**
Steps to reproduce the behavior:
loadtest the admin api
**Expected behavior**
A clear and concise description of what you expected to happen.
metadata-store will not be blocked
**Screenshots**
If applicable, add screenshots to help explain your problem.
```
"metadata-store-6-1" #174 prio=5 os_prio=0 cpu=336008.23ms elapsed=14968.31s tid=0x00007fdef0005000 nid=0x5680 waiting on condition [0x00007fdcd33f1000]
java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for <0x00007fdfdc100c00> (a java.util.concurrent.locks.ReentrantLock$NonfairSync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireQueued(AbstractQueuedSynchronizer.java:870)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquire(AbstractQueuedSynchronizer.java:1199)
at java.util.concurrent.locks.ReentrantLock$NonfairSync.lock(ReentrantLock.java:209)
at java.util.concurrent.locks.ReentrantLock.lock(ReentrantLock.java:285)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.offer(ScheduledThreadPoolExecutor.java:1010)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.add(ScheduledThreadPoolExecutor.java:1037)
at java.util.concurrent.ScheduledThreadPoolExecutor$DelayedWorkQueue.add(ScheduledThreadPoolExecutor.java:809)
at java.util.concurrent.ScheduledThreadPoolExecutor.delayedExecute(ScheduledThreadPoolExecutor.java:328)
at java.util.concurrent.ScheduledThreadPoolExecutor.scheduleAtFixedRate(ScheduledThreadPoolExecutor.java:573)
at org.apache.pulsar.common.util.RateLimiter.createTask(RateLimiter.java:259)
at org.apache.pulsar.common.util.RateLimiter.setRate(RateLimiter.java:238)
- locked <0x00007fe0d14eb950> (a org.apache.pulsar.common.util.RateLimiter)
at org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.updateDispatchRate(DispatchRateLimiter.java:398)
- locked <0x00007fe0d14eb8a8> (a org.apache.pulsar.broker.service.persistent.DispatchRateLimiter)
at org.apache.pulsar.broker.service.persistent.DispatchRateLimiter.onPoliciesUpdate(DispatchRateLimiter.java:286)
at org.apache.pulsar.broker.service.persistent.PersistentTopic.lambda$null$113(PersistentTopic.java:2632)
at org.apache.pulsar.broker.service.persistent.PersistentTopic$$Lambda$1702/1325515424.accept(Unknown Source)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.pulsar.broker.service.persistent.PersistentTopic.lambda$null$114(PersistentTopic.java:2632)
at org.apache.pulsar.broker.service.persistent.PersistentTopic$$Lambda$1696/1099205637.run(Unknown Source)
at java.util.concurrent.CompletableFuture.uniRun(CompletableFuture.java:719)
at java.util.concurrent.CompletableFuture.uniRunStage(CompletableFuture.java:731)
at java.util.concurrent.CompletableFuture.thenRun(CompletableFuture.java:2023)
at org.apache.pulsar.broker.service.persistent.PersistentTopic.lambda$null$115(PersistentTopic.java:2625)
at org.apache.pulsar.broker.service.persistent.PersistentTopic$$Lambda$1693/1143110949.accept(Unknown Source)
at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:544)
at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:272)
at org.apache.pulsar.broker.service.persistent.PersistentTopic.lambda$onPoliciesUpdate$118(PersistentTopic.java:2622)
at org.apache.pulsar.broker.service.persistent.PersistentTopic$$Lambda$1692/39951688.apply(Unknown Source)
at java.util.concurrent.CompletableFuture.uniComposeStage(CompletableFuture.java:995)
at java.util.concurrent.CompletableFuture.thenCompose(CompletableFuture.java:2137)
at org.apache.pulsar.broker.service.persistent.PersistentTopic.onPoliciesUpdate(PersistentTopic.java:2620)
at org.apache.pulsar.broker.service.BrokerService.lambda$null$91(BrokerService.java:1980)
at org.apache.pulsar.broker.service.BrokerService$$Lambda$1690/1978241112.accept(Unknown Source)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.pulsar.broker.service.BrokerService.lambda$null$92(BrokerService.java:1980)
at org.apache.pulsar.broker.service.BrokerService$$Lambda$1689/1018119492.accept(Unknown Source)
at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:670)
at java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:683)
at java.util.concurrent.CompletableFuture.thenAccept(CompletableFuture.java:2010)
at org.apache.pulsar.broker.service.BrokerService.lambda$null$93(BrokerService.java:1975)
at org.apache.pulsar.broker.service.BrokerService$$Lambda$1670/803871005.accept(Unknown Source)
at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap$Section.forEach(ConcurrentOpenHashMap.java:544)
at org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap.forEach(ConcurrentOpenHashMap.java:272)
at org.apache.pulsar.broker.service.BrokerService.lambda$handlePoliciesUpdates$94(BrokerService.java:1971)
at org.apache.pulsar.broker.service.BrokerService$$Lambda$1669/28868178.accept(Unknown Source)
at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:670)
at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:646)
at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:488)
at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1975)
at org.apache.pulsar.metadata.impl.ZKMetadataStore.lambda$null$7(ZKMetadataStore.java:139)
at org.apache.pulsar.metadata.impl.ZKMetadataStore$$Lambda$112/660057515.run(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:750)
```
**Desktop (please complete the following information):**
- OS: [e.g. iOS]
**Additional context**
Add any other context about the problem here.
2.9
Contributor guide
Research direction
Start with PersistentTopic.onPoliciesUpdate and DispatchRateLimiter.onPoliciesUpdate, using the supplied thread dump to trace the path through BrokerService and ZKMetadataStore. Reproduce the admin API load test described in the issue and compare the metadata-store thread state. Done means the reported blocking behavior is addressed and covered by an appropriate broker test.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100