You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2026/09/28 21:06:33 UTC
(pulsar) branch master updated: [fix][client] Release connections opened after close (#26733)
This is an automated email from the ASF dual-hosted git repository.
merlimat pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new c958fdd6c469 [fix][client] Release connections opened after close (#26733)
c958fdd6c469 is described below
commit c958fdd6c4697143fe120f7cc26bf950f9c65610
Author: Stochastic <13...@users.noreply.github.com>
AuthorDate: Tue Sep 29 05:06:26 2026 +0800
[fix][client] Release connections opened after close (#26733)
---
.../java/org/apache/pulsar/client/impl/ConsumerImpl.java | 1 +
.../java/org/apache/pulsar/client/impl/ProducerImpl.java | 1 +
.../org/apache/pulsar/client/impl/ConsumerImplTest.java | 12 ++++++++++++
.../org/apache/pulsar/client/impl/ProducerImplTest.java | 16 ++++++++++++++++
4 files changed, 30 insertions(+)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 2a81cf336944..fc3ce28b4121 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -886,6 +886,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
deregisterFromClientCnx();
client.cleanupConsumer(this);
clearReceiverQueue(false);
+ client.getCnxPool().releaseConnection(cnx);
return CompletableFuture.completedFuture(null);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index 7701fda5e9ec..d2d634bbccde 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -2142,6 +2142,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
failPendingMessages(cnx,
new PulsarClientException.ProducerFencedException("producer has been closed"));
}
+ client.getCnxPool().releaseConnection(cnx);
return CompletableFuture.completedFuture(null);
}
// We set the cnx reference before registering the producer on the cnx, so if the cnx breaks before creating
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java
index 9b122da2cb45..18c2198dfa89 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ConsumerImplTest.java
@@ -239,6 +239,18 @@ public class ConsumerImplTest {
Assert.assertNull(checkException);
}
+ @Test
+ public void testConnectionOpenedAfterConsumerCloseReleasesConnection() {
+ ConnectionPool connectionPool = mock(ConnectionPool.class);
+ when(consumer.client.getCnxPool()).thenReturn(connectionPool);
+ ClientCnx clientCnx = mock(ClientCnx.class);
+
+ consumer.setState(HandlerState.State.Closed);
+ consumer.connectionOpened(clientCnx).join();
+
+ verify(connectionPool).releaseConnection(clientCnx);
+ }
+
@Test
public void testConsumerCreatedWhilePaused() throws InterruptedException {
PulsarClientImpl client = ClientTestFixtures.createPulsarClientMock(executorProvider, internalExecutor);
diff --git a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
index df465be819b5..9fce50f781ad 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/ProducerImplTest.java
@@ -60,6 +60,7 @@ import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.MessageCrypto;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
@@ -285,6 +286,21 @@ public class ProducerImplTest {
}
}
+ @Test
+ public void testConnectionOpenedAfterProducerCloseReleasesConnection() {
+ PulsarClientImpl client = mockedPulsarClient();
+ CompletableFuture<Pair<ClientCnx, Boolean>> connectionFuture = new CompletableFuture<>();
+ when(client.getConnection(Mockito.anyString(), Mockito.anyInt())).thenReturn(connectionFuture);
+ ProducerImpl<byte[]> producer = constructProducer(client, new ProducerConfigurationData());
+ ClientCnx clientCnx = mock(ClientCnx.class, Mockito.RETURNS_DEEP_STUBS);
+
+ producer.setState(ProducerImpl.State.Closed);
+ producer.connectionOpened(clientCnx).join();
+
+ verify(client.getCnxPool()).releaseConnection(clientCnx);
+ connectionFuture.completeExceptionally(new PulsarClientException("producer closed during connect"));
+ }
+
private ProducerConfigurationData encryptedProducerConf() {
ProducerConfigurationData conf = new ProducerConfigurationData();
conf.setEncryptionKeys(new TreeSet<>(Collections.singleton("key")));