apache / apache/pulsar

[Enhancement][client][schema] Remove synchronous calls in asynchronous callback

Open
#23,432 0 comments 0 reactions 0 assignees View on GitHub
type/enhancement
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.