You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2021/07/01 14:29:32 UTC
[GitHub] [pulsar] codelipenghui opened a new pull request #11183: Fix the dead lock when using hasMessageAvailableAsync and readNextAsync
codelipenghui opened a new pull request #11183:
URL: https://github.com/apache/pulsar/pull/11183
The issue will happens after satisfying the following conditions:
1. The messages are added to the incoming queue before reading messages
2. The result future of the readNextAsync been complete before call future.whenComplete by users,
This won't always appear.
After that, since we are using the IO thread to call the callback of the hasMessageAvailableAsync,
so the IO thread will process the message.getValue(). Then might get a deadlock as followings:
```
jdk.internal.misc.Unsafe.park(boolean, long) Unsafe.java (native)
java.util.concurrent.locks.LockSupport.park(Object) LockSupport.java:194
java.util.concurrent.CompletableFuture$Signaller.block() CompletableFuture.java:1796
java.util.concurrent.ForkJoinPool.managedBlock(ForkJoinPool$ManagedBlocker) ForkJoinPool.java:3128
java.util.concurrent.CompletableFuture.waitingGet(boolean) CompletableFuture.java:1823
java.util.concurrent.CompletableFuture.get() CompletableFuture.java:1998
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.getSchemaInfoByVersion(byte[]) AbstractMultiVersionReader.java:115
org.apache.pulsar.client.impl.schema.reader.MultiVersionAvroReader.loadReader(BytesSchemaVersion) MultiVersionAvroReader.java:47
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader$1.load(BytesSchemaVersion) AbstractMultiVersionReader.java:52
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader$1.load(Object) AbstractMultiVersionReader.java:49
com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(Object, CacheLoader) LocalCache.java:3529
com.google.common.cache.LocalCache$Segment.loadSync(Object, int, LocalCache$LoadingValueReference, CacheLoader) LocalCache.java:2278
com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(Object, int, CacheLoader) LocalCache.java:2155
com.google.common.cache.LocalCache$Segment.get(Object, int, CacheLoader) LocalCache.java:2045
com.google.common.cache.LocalCache.get(Object, CacheLoader) LocalCache.java:3951
com.google.common.cache.LocalCache.getOrLoad(Object) LocalCache.java:3974
com.google.common.cache.LocalCache$LocalLoadingCache.get(Object) LocalCache.java:4935
org.apache.pulsar.client.impl.schema.reader.AbstractMultiVersionReader.read(byte[], byte[]) AbstractMultiVersionReader.java:86
org.apache.pulsar.client.impl.schema.AbstractStructSchema.decode(byte[], byte[]) AbstractStructSchema.java:60
org.apache.pulsar.client.impl.MessageImpl.getValue() MessageImpl.java:301
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.refreshTopicPoliciesCache(Message) SystemTopicBasedTopicPoliciesService.java:302
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.lambda$initPolicesCache$7(SystemTopicClient$Reader, Throwable, CompletableFuture, Message, Throwable) SystemTopicBasedTopicPoliciesService.java:254
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService$$Lambda$817.accept(Object, Object)
java.util.concurrent.CompletableFuture.uniWhenComplete(Object, BiConsumer, CompletableFuture$UniWhenComplete) CompletableFuture.java:859
java.util.concurrent.CompletableFuture.uniWhenCompleteStage(Executor, BiConsumer) CompletableFuture.java:883
java.util.concurrent.CompletableFuture.whenComplete(BiConsumer) CompletableFuture.java:2251
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.lambda$initPolicesCache$10(SystemTopicClient$Reader, CompletableFuture, Boolean, Throwable) SystemTopicBasedTopicPoliciesService.java:246
org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService$$Lambda$725.accept(Object, Object)
java.util.concurrent.CompletableFuture.uniWhenComplete(Object, BiConsumer, CompletableFuture$UniWhenComplete) CompletableFuture.java:859
java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(int) CompletableFuture.java:837
java.util.concurrent.CompletableFuture.postComplete() CompletableFuture.java:506
java.util.concurrent.CompletableFuture.complete(Object) CompletableFuture.java:2073
org.apache.pulsar.client.impl.ConsumerImpl.lambda$hasMessageAvailableAsync$48(CompletableFuture, MessageId) ConsumerImpl.java:2075
org.apache.pulsar.client.impl.ConsumerImpl$$Lambda$723.accept(Object)
java.util.concurrent.CompletableFuture$UniAccept.tryFire(int) CompletableFuture.java:714
java.util.concurrent.CompletableFuture.postComplete() CompletableFuture.java:506
java.util.concurrent.CompletableFuture.complete(Object) CompletableFuture.java:2073
org.apache.pulsar.client.impl.ConsumerImpl.lambda$internalGetLastMessageIdAsync$53(CompletableFuture, PulsarApi$CommandGetLastMessageIdResponse) ConsumerImpl.java:2187
org.apache.pulsar.client.impl.ConsumerImpl$$Lambda$720.accept(Object)
java.util.concurrent.CompletableFuture$UniAccept.tryFire(int) CompletableFuture.java:714
java.util.concurrent.CompletableFuture.postComplete() CompletableFuture.java:506
java.util.concurrent.CompletableFuture.complete(Object) CompletableFuture.java:2073
org.apache.pulsar.client.impl.ClientCnx.handleGetLastMessageIdSuccess(PulsarApi$CommandGetLastMessageIdResponse) ClientCnx.java:468
org.apache.pulsar.common.protocol.PulsarDecoder.channelRead(ChannelHandlerContext, Object) PulsarDecoder.java:342
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(Object) AbstractChannelHandlerContext.java:379
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext, Object) AbstractChannelHandlerContext.java:365
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(Object) AbstractChannelHandlerContext.java:357
io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ChannelHandlerContext, CodecOutputList, int) ByteToMessageDecoder.java:324
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ChannelHandlerContext, Object) ByteToMessageDecoder.java:296
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(Object) AbstractChannelHandlerContext.java:379
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext, Object) AbstractChannelHandlerContext.java:365
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(Object) AbstractChannelHandlerContext.java:357
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(ChannelHandlerContext, Object) DefaultChannelPipeline.java:1410
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(Object) AbstractChannelHandlerContext.java:379
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext, Object) AbstractChannelHandlerContext.java:365
io.netty.channel.DefaultChannelPipeline.fireChannelRead(Object) DefaultChannelPipeline.java:919
io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read() AbstractNioByteChannel.java:166
io.netty.channel.nio.NioEventLoop.processSelectedKey(SelectionKey, AbstractNioChannel) NioEventLoop.java:719
io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized() NioEventLoop.java:655
io.netty.channel.nio.NioEventLoop.processSelectedKeys() NioEventLoop.java:581
io.netty.channel.nio.NioEventLoop.run() NioEventLoop.java:493
io.netty.util.concurrent.SingleThreadEventExecutor$4.run() SingleThreadEventExecutor.java:989
io.netty.util.internal.ThreadExecutorMap$2.run() ThreadExecutorMap.java:74
io.netty.util.concurrent.FastThreadLocalRunnable.run() FastThreadLocalRunnable.java:30
java.lang.Thread.run() Thread.java:834
```
Since we are introduced the internal thread pool for handling the client internal executions.
So the fix is using the internal thread to process the callback of the hasMessageAvailableAsync
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] 315157973 commented on pull request #11183: Fix the dead lock when using hasMessageAvailableAsync and readNextAsync
Posted by GitBox <gi...@apache.org>.
315157973 commented on pull request #11183:
URL: https://github.com/apache/pulsar/pull/11183#issuecomment-872699118
Great work! This is very difficult to troubleshoot, thanks penghui
When the Future has been completed, use future.whenComplete at this time, and the current thread will be used. A thread switch causes a deadlock. Here is a Demo:
```
CompletableFuture<Void> future = new CompletableFuture<>();
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,
new BasicThreadFactory.Builder().namingPattern("metrics-schedule-pool-%d").build());
executorService.execute(() -> {
System.out.println("executorService:" + Thread.currentThread().getName());
future.complete(null);
});
Thread.sleep(5000);
future.whenComplete(((unused, throwable) -> {
System.out.println("whenComplete:" + Thread.currentThread().getName());
}));
System.out.println("out:" + Thread.currentThread().getName());
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] gaoran10 commented on a change in pull request #11183: Fix the dead lock when using hasMessageAvailableAsync and readNextAsync
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on a change in pull request #11183:
URL: https://github.com/apache/pulsar/pull/11183#discussion_r662364584
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
##########
@@ -510,4 +514,44 @@ public void testSameSubName() throws Exception {
}
+ @Test(timeOut = 30000)
+ public void testAvoidUsingIoThreadToGetValueOfMessage() throws Exception {
+ final String topic = "persistent://my-property/my-ns/testAvoidUsingIoThreadToGetValueOfMessage";
+
+ @Cleanup
+ Producer<Schemas.PersonOne> producer = pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic)
+ .create();
+
+ producer.send(new Schemas.PersonOne(1));
+
+ @Cleanup
+ Reader<Schemas.PersonOne> reader = pulsarClient.newReader(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic)
+ .startMessageId(MessageId.earliest)
+ .create();
+
+ CountDownLatch latch = new CountDownLatch(1);
+ List<Schemas.PersonOne> received = new ArrayList<>(1);
+ // Make sure the message is added to the incoming queue
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(((ReaderImpl<?>) reader).getConsumer().incomingMessages.size() > 0));
+ reader.hasMessageAvailableAsync().whenComplete((has, e) -> {
+ if (e == null && has) {
+ CompletableFuture<Message<Schemas.PersonOne>> future = reader.readNextAsync();
+ // Make sure the future completed
+ Awaitility.await().pollInterval(1, TimeUnit.MILLISECONDS).untilAsserted(future::isDone);
Review comment:
Why need to wait for the future?
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #11183: Fix the dead lock when using hasMessageAvailableAsync and readNextAsync
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #11183:
URL: https://github.com/apache/pulsar/pull/11183
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on a change in pull request #11183: Fix the dead lock when using hasMessageAvailableAsync and readNextAsync
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on a change in pull request #11183:
URL: https://github.com/apache/pulsar/pull/11183#discussion_r662369792
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
##########
@@ -510,4 +514,44 @@ public void testSameSubName() throws Exception {
}
+ @Test(timeOut = 30000)
+ public void testAvoidUsingIoThreadToGetValueOfMessage() throws Exception {
+ final String topic = "persistent://my-property/my-ns/testAvoidUsingIoThreadToGetValueOfMessage";
+
+ @Cleanup
+ Producer<Schemas.PersonOne> producer = pulsarClient.newProducer(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic)
+ .create();
+
+ producer.send(new Schemas.PersonOne(1));
+
+ @Cleanup
+ Reader<Schemas.PersonOne> reader = pulsarClient.newReader(Schema.AVRO(Schemas.PersonOne.class))
+ .topic(topic)
+ .startMessageId(MessageId.earliest)
+ .create();
+
+ CountDownLatch latch = new CountDownLatch(1);
+ List<Schemas.PersonOne> received = new ArrayList<>(1);
+ // Make sure the message is added to the incoming queue
+ Awaitility.await().untilAsserted(() ->
+ assertTrue(((ReaderImpl<?>) reader).getConsumer().incomingMessages.size() > 0));
+ reader.hasMessageAvailableAsync().whenComplete((has, e) -> {
+ if (e == null && has) {
+ CompletableFuture<Message<Schemas.PersonOne>> future = reader.readNextAsync();
+ // Make sure the future completed
+ Awaitility.await().pollInterval(1, TimeUnit.MILLISECONDS).untilAsserted(future::isDone);
Review comment:
This is to simulate the race condition here, the problem happens only the future is already completed, so that will continue using the IO thread to execute the callback.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] hangc0276 commented on pull request #11183: Fix the dead lock when using hasMessageAvailableAsync and readNextAsync
Posted by GitBox <gi...@apache.org>.
hangc0276 commented on pull request #11183:
URL: https://github.com/apache/pulsar/pull/11183#issuecomment-873009235
Great catch! The deadlock it hard to debug.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org