wuzhanpeng opened a new pull request #12202:
URL: https://github.com/apache/pulsar/pull/12202
### Motivation
In our production environment, when the broker receives a large number of
`PRODUCER` requests in a short period of time, we have observed that the broker
will have a loop waiting problem when handling these requests, which will cause
the broker to get stuck in severe cases and cause a huge amonut of other
requests to time out.
To simplify the description of the problem, we assume that multiple
producers initiate `PRODUCER` requests to a topic at the same time. As we can
see in `AbstractTopic#addProducer`, we can simplify the process so that the
process of `addProducer` in each `PRODUCER` request is broken down into:
1. acquire the lock in thread#1
2. load data from zk in thread#2 with timeout (i.e.
`AbstractTopic#isProducersExceeded` in internal adding producer)
3. return back to thread#1 to release the lock
It should be noted that these 3 processes are serial.
Assuming that the core size of the thread pool(actually is
`ForkJoinPool.commonPool()`) that processes the above threads is only 1, and
only one thread can successfully obtain the lock(`AbstractTopic#lock`) in the
simultaneous `PRODUCER` requests, the remaining threads must be queued in the
submission queue of the thread pool. Unfortunately, there is a high probability
that thread#2 will be put into the queue waiting for scheduling. In this
situation, the thread#1 that acquired the lock cannot complete because it needs
to wait for the thread#2, and the other threads that have not acquired the lock
need to acquire the lock first. This process cannot continue until the thread#2
times out and throws an exception.
For jstack result, we can easily see
```
"ForkJoinPool.commonPool-worker-110" #482 daemon prio=5 os_prio=0
tid=0x00007fd714021000 nid=0x61a3 waiting on condition [0x00007fd562772000]
java.lang.Thread.State: TIMED_WAITING (parking)
at jdk.internal.misc.Unsafe.park([email protected]/Native Method)
- parking to wait for <0x00000006284caad0> (a
java.util.concurrent.CompletableFuture$Signaller)
at
java.util.concurrent.locks.LockSupport.parkNanos([email protected]/LockSupport.java:234)
at
java.util.concurrent.CompletableFuture$Signaller.block([email protected]/CompletableFuture.java:1798)
at
java.util.concurrent.ForkJoinPool.managedBlock([email protected]/ForkJoinPool.java:3146)
at
java.util.concurrent.CompletableFuture.timedGet([email protected]/CompletableFuture.java:1868)
at
java.util.concurrent.CompletableFuture.get([email protected]/CompletableFuture.java:2021)
at
org.apache.pulsar.zookeeper.ZooKeeperDataCache.get(ZooKeeperDataCache.java:97)
at
org.apache.pulsar.broker.service.AbstractTopic.isProducersExceeded(AbstractTopic.java:156)
at
org.apache.pulsar.broker.service.AbstractTopic.internalAddProducer(AbstractTopic.java:629)
at
org.apache.pulsar.broker.service.AbstractTopic.lambda$addProducer$8(AbstractTopic.java:405)
at
org.apache.pulsar.broker.service.AbstractTopic$$Lambda$1433/1422007940.accept(Unknown
Source)
at
java.util.concurrent.CompletableFuture.uniAcceptNow([email protected]/CompletableFuture.java:753)
at
java.util.concurrent.CompletableFuture.uniAcceptStage([email protected]/CompletableFuture.java:731)
at
java.util.concurrent.CompletableFuture.thenAccept([email protected]/CompletableFuture.java:2108)
at
org.apache.pulsar.broker.service.AbstractTopic.addProducer(AbstractTopic.java:392)
at
org.apache.pulsar.broker.service.persistent.PersistentTopic.addProducer(PersistentTopic.java:540)
at
org.apache.pulsar.broker.service.ServerCnx.lambda$null$22(ServerCnx.java:1233)
at
org.apache.pulsar.broker.service.ServerCnx$$Lambda$1428/932296811.accept(Unknown
Source)
at
java.util.concurrent.CompletableFuture$UniAccept.tryFire([email protected]/CompletableFuture.java:714)
at
java.util.concurrent.CompletableFuture.postComplete([email protected]/CompletableFuture.java:506)
at
java.util.concurrent.CompletableFuture.complete([email protected]/CompletableFuture.java:2073)
at
org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.lambda$null$6(BookkeeperSchemaStorage.java:217)
at
org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage$$Lambda$1421/1611023719.apply(Unknown
Source)
at
java.util.concurrent.CompletableFuture.uniHandle([email protected]/CompletableFuture.java:930)
at
java.util.concurrent.CompletableFuture$UniHandle.tryFire([email protected]/CompletableFuture.java:907)
at
java.util.concurrent.CompletableFuture$Completion.exec([email protected]/CompletableFuture.java:479)
at
java.util.concurrent.ForkJoinTask.doExec([email protected]/ForkJoinTask.java:290)
at
java.util.concurrent.ForkJoinPool.runWorker([email protected]/ForkJoinPool.java:1603)
at
java.util.concurrent.ForkJoinWorkerThread.run([email protected]/ForkJoinWorkerThread.java:177)
Locked ownable synchronizers:
- <0x0000000624e2e9a0> (a
java.util.concurrent.locks.ReentrantReadWriteLock$NonfairSync)
```
To make matters worse, in the default configuration, both zk timeout and
client operation timeout are 30 seconds. This will cause each retry request to
end with a timeout and iteratively.
### Modifications
This problem is so hidden that it is very difficult to detect, and even we
still have no way to reproduce it in the test environment. However, in our
production environment, this problem is more likely to occur if the bundle
re-load or broker restart operation is triggered frequently(This phenomenon may
be more obvious in our production scenarios. Each of our independent topics may
have thousands of producers). Once the problem occurs in the cluster, there
will be a lot of operation timeout exceptions.
Below we give a solution to reduce the use of locks, because we think that
for the conventional production model, it is sufficient to use read locks when
adding producers in `Shared` mode.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
