You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2025/01/17 06:56:51 UTC
(pulsar) 02/02: [fix][client] Orphan producer when concurrently calling producer closing and reconnection (#23853)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 54a7efd558e2f231fe6c503e8c25883262cb90bd
Author: fengyubiao <yu...@streamnative.io>
AuthorDate: Fri Jan 17 14:48:46 2025 +0800
[fix][client] Orphan producer when concurrently calling producer closing and reconnection (#23853)
(cherry picked from commit 56adefaf3a244514a7eacbee17ac31b1a8739391)
---
.../pulsar/client/impl/ProducerCloseTest.java | 41 +++++++
.../client/impl/ProducerReconnectionTest.java | 130 +++++++++++++++++++++
.../pulsar/client/impl/ConnectionHandler.java | 7 +-
.../apache/pulsar/client/impl/ProducerImpl.java | 32 +++--
.../pulsar/client/impl/ProducerImplTest.java | 39 -------
5 files changed, 197 insertions(+), 52 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerCloseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerCloseTest.java
index 1141af88e72..623d8e7505e 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerCloseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerCloseTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import java.time.Duration;
import lombok.Cleanup;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.pulsar.broker.service.Topic;
@@ -71,6 +72,46 @@ public class ProducerCloseTest extends ProducerConsumerBase {
};
}
+ /**
+ * Param1: Producer enableBatch or not
+ * Param2: Send in async way or not
+ */
+ @DataProvider(name = "brokenPipeline")
+ public Object[][] brokenPipeline() {
+ return new Object[][]{
+ {true},
+ {false}
+ };
+ }
+
+ @Test(dataProvider = "brokenPipeline")
+ public void testProducerCloseCallback2(boolean brokenPipeline) throws Exception {
+ initClient();
+ @Cleanup
+ ProducerImpl<byte[]> producer = (ProducerImpl<byte[]>) pulsarClient.newProducer()
+ .topic("testProducerClose")
+ .sendTimeout(5, TimeUnit.SECONDS)
+ .maxPendingMessages(0)
+ .enableBatching(false)
+ .create();
+ final TypedMessageBuilder<byte[]> messageBuilder = producer.newMessage();
+ final TypedMessageBuilder<byte[]> value = messageBuilder.value("test-msg".getBytes(StandardCharsets.UTF_8));
+ producer.getClientCnx().channel().config().setAutoRead(false);
+ final CompletableFuture<MessageId> completableFuture = value.sendAsync();
+ producer.closeAsync();
+ Thread.sleep(3000);
+ if (brokenPipeline) {
+ //producer.getClientCnx().channel().config().setAutoRead(true);
+ producer.getClientCnx().channel().close();
+ } else {
+ producer.getClientCnx().channel().config().setAutoRead(true);
+ }
+ Awaitility.await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
+ System.out.println(1);
+ Assert.assertTrue(completableFuture.isDone());
+ });
+ }
+
@Test(timeOut = 10_000)
public void testProducerCloseCallback() throws Exception {
initClient();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerReconnectionTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerReconnectionTest.java
new file mode 100644
index 00000000000..e05e666d62e
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerReconnectionTest.java
@@ -0,0 +1,130 @@
+/*
+ * 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.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.spy;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotEquals;
+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.AtomicBoolean;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.broker.BrokerTestUtil;
+import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.client.api.Schema;
+import org.awaitility.Awaitility;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+@Slf4j
+@Test(groups = "broker-api")
+public class ProducerReconnectionTest extends ProducerConsumerBase {
+
+ @BeforeClass(alwaysRun = true)
+ @Override
+ protected void setup() throws Exception {
+ super.internalSetup();
+ super.producerBaseSetup();
+ }
+
+ @AfterClass(alwaysRun = true)
+ @Override
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ }
+
+ @Test
+ public void testConcurrencyReconnectAndClose() throws Exception {
+ final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp_");
+ admin.topics().createNonPartitionedTopic(topicName);
+ PulsarClientImpl client = (PulsarClientImpl) pulsarClient;
+
+ // Create producer which will run with special steps.
+ ProducerBuilderImpl<byte[]> producerBuilder = (ProducerBuilderImpl<byte[]>) client.newProducer()
+ .blockIfQueueFull(false).maxPendingMessages(1).producerName("p1")
+ .enableBatching(true).topic(topicName);
+ CompletableFuture<Producer<byte[]>> producerFuture = new CompletableFuture<>();
+ AtomicBoolean reconnectionStartTrigger = new AtomicBoolean();
+ CountDownLatch reconnectingSignal = new CountDownLatch(1);
+ CountDownLatch closedSignal = new CountDownLatch(1);
+ ProducerImpl<byte[]> producer = new ProducerImpl<>(client, topicName, producerBuilder.getConf(), producerFuture,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ ConnectionHandler initConnectionHandler() {
+ ConnectionHandler connectionHandler = super.initConnectionHandler();
+ ConnectionHandler spyConnectionHandler = spy(connectionHandler);
+ doAnswer(invocation -> {
+ boolean result = (boolean) invocation.callRealMethod();
+ if (reconnectionStartTrigger.get()) {
+ log.info("[testConcurrencyReconnectAndClose] verified state for reconnection");
+ reconnectingSignal.countDown();
+ closedSignal.await();
+ log.info("[testConcurrencyReconnectAndClose] reconnected");
+ }
+ return result;
+ }).when(spyConnectionHandler).isValidStateForReconnection();
+ return spyConnectionHandler;
+ }
+ };
+ log.info("[testConcurrencyReconnectAndClose] producer created");
+ producerFuture.get(5, TimeUnit.SECONDS);
+
+ // Reconnect.
+ log.info("[testConcurrencyReconnectAndClose] trigger a reconnection");
+ ServerCnx serverCnx = (ServerCnx) pulsar.getBrokerService().getTopic(topicName, false).join()
+ .get().getProducers().values().iterator().next().getCnx();
+ reconnectionStartTrigger.set(true);
+ serverCnx.ctx().close();
+ producer.sendAsync("1".getBytes(StandardCharsets.UTF_8));
+ Awaitility.await().untilAsserted(() -> {
+ assertNotEquals(producer.getPendingQueueSize(), 0);
+ });
+
+ // Close producer when reconnecting.
+ reconnectingSignal.await();
+ log.info("[testConcurrencyReconnectAndClose] producer close");
+ producer.closeAsync();
+ Awaitility.await().untilAsserted(() -> {
+ HandlerState.State state1 = producer.getState();
+ assertTrue(state1 == HandlerState.State.Closed || state1 == HandlerState.State.Closing);
+ });
+ // give another thread time to call "signalToChangeStateToConnecting.await()".
+ closedSignal.countDown();
+
+ // Wait for reconnection.
+ Thread.sleep(3000);
+
+ HandlerState.State state2 = producer.getState();
+ log.info("producer state: {}", state2);
+ assertTrue(state2 == HandlerState.State.Closed || state2 == HandlerState.State.Closing);
+ assertEquals(producer.getPendingQueueSize(), 0);
+
+ // Verify: ref is expected.
+ producer.close();
+ admin.topics().delete(topicName);
+ }
+}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
index 93498594919..fb5b6788d08 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConnectionHandler.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import com.google.common.annotations.VisibleForTesting;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Optional;
@@ -192,13 +193,12 @@ public class ConnectionHandler {
duringConnect.set(false);
state.client.getCnxPool().releaseConnection(cnx);
if (CLIENT_CNX_UPDATER.compareAndSet(this, cnx, null)) {
- if (!isValidStateForReconnection()) {
+ if (!state.changeToConnecting()) {
log.info("[{}] [{}] Ignoring reconnection request (state: {})",
state.topic, state.getHandlerName(), state.getState());
return;
}
long delayMs = initialConnectionDelayMs.orElse(backoff.next());
- state.setState(State.Connecting);
log.info("[{}] [{}] Closed connection {} -- Will try again in {} s, hostUrl: {}",
state.topic, state.getHandlerName(), cnx.channel(), delayMs / 1000.0, hostUrl.orElse(null));
state.client.timer().newTimeout(timeout -> {
@@ -232,7 +232,8 @@ public class ConnectionHandler {
return EPOCH_UPDATER.incrementAndGet(this);
}
- private boolean isValidStateForReconnection() {
+ @VisibleForTesting
+ public boolean isValidStateForReconnection() {
State state = this.state.getState();
switch (state) {
case Uninitialized:
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 64b706cc5fa..304b84961b7 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
@@ -303,16 +303,20 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
producersClosedCounter = ip.newCounter("pulsar.client.producer.closed", Unit.Sessions,
"The number of producer sessions closed", topic, Attributes.empty());
- this.connectionHandler = new ConnectionHandler(this,
+ this.connectionHandler = initConnectionHandler();
+ setChunkMaxMessageSize();
+ grabCnx();
+ producersOpenedCounter.increment();
+ }
+
+ ConnectionHandler initConnectionHandler() {
+ return new ConnectionHandler(this,
new BackoffBuilder()
.setInitialTime(client.getConfiguration().getInitialBackoffIntervalNanos(), TimeUnit.NANOSECONDS)
.setMax(client.getConfiguration().getMaxBackoffIntervalNanos(), TimeUnit.NANOSECONDS)
.setMandatoryStop(Math.max(100, conf.getSendTimeoutMs() - 100), TimeUnit.MILLISECONDS)
.create(),
- this);
- setChunkMaxMessageSize();
- grabCnx();
- producersOpenedCounter.increment();
+ this);
}
private void setChunkMaxMessageSize() {
@@ -1151,7 +1155,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
@Override
- public CompletableFuture<Void> closeAsync() {
+ public synchronized CompletableFuture<Void> closeAsync() {
final State currentState = getAndUpdateState(state -> {
if (state == State.Closed) {
return state;
@@ -1179,11 +1183,11 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
cnx.sendRequestWithId(cmd, requestId).handle((v, exception) -> {
cnx.removeProducer(producerId);
- closeAndClearPendingMessages();
if (exception == null || !cnx.ctx().channel().isActive()) {
// Either we've received the success response for the close producer command from the broker, or the
// connection did break in the meantime. In any case, the producer is gone.
log.info("[{}] [{}] Closed Producer", topic, producerName);
+ closeAndClearPendingMessages();
closeFuture.complete(null);
} else {
closeFuture.completeExceptionally(exception);
@@ -1795,6 +1799,12 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// Because the state could have been updated while retrieving the connection, we set it back to connecting,
// as long as the change from current state to connecting is a valid state change.
if (!changeToConnecting()) {
+ if (getState() == State.Closing || getState() == State.Closed) {
+ // Producer was closed while reconnecting, close the connection to make sure the broker
+ // drops the producer on its side
+ failPendingMessages(cnx,
+ new PulsarClientException.ProducerFencedException("producer has been closed"));
+ }
return CompletableFuture.completedFuture(null);
}
// We set the cnx reference before registering the producer on the cnx, so if the cnx breaks before creating
@@ -1855,6 +1865,8 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// Producer was closed while reconnecting, close the connection to make sure the broker
// drops the producer on its side
cnx.removeProducer(producerId);
+ failPendingMessages(cnx,
+ new PulsarClientException.ProducerFencedException("producer has been closed"));
cnx.channel().close();
future.complete(null);
return;
@@ -2025,7 +2037,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
private void resendMessages(ClientCnx cnx, long expectedEpoch) {
cnx.ctx().channel().eventLoop().execute(() -> {
- synchronized (this) {
+ synchronized (ProducerImpl.this) {
if (getState() == State.Closing || getState() == State.Closed) {
// Producer was closed while reconnecting, close the connection to make sure the broker
// drops the producer on its side
@@ -2181,7 +2193,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
* This fails and clears the pending messages with the given exception. This method should be called from within the
* ProducerImpl object mutex.
*/
- private void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
+ private synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
if (cnx == null) {
final AtomicInteger releaseCount = new AtomicInteger();
final boolean batchMessagingEnabled = isBatchMessagingEnabled();
@@ -2333,7 +2345,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
}
- protected void processOpSendMsg(OpSendMsg op) {
+ protected synchronized void processOpSendMsg(OpSendMsg op) {
if (op == null) {
return;
}
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 5f690ead6c5..ce5a94592fd 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
@@ -25,16 +25,8 @@ import static org.mockito.Mockito.withSettings;
import static org.testng.Assert.*;
import java.nio.ByteBuffer;
-import java.util.Optional;
-import java.util.concurrent.CompletableFuture;
-
-import io.netty.util.HashedWheelTimer;
-import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
-import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
-import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.client.impl.metrics.LatencyHistogram;
-import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.mockito.Mockito;
import org.testng.annotations.Test;
@@ -74,35 +66,4 @@ public class ProducerImplTest {
assertTrue(producer.populateMessageSchema(msg, null));
verify(msg).setSchemaState(MessageImpl.SchemaState.Ready);
}
-
- @Test
- public void testClearPendingMessageWhenCloseAsync() {
- PulsarClientImpl client = mock(PulsarClientImpl.class);
- Mockito.doReturn(1L).when(client).newProducerId();
- ClientConfigurationData clientConf = new ClientConfigurationData();
- clientConf.setStatsIntervalSeconds(-1);
- Mockito.doReturn(clientConf).when(client).getConfiguration();
- Mockito.doReturn(new InstrumentProvider(null)).when(client).instrumentProvider();
- ConnectionPool connectionPool = mock(ConnectionPool.class);
- Mockito.doReturn(1).when(connectionPool).genRandomKeyToSelectCon();
- Mockito.doReturn(connectionPool).when(client).getCnxPool();
- HashedWheelTimer timer = mock(HashedWheelTimer.class);
- Mockito.doReturn(null).when(timer).newTimeout(Mockito.any(), Mockito.anyLong(), Mockito.any());
- Mockito.doReturn(timer).when(client).timer();
- ProducerConfigurationData producerConf = new ProducerConfigurationData();
- producerConf.setSendTimeoutMs(-1);
- ProducerImpl<?> producer = Mockito.spy(new ProducerImpl<>(client, "topicName", producerConf, null, 0, null, null, Optional.empty()));
-
- // make sure throw exception when send request to broker
- ClientCnx clientCnx = mock(ClientCnx.class);
- CompletableFuture<ProducerResponse> tCompletableFuture = new CompletableFuture<>();
- tCompletableFuture.completeExceptionally(new PulsarClientException("error"));
- when(clientCnx.sendRequestWithId(Mockito.any(), Mockito.anyLong())).thenReturn(tCompletableFuture);
- Mockito.doReturn(clientCnx).when(producer).cnx();
-
- // run closeAsync and verify
- CompletableFuture<Void> voidCompletableFuture = producer.closeAsync();
- verify(producer).closeAndClearPendingMessages();
- }
-
}