[Enhancement][client][schema] Remove synchronous calls in asynchronous callback
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
### Search before asking
- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar.
### Motivation
#23378 CI failed, which reports the deadlock issue in the pulsar client.
The binary lookup uses the pulsar client's internal executor.
```
2024-10-09T14:07:01.8281866Z "broker-client-shared-internal-executor-203-1" #398 [4208] prio=5 os_prio=0 cpu=5.08ms elapsed=3422.76s tid=0x00007efce4013ce0 nid=4208 waiting on condition [0x00007efca57fd000]
2024-10-09T14:07:01.8282015Z java.lang.Thread.State: WAITING (parking)
2024-10-09T14:07:01.8282244Z at jdk.internal.misc.Unsafe.park(java.base@21.0.4/Native Method)
2024-10-09T14:07:01.8282658Z - parking to wait for <0x000010000c5f6c58> (a java.util.concurrent.CompletableFuture$Signaller)
2024-10-09T14:07:01.8283361Z at java.util.concurrent.locks.LockSupport.park(java.base@21.0.4/LockSupport.java:221)
2024-10-09T14:07:01.8283798Z at java.util.concurrent.CompletableFuture$Signaller.block(java.base@21.0.4/CompletableFuture.java:1864)
2024-10-09T14:07:01.8284421Z at java.util.concurrent.ForkJoinPool.unmanagedBlock(java.base@21.0.4/ForkJoinPool.java:3780)
2024-10-09T14:07:01.8284853Z at java.util.concurrent.ForkJoinPool.managedBlock(java.base@21.0.4/ForkJoinPool.java:3725)
2024-10-09T14:07:01.8285257Z at java.util.concurrent.CompletableFuture.waitingGet(java.base@21.0.4/CompletableFuture.java:1898)
2024-10-09T14:07:01.8285614Z at java.util.concurrent.CompletableFuture.get(java.base@21.0.4/CompletableFuture.java:2072)
2024-10-09T14:07:01.8286326Z at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.getSchemaInfoByVersion(AbstractMultiVersionReader.java:118)
2024-10-09T14:07:01.8286927Z at org.apache.pulsar.client.impl.schema.reader.MultiVersionAvroReader.loadReader(MultiVersionAvroReader.java:45)
2024-10-09T14:07:01.8287435Z at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader$1.load(AbstractMultiVersionReader.java:51)
2024-10-09T14:07:01.8287942Z at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader$1.load(AbstractMultiVersionReader.java:48)
2024-10-09T14:07:01.8288307Z at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3571)
2024-10-09T14:07:01.8288582Z at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2313)
2024-10-09T14:07:01.8288898Z at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2190)
2024-10-09T14:07:01.8289291Z - locked <0x000010000c5f4f80> (a com.google.common.cache.LocalCache$StrongAccessEntry)
2024-10-09T14:07:01.8289762Z at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2080)
2024-10-09T14:07:01.8289989Z at com.google.common.cache.LocalCache.get(LocalCache.java:4012)
2024-10-09T14:07:01.8290251Z at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:4035)
2024-10-09T14:07:01.8290554Z at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:5013)
2024-10-09T14:07:01.8291133Z at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.getSchemaReader(AbstractMultiVersionReader.java:82)
2024-10-09T14:07:01.8291642Z at org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.read(AbstractMultiVersionReader.java:73)
2024-10-09T14:07:01.8292052Z at org.apache.pulsar.client.impl.schema.AbstractStructSchema.decode(AbstractStructSchema.java:90)
2024-10-09T14:07:01.8292382Z at org.apache.pulsar.client.impl.MessageImpl.decodeBySchema(MessageImpl.java:512)
2024-10-09T14:07:01.8292665Z at org.apache.pulsar.client.impl.MessageImpl.decode(MessageImpl.java:493)
2024-10-09T14:07:01.8292959Z at org.apache.pulsar.client.impl.MessageImpl.getValue(MessageImpl.java:478)
2024-10-09T14:07:01.8293681Z at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.refreshTopicPoliciesCache(SystemTopicBasedTopicPoliciesService.java:523)
2024-10-09T14:07:01.8294463Z at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.lambda$readMorePoliciesAsync$24(SystemTopicBasedTopicPoliciesService.java:500)
2024-10-09T14:07:01.8294991Z at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService$$Lambda/0x00007efd04b54f00.accept(Unknown Source)
2024-10-09T14:07:01.8295396Z at java.util.concurrent.CompletableFuture$UniAccept.tryFire(java.base@21.0.4/CompletableFuture.java:718)
2024-10-09T14:07:01.8295796Z at java.util.concurrent.CompletableFuture.postComplete(java.base@21.0.4/CompletableFuture.java:510)
2024-10-09T14:07:01.8296257Z at java.util.concurrent.CompletableFuture.complete(java.base@21.0.4/CompletableFuture.java:2179)
2024-10-09T14:07:01.8296677Z at org.apache.pulsar.client.impl.ConsumerBase.lambda$completePendingReceive$0(ConsumerBase.java:333)
2024-10-09T14:07:01.8297062Z at org.apache.pulsar.client.impl.ConsumerBase$$Lambda/0x00007efd04bb29a8.run(Unknown Source)
2024-10-09T14:07:01.8297599Z at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@21.0.4/ThreadPoolExecutor.java:1144)
2024-10-09T14:07:01.8298124Z at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@21.0.4/ThreadPoolExecutor.java:642)
2024-10-09T14:07:01.8298487Z at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
2024-10-09T14:07:01.8298698Z at java.lang.Thread.runWith(java.base@21.0.4/Thread.java:1596)
2024-10-09T14:07:01.8298883Z at java.lang.Thread.run(java.base@21.0.4/Thread.java:1583)
2024-10-09T14:07:01.8298896Z
2024-10-09T14:07:01.8299012Z Locked ownable synchronizers:
2024-10-09T14:07:01.8299347Z - <0x000010000c50c540> (a java.util.concurrent.ThreadPoolExecutor$Worker)
```
We are using synchronous calls in asynchronous callback.
### Solution
## Solution 1:
1. Add a set of async methods:
- `org.apache.pulsar.client.api.schema.SchemaReader#read`
- `org.apache.pulsar.client.api.Message#getValue`
2. Check the call stack, and then remove synchronous calls in asynchronous callback.
This solution will change the method signature, which breaks the public interface(SchemaReader and Message).
## Solution 2:
```
private void readMorePoliciesAsync(SystemTopicClient.Reader reader) {
if (closed.get()) {
cleanCacheAndCloseReader(reader.getSystemTopic().getTopicName().getNamespaceObject(), false);
return;
}
reader.readNextAsync()
.thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
notifyListener(msg);
})
```
It looks like refreshTopicPoliciesCache has a deadlock, and the callback thread is broker-client-shared-internal-executor.
We can use the `pulsarService.getExecutor()` to run the callback, this can avoid the deadlock.
## Solution 3:
Don't use the pulsar client's internal executor in the `org.apache.pulsar.client.impl.BinaryProtoLookupService`.
### Alternatives
_No response_
### Anything else?
_No response_
### Are you willing to submit a PR?
- [ ] I'm willing to submit a PR!
Contributor guide
Research direction
Start by tracing the callback from SystemTopicBasedTopicPoliciesService.readMorePoliciesAsync and refreshTopicPoliciesCache, then inspect AbstractMultiVersionReader and BinaryProtoLookupService alongside the reported deadlock stack. Compare the proposed executor and async-interface approaches; done means the asynchronous callback no longer blocks the Pulsar client's internal executor without breaking required behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, distributed-systems
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100