You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2024/10/22 14:15:38 UTC
[PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
poorbarcode opened a new pull request, #23499:
URL: https://github.com/apache/pulsar/pull/23499
### Motivation
**Issue 1**: see the test `testUnstableNetWorkWhenCreatingProducer`
The producer fails to start if the network is unstable.
| time | socket closed | start producer |
| --- | --- | --- |
| 1 | | get a pooled connection from the pool |
| 2 | | before registering the producer to the connection |
| 3 | socket closed |
| 4 | reconnect all producers, which is empty now |
| 5 | register the producer to a closed connection |
| 6 | failed to send `Command.newProducer` to brokers |
| 7 | users receive an error `io.netty.channel.StacklessClosedChannelException: null` |
---
**Issue 2**: see the test: `testSendAfterCreateProducerAsync`
- Start a producer/consumer async
- Call a sync method in the callback method of method `createAsync`, such as `producer.send`, and `consumer.receive`.
- The IO thread encounters a deadlock
### Modifications
- fix the issues above
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: x
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#discussion_r1811666334
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -915,13 +915,24 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
resetBackoff();
- boolean firstTimeConnect = subscribeFuture.complete(this);
- // if the consumer is not partitioned or is re-connected and is partitioned, we send the flow
- // command to receive messages.
- if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
- increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
- }
- future.complete(null);
+ // Switch to user thread to avoid a deadlock as follows:
+ // - create consumer async.
+ // - use IO thread to call a sync method.
+ // - the thread was stuck, and will never respond for brokers.
+ client.externalExecutorProvider().getExecutor(ConsumerImpl.this).execute(() -> {
Review Comment:
The external thread pool is used by message listeners, we usually use internal thread pool for the client's internal tasks.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -915,13 +915,24 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
resetBackoff();
- boolean firstTimeConnect = subscribeFuture.complete(this);
- // if the consumer is not partitioned or is re-connected and is partitioned, we send the flow
- // command to receive messages.
- if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
- increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
- }
- future.complete(null);
+ // Switch to user thread to avoid a deadlock as follows:
+ // - create consumer async.
+ // - use IO thread to call a sync method.
+ // - the thread was stuck, and will never respond for brokers.
+ client.externalExecutorProvider().getExecutor(ConsumerImpl.this).execute(() -> {
Review Comment:
And do we really need to fix this issue? User should not call a sync method in the callback of the async method. This could be something of the scope of Pulsar.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2001,8 +2001,14 @@ private void resendMessages(ClientCnx cnx, long expectedEpoch) {
log.debug("[{}] [{}] No pending messages to resend {}", topic, producerName, messagesToResend);
}
if (changeToReadyState()) {
- producerCreatedFuture.complete(ProducerImpl.this);
- scheduleBatchFlushTask(0);
+ // Switch to user thread to avoid a deadlock as follows:
+ // - create producer async.
+ // - use IO thread to call a sync method.
+ // - the thread was stuck, and will never respond for brokers.
+ client.externalExecutorProvider().getExecutor(ProducerImpl.this).submit(() -> {
+ producerCreatedFuture.complete(ProducerImpl.this);
+ scheduleBatchFlushTask(0);
+ });
Review Comment:
Same as the above comment.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TopicMetadata;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class SimpleProduceConsumeIoTest extends ProducerConsumerBase {
+
+ private PulsarClientImpl singleConnectionPerBrokerClient;
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ singleConnectionPerBrokerClient = (PulsarClientImpl) PulsarClient.builder().connectionsPerBroker(1)
+ .serviceUrl(lookupUrl.toString()).build();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (singleConnectionPerBrokerClient != null) {
+ singleConnectionPerBrokerClient.close();
+ }
+ super.internalCleanup();
+ }
+
+ /**
+ * 1. Create a producer with a pooled connection.
+ * 2. When executing "producer.connectionOpened", the pooled connection has been closed due to a network issue.
+ * 3. Verify: the producer can be created successfully.
+ */
+ @Test
+ public void testUnstableNetWorkWhenCreatingProducer() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ // Trigger a pooled connection creation.
+ ProducerImpl p = (ProducerImpl) singleConnectionPerBrokerClient.newProducer().topic(topic).create();
+ ClientCnx cnx = p.getClientCnx();
+ p.close();
+
+ // 1. Create a new producer with the pooled connection(since there is a pooled connection, the new producer
+ // will reuse it).
+ // 2. Trigger a network issue.
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ // A task for trigger network issue.
+ new Thread(() -> {
+ try {
+ countDownLatch.await();
+ cnx.ctx().close();
+ } catch (Exception ex) {
+ }
+ }).start();
+ // Create a new producer with the pooled connection.
+ AtomicReference<CompletableFuture<Producer<byte[]>>> p2FutureWrap = new AtomicReference<>();
+ new Thread(() -> {
+ ProducerBuilder producerBuilder = singleConnectionPerBrokerClient.newProducer().topic(topic);
+ ProducerConfigurationData producerConf = WhiteboxImpl.getInternalState(producerBuilder, "conf");
+ CompletableFuture<Producer<byte[]>> p2Future = new CompletableFuture();
+ p2FutureWrap.set(p2Future);
+ new ProducerImpl<>(singleConnectionPerBrokerClient, "public/default/tp1", producerConf, p2Future,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
+ // Mock a network issue, and wait for the issue occurred.
+ countDownLatch.countDown();
+ try {
+ Thread.sleep(1500);
+ } catch (InterruptedException e) {
+ }
+ // Call the real implementation.
+ return super.connectionOpened(cnx);
+ }
+ };
+ }).start();
+
+ // Verify: the producer can be created successfully.
+ Awaitility.await().untilAsserted(() -> {
+ assertNotNull(p2FutureWrap.get());
+ assertTrue(p2FutureWrap.get().isDone());
+ });
+ // Print log.
+ p2FutureWrap.get().exceptionally(ex -> {
+ log.error("Failed to create producer", ex);
+ return null;
+ });
+ Awaitility.await().untilAsserted(() -> {
+ assertFalse(p2FutureWrap.get().isCompletedExceptionally());
+ assertTrue("Ready".equals(
+ WhiteboxImpl.getInternalState(p2FutureWrap.get().join(), "state").toString()));
+ });
+
+ // Cleanup.
+ p2FutureWrap.get().join().close();
+ admin.topics().delete(topic);
+ }
+
+ @Test(timeOut = 10000)
+ public void testSendAfterCreateProducerAsync() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ AtomicReference<Producer<byte[]>> producerWrap = new AtomicReference<>();
+ CompletableFuture<Void> sendFuture = new CompletableFuture<>();
+ singleConnectionPerBrokerClient.newProducer().topic(topic).createAsync().thenAccept(p -> {
+ try {
+ producerWrap.set(p);
+ p.send(new byte[]{1, 2, 3});
+ sendFuture.complete(null);
+ } catch (PulsarClientException e) {
+ sendFuture.completeExceptionally(e);
+ }
+ });
+ sendFuture.get();
+
+ // cleanup.
+ producerWrap.get().close();
+ admin.topics().delete(topic);
+ }
+
+ @Test(timeOut = 20000)
+ public void testConsumeAfterCreateConsumerAsync() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ final String subscription = "s1";
+ admin.topics().createNonPartitionedTopic(topic);
+ admin.topics().createSubscription(topic, subscription, MessageId.earliest);
+ Producer<byte[]> producer = singleConnectionPerBrokerClient.newProducer().topic(topic).create();
+ producer.send("1".getBytes(StandardCharsets.UTF_8));
+
+ AtomicReference<Consumer<byte[]>> consumerWrap = new AtomicReference<>();
+ CompletableFuture<Void> consumeTask = new CompletableFuture<>();
+ singleConnectionPerBrokerClient.newConsumer().subscriptionName(subscription).topic(topic)
+ .isAckReceiptEnabled(true).subscribeAsync().thenAccept(c -> {
+ try {
+ consumerWrap.set(c);
+ Message<byte[]> msg = c.receive(10, TimeUnit.SECONDS);
Review Comment:
Same as the above comment, this is a misuse of the Async API.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TopicMetadata;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class SimpleProduceConsumeIoTest extends ProducerConsumerBase {
+
+ private PulsarClientImpl singleConnectionPerBrokerClient;
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ singleConnectionPerBrokerClient = (PulsarClientImpl) PulsarClient.builder().connectionsPerBroker(1)
+ .serviceUrl(lookupUrl.toString()).build();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (singleConnectionPerBrokerClient != null) {
+ singleConnectionPerBrokerClient.close();
+ }
+ super.internalCleanup();
+ }
+
+ /**
+ * 1. Create a producer with a pooled connection.
+ * 2. When executing "producer.connectionOpened", the pooled connection has been closed due to a network issue.
+ * 3. Verify: the producer can be created successfully.
+ */
+ @Test
+ public void testUnstableNetWorkWhenCreatingProducer() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ // Trigger a pooled connection creation.
+ ProducerImpl p = (ProducerImpl) singleConnectionPerBrokerClient.newProducer().topic(topic).create();
+ ClientCnx cnx = p.getClientCnx();
+ p.close();
+
+ // 1. Create a new producer with the pooled connection(since there is a pooled connection, the new producer
+ // will reuse it).
+ // 2. Trigger a network issue.
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ // A task for trigger network issue.
+ new Thread(() -> {
+ try {
+ countDownLatch.await();
+ cnx.ctx().close();
+ } catch (Exception ex) {
+ }
+ }).start();
+ // Create a new producer with the pooled connection.
+ AtomicReference<CompletableFuture<Producer<byte[]>>> p2FutureWrap = new AtomicReference<>();
+ new Thread(() -> {
+ ProducerBuilder producerBuilder = singleConnectionPerBrokerClient.newProducer().topic(topic);
+ ProducerConfigurationData producerConf = WhiteboxImpl.getInternalState(producerBuilder, "conf");
+ CompletableFuture<Producer<byte[]>> p2Future = new CompletableFuture();
+ p2FutureWrap.set(p2Future);
+ new ProducerImpl<>(singleConnectionPerBrokerClient, "public/default/tp1", producerConf, p2Future,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
+ // Mock a network issue, and wait for the issue occurred.
+ countDownLatch.countDown();
+ try {
+ Thread.sleep(1500);
+ } catch (InterruptedException e) {
+ }
+ // Call the real implementation.
+ return super.connectionOpened(cnx);
+ }
+ };
+ }).start();
+
+ // Verify: the producer can be created successfully.
+ Awaitility.await().untilAsserted(() -> {
+ assertNotNull(p2FutureWrap.get());
+ assertTrue(p2FutureWrap.get().isDone());
+ });
+ // Print log.
+ p2FutureWrap.get().exceptionally(ex -> {
+ log.error("Failed to create producer", ex);
+ return null;
+ });
+ Awaitility.await().untilAsserted(() -> {
+ assertFalse(p2FutureWrap.get().isCompletedExceptionally());
+ assertTrue("Ready".equals(
+ WhiteboxImpl.getInternalState(p2FutureWrap.get().join(), "state").toString()));
+ });
+
+ // Cleanup.
+ p2FutureWrap.get().join().close();
+ admin.topics().delete(topic);
+ }
+
+ @Test(timeOut = 10000)
+ public void testSendAfterCreateProducerAsync() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ AtomicReference<Producer<byte[]>> producerWrap = new AtomicReference<>();
+ CompletableFuture<Void> sendFuture = new CompletableFuture<>();
+ singleConnectionPerBrokerClient.newProducer().topic(topic).createAsync().thenAccept(p -> {
+ try {
+ producerWrap.set(p);
+ p.send(new byte[]{1, 2, 3});
Review Comment:
It should be a misuse of the async API. The deadlock is expected if users try to call a sync method in a callback of the async method. Do we really need to fix this issue?
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TopicMetadata;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class SimpleProduceConsumeIoTest extends ProducerConsumerBase {
+
+ private PulsarClientImpl singleConnectionPerBrokerClient;
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ singleConnectionPerBrokerClient = (PulsarClientImpl) PulsarClient.builder().connectionsPerBroker(1)
+ .serviceUrl(lookupUrl.toString()).build();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (singleConnectionPerBrokerClient != null) {
+ singleConnectionPerBrokerClient.close();
+ }
+ super.internalCleanup();
+ }
+
+ /**
+ * 1. Create a producer with a pooled connection.
+ * 2. When executing "producer.connectionOpened", the pooled connection has been closed due to a network issue.
+ * 3. Verify: the producer can be created successfully.
+ */
+ @Test
+ public void testUnstableNetWorkWhenCreatingProducer() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ // Trigger a pooled connection creation.
+ ProducerImpl p = (ProducerImpl) singleConnectionPerBrokerClient.newProducer().topic(topic).create();
+ ClientCnx cnx = p.getClientCnx();
+ p.close();
+
+ // 1. Create a new producer with the pooled connection(since there is a pooled connection, the new producer
+ // will reuse it).
+ // 2. Trigger a network issue.
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ // A task for trigger network issue.
+ new Thread(() -> {
+ try {
+ countDownLatch.await();
+ cnx.ctx().close();
+ } catch (Exception ex) {
+ }
+ }).start();
+ // Create a new producer with the pooled connection.
+ AtomicReference<CompletableFuture<Producer<byte[]>>> p2FutureWrap = new AtomicReference<>();
+ new Thread(() -> {
+ ProducerBuilder producerBuilder = singleConnectionPerBrokerClient.newProducer().topic(topic);
+ ProducerConfigurationData producerConf = WhiteboxImpl.getInternalState(producerBuilder, "conf");
+ CompletableFuture<Producer<byte[]>> p2Future = new CompletableFuture();
+ p2FutureWrap.set(p2Future);
+ new ProducerImpl<>(singleConnectionPerBrokerClient, "public/default/tp1", producerConf, p2Future,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
+ // Mock a network issue, and wait for the issue occurred.
+ countDownLatch.countDown();
+ try {
+ Thread.sleep(1500);
+ } catch (InterruptedException e) {
+ }
+ // Call the real implementation.
+ return super.connectionOpened(cnx);
+ }
+ };
+ }).start();
+
+ // Verify: the producer can be created successfully.
+ Awaitility.await().untilAsserted(() -> {
+ assertNotNull(p2FutureWrap.get());
+ assertTrue(p2FutureWrap.get().isDone());
+ });
+ // Print log.
+ p2FutureWrap.get().exceptionally(ex -> {
+ log.error("Failed to create producer", ex);
+ return null;
+ });
+ Awaitility.await().untilAsserted(() -> {
+ assertFalse(p2FutureWrap.get().isCompletedExceptionally());
+ assertTrue("Ready".equals(
+ WhiteboxImpl.getInternalState(p2FutureWrap.get().join(), "state").toString()));
+ });
+
+ // Cleanup.
+ p2FutureWrap.get().join().close();
+ admin.topics().delete(topic);
+ }
+
+ @Test(timeOut = 10000)
+ public void testSendAfterCreateProducerAsync() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ AtomicReference<Producer<byte[]>> producerWrap = new AtomicReference<>();
+ CompletableFuture<Void> sendFuture = new CompletableFuture<>();
+ singleConnectionPerBrokerClient.newProducer().topic(topic).createAsync().thenAccept(p -> {
+ try {
+ producerWrap.set(p);
+ p.send(new byte[]{1, 2, 3});
Review Comment:
It should be a misuse of the async API. The deadlock is expected if users try to call a sync method in a callback of the async method. Do we really need to fix this issue?
--
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
Re: [PR] [fix][client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "Technoboy- (via GitHub)" <gi...@apache.org>.
Technoboy- merged PR #23499:
URL: https://github.com/apache/pulsar/pull/23499
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#discussion_r1811674997
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TopicMetadata;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class SimpleProduceConsumeIoTest extends ProducerConsumerBase {
+
+ private PulsarClientImpl singleConnectionPerBrokerClient;
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ singleConnectionPerBrokerClient = (PulsarClientImpl) PulsarClient.builder().connectionsPerBroker(1)
+ .serviceUrl(lookupUrl.toString()).build();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (singleConnectionPerBrokerClient != null) {
+ singleConnectionPerBrokerClient.close();
+ }
+ super.internalCleanup();
+ }
+
+ /**
+ * 1. Create a producer with a pooled connection.
+ * 2. When executing "producer.connectionOpened", the pooled connection has been closed due to a network issue.
+ * 3. Verify: the producer can be created successfully.
+ */
+ @Test
+ public void testUnstableNetWorkWhenCreatingProducer() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ // Trigger a pooled connection creation.
+ ProducerImpl p = (ProducerImpl) singleConnectionPerBrokerClient.newProducer().topic(topic).create();
+ ClientCnx cnx = p.getClientCnx();
+ p.close();
+
+ // 1. Create a new producer with the pooled connection(since there is a pooled connection, the new producer
+ // will reuse it).
+ // 2. Trigger a network issue.
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ // A task for trigger network issue.
+ new Thread(() -> {
+ try {
+ countDownLatch.await();
+ cnx.ctx().close();
+ } catch (Exception ex) {
+ }
+ }).start();
+ // Create a new producer with the pooled connection.
+ AtomicReference<CompletableFuture<Producer<byte[]>>> p2FutureWrap = new AtomicReference<>();
+ new Thread(() -> {
+ ProducerBuilder producerBuilder = singleConnectionPerBrokerClient.newProducer().topic(topic);
+ ProducerConfigurationData producerConf = WhiteboxImpl.getInternalState(producerBuilder, "conf");
+ CompletableFuture<Producer<byte[]>> p2Future = new CompletableFuture();
+ p2FutureWrap.set(p2Future);
+ new ProducerImpl<>(singleConnectionPerBrokerClient, "public/default/tp1", producerConf, p2Future,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
+ // Mock a network issue, and wait for the issue occurred.
+ countDownLatch.countDown();
+ try {
+ Thread.sleep(1500);
+ } catch (InterruptedException e) {
+ }
+ // Call the real implementation.
+ return super.connectionOpened(cnx);
+ }
+ };
+ }).start();
+
+ // Verify: the producer can be created successfully.
+ Awaitility.await().untilAsserted(() -> {
+ assertNotNull(p2FutureWrap.get());
+ assertTrue(p2FutureWrap.get().isDone());
+ });
+ // Print log.
+ p2FutureWrap.get().exceptionally(ex -> {
+ log.error("Failed to create producer", ex);
+ return null;
+ });
+ Awaitility.await().untilAsserted(() -> {
+ assertFalse(p2FutureWrap.get().isCompletedExceptionally());
+ assertTrue("Ready".equals(
+ WhiteboxImpl.getInternalState(p2FutureWrap.get().join(), "state").toString()));
+ });
+
+ // Cleanup.
+ p2FutureWrap.get().join().close();
+ admin.topics().delete(topic);
+ }
+
+ @Test(timeOut = 10000)
+ public void testSendAfterCreateProducerAsync() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ AtomicReference<Producer<byte[]>> producerWrap = new AtomicReference<>();
+ CompletableFuture<Void> sendFuture = new CompletableFuture<>();
+ singleConnectionPerBrokerClient.newProducer().topic(topic).createAsync().thenAccept(p -> {
+ try {
+ producerWrap.set(p);
+ p.send(new byte[]{1, 2, 3});
Review Comment:
Removed the improvement
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/SimpleProduceConsumeIoTest.java:
##########
@@ -0,0 +1,266 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.client.impl;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRouter;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TopicMetadata;
+import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
+import org.awaitility.Awaitility;
+import org.awaitility.reflect.WhiteboxImpl;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+public class SimpleProduceConsumeIoTest extends ProducerConsumerBase {
+
+ private PulsarClientImpl singleConnectionPerBrokerClient;
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ singleConnectionPerBrokerClient = (PulsarClientImpl) PulsarClient.builder().connectionsPerBroker(1)
+ .serviceUrl(lookupUrl.toString()).build();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ if (singleConnectionPerBrokerClient != null) {
+ singleConnectionPerBrokerClient.close();
+ }
+ super.internalCleanup();
+ }
+
+ /**
+ * 1. Create a producer with a pooled connection.
+ * 2. When executing "producer.connectionOpened", the pooled connection has been closed due to a network issue.
+ * 3. Verify: the producer can be created successfully.
+ */
+ @Test
+ public void testUnstableNetWorkWhenCreatingProducer() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ // Trigger a pooled connection creation.
+ ProducerImpl p = (ProducerImpl) singleConnectionPerBrokerClient.newProducer().topic(topic).create();
+ ClientCnx cnx = p.getClientCnx();
+ p.close();
+
+ // 1. Create a new producer with the pooled connection(since there is a pooled connection, the new producer
+ // will reuse it).
+ // 2. Trigger a network issue.
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ // A task for trigger network issue.
+ new Thread(() -> {
+ try {
+ countDownLatch.await();
+ cnx.ctx().close();
+ } catch (Exception ex) {
+ }
+ }).start();
+ // Create a new producer with the pooled connection.
+ AtomicReference<CompletableFuture<Producer<byte[]>>> p2FutureWrap = new AtomicReference<>();
+ new Thread(() -> {
+ ProducerBuilder producerBuilder = singleConnectionPerBrokerClient.newProducer().topic(topic);
+ ProducerConfigurationData producerConf = WhiteboxImpl.getInternalState(producerBuilder, "conf");
+ CompletableFuture<Producer<byte[]>> p2Future = new CompletableFuture();
+ p2FutureWrap.set(p2Future);
+ new ProducerImpl<>(singleConnectionPerBrokerClient, "public/default/tp1", producerConf, p2Future,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
+ // Mock a network issue, and wait for the issue occurred.
+ countDownLatch.countDown();
+ try {
+ Thread.sleep(1500);
+ } catch (InterruptedException e) {
+ }
+ // Call the real implementation.
+ return super.connectionOpened(cnx);
+ }
+ };
+ }).start();
+
+ // Verify: the producer can be created successfully.
+ Awaitility.await().untilAsserted(() -> {
+ assertNotNull(p2FutureWrap.get());
+ assertTrue(p2FutureWrap.get().isDone());
+ });
+ // Print log.
+ p2FutureWrap.get().exceptionally(ex -> {
+ log.error("Failed to create producer", ex);
+ return null;
+ });
+ Awaitility.await().untilAsserted(() -> {
+ assertFalse(p2FutureWrap.get().isCompletedExceptionally());
+ assertTrue("Ready".equals(
+ WhiteboxImpl.getInternalState(p2FutureWrap.get().join(), "state").toString()));
+ });
+
+ // Cleanup.
+ p2FutureWrap.get().join().close();
+ admin.topics().delete(topic);
+ }
+
+ @Test(timeOut = 10000)
+ public void testSendAfterCreateProducerAsync() throws Exception {
+ final String topic = BrokerTestUtil.newUniqueName("persistent://public/default/tp");
+ admin.topics().createNonPartitionedTopic(topic);
+ AtomicReference<Producer<byte[]>> producerWrap = new AtomicReference<>();
+ CompletableFuture<Void> sendFuture = new CompletableFuture<>();
+ singleConnectionPerBrokerClient.newProducer().topic(topic).createAsync().thenAccept(p -> {
+ try {
+ producerWrap.set(p);
+ p.send(new byte[]{1, 2, 3});
Review Comment:
Removed the improvement
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -915,13 +915,24 @@ public CompletableFuture<Void> connectionOpened(final ClientCnx cnx) {
resetBackoff();
- boolean firstTimeConnect = subscribeFuture.complete(this);
- // if the consumer is not partitioned or is re-connected and is partitioned, we send the flow
- // command to receive messages.
- if (!(firstTimeConnect && hasParentConsumer) && getCurrentReceiverQueueSize() != 0) {
- increaseAvailablePermits(cnx, getCurrentReceiverQueueSize());
- }
- future.complete(null);
+ // Switch to user thread to avoid a deadlock as follows:
+ // - create consumer async.
+ // - use IO thread to call a sync method.
+ // - the thread was stuck, and will never respond for brokers.
+ client.externalExecutorProvider().getExecutor(ConsumerImpl.this).execute(() -> {
Review Comment:
Removed the improvement
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2001,8 +2001,14 @@ private void resendMessages(ClientCnx cnx, long expectedEpoch) {
log.debug("[{}] [{}] No pending messages to resend {}", topic, producerName, messagesToResend);
}
if (changeToReadyState()) {
- producerCreatedFuture.complete(ProducerImpl.this);
- scheduleBatchFlushTask(0);
+ // Switch to user thread to avoid a deadlock as follows:
+ // - create producer async.
+ // - use IO thread to call a sync method.
+ // - the thread was stuck, and will never respond for brokers.
+ client.externalExecutorProvider().getExecutor(ProducerImpl.this).submit(() -> {
+ producerCreatedFuture.complete(ProducerImpl.this);
+ scheduleBatchFlushTask(0);
+ });
Review Comment:
Removed the improvement
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#discussion_r1811052143
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java:
##########
@@ -275,7 +276,8 @@ public CompletableFuture<ClientCnx> getConnection(InetSocketAddress logicalAddre
}
// Try use exists connection.
if (clientCnx.getIdleState().tryMarkUsingAndClearIdleTime()) {
- return CompletableFuture.completedFuture(clientCnx);
+ return CompletableFuture.completedFuture(clientCnx).supplyAsync(
Review Comment:
there's no need for the `completedFuture(clientCnx)` part, that should be removed completely.
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#discussion_r1811091145
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java:
##########
@@ -275,7 +276,8 @@ public CompletableFuture<ClientCnx> getConnection(InetSocketAddress logicalAddre
}
// Try use exists connection.
if (clientCnx.getIdleState().tryMarkUsingAndClearIdleTime()) {
- return CompletableFuture.completedFuture(clientCnx);
+ return CompletableFuture.completedFuture(clientCnx).supplyAsync(
Review Comment:
Fixed, Thanks
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#issuecomment-2430204031
> You call the sync method in the async callback thread, this is an incorrect way. I don't suggest that you switch the thread to complete the callback.
@nodece please provide the exact code location that you are referring to. Thanks!
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#discussion_r1811008893
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java:
##########
@@ -275,7 +276,8 @@ public CompletableFuture<ClientCnx> getConnection(InetSocketAddress logicalAddre
}
// Try use exists connection.
if (clientCnx.getIdleState().tryMarkUsingAndClearIdleTime()) {
- return CompletableFuture.completedFuture(clientCnx);
+ return CompletableFuture.completedFuture(clientCnx).thenApplyAsync(Function.identity(),
+ clientCnx.ctx().executor());
Review Comment:
Improved
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2001,8 +2001,14 @@ private void resendMessages(ClientCnx cnx, long expectedEpoch) {
log.debug("[{}] [{}] No pending messages to resend {}", topic, producerName, messagesToResend);
}
if (changeToReadyState()) {
- producerCreatedFuture.complete(ProducerImpl.this);
- scheduleBatchFlushTask(0);
+ // Switch to user thread to avoid a deadlock as follows:
+ // - create producer async.
+ // - use IO thread to call a sync method.
+ // - the thread was stuck, and will never respond for brokers.
+ client.externalExecutorProvider().getExecutor().submit(() -> {
Review Comment:
Improved
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#issuecomment-2430211146
> It looks the CompletableFuture.supplyAsync(() -> clientCnx, clientCnx.ctx().executor()); can fix this issue, but I don’t understand this fix way.
Indeed, it would be more consistent to use `thenComposeAsync` to switch the executor for all code branches instead of just a single branch.
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#discussion_r1811052966
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java:
##########
@@ -275,7 +276,8 @@ public CompletableFuture<ClientCnx> getConnection(InetSocketAddress logicalAddre
}
// Try use exists connection.
if (clientCnx.getIdleState().tryMarkUsingAndClearIdleTime()) {
- return CompletableFuture.completedFuture(clientCnx);
+ return CompletableFuture.completedFuture(clientCnx).thenApplyAsync(Function.identity(),
+ clientCnx.ctx().executor());
Review Comment:
follow up comment in https://github.com/apache/pulsar/pull/23499#discussion_r1811052143
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#issuecomment-2430846017
> > You call the sync method in the async callback thread, this is an incorrect way. I don't suggest that you switch the thread to complete the callback.
>
> @nodece please provide the exact code location that you are referring to. Thanks!
```java
client.newProducer().topic("test").createAsync().thenApply(producer -> {
producer.send("hello-1".getBytes(StandardCharsets.UTF_8)); // incorrect, which will block the pulsar client thread.
producer.sendAsync("hello-1".getBytes(StandardCharsets.UTF_8)); // correct
});
```
> User should not call a sync method in the callback of the async method.
+1
--
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
Re: [PR] [fix] [client] Fix producer/consumer stop to reconnect or Pub/Sub due to IO thread race-condition [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23499:
URL: https://github.com/apache/pulsar/pull/23499#discussion_r1810864940
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2001,8 +2001,14 @@ private void resendMessages(ClientCnx cnx, long expectedEpoch) {
log.debug("[{}] [{}] No pending messages to resend {}", topic, producerName, messagesToResend);
}
if (changeToReadyState()) {
- producerCreatedFuture.complete(ProducerImpl.this);
- scheduleBatchFlushTask(0);
+ // Switch to user thread to avoid a deadlock as follows:
+ // - create producer async.
+ // - use IO thread to call a sync method.
+ // - the thread was stuck, and will never respond for brokers.
+ client.externalExecutorProvider().getExecutor().submit(() -> {
Review Comment:
is there a need to pin a specific thread in the executor with `getExecutor(ProducerImpl.this)`?
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionPool.java:
##########
@@ -275,7 +276,8 @@ public CompletableFuture<ClientCnx> getConnection(InetSocketAddress logicalAddre
}
// Try use exists connection.
if (clientCnx.getIdleState().tryMarkUsingAndClearIdleTime()) {
- return CompletableFuture.completedFuture(clientCnx);
+ return CompletableFuture.completedFuture(clientCnx).thenApplyAsync(Function.identity(),
+ clientCnx.ctx().executor());
Review Comment:
This could be simplified with `return CompletableFuture.supplyAsync(() -> clientCnx, clientCnx.ctx().executor);`
--
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