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 2026/10/01 19:50:03 UTC
(pulsar) 04/32: [fix][client] Prevent duplicate cleanup and leaks on producer send failures (#26641)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.0-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit fdec41de3257d918f7b563f010da5fb36c9caf85
Author: sinan liu <li...@gmail.com>
AuthorDate: Fri Sep 25 14:32:34 2026 +0800
[fix][client] Prevent duplicate cleanup and leaks on producer send failures (#26641)
(cherry picked from commit 7f4c53ec3a0acf2a837b424866897a6a037d56aa)
---
.../pulsar/client/impl/ProducerMemoryLeakTest.java | 341 ++++++++++++++++++++-
.../apache/pulsar/client/impl/ProducerImpl.java | 110 +++++--
2 files changed, 415 insertions(+), 36 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLeakTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLeakTest.java
index 43cee2421606..43b46f5752be 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLeakTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerMemoryLeakTest.java
@@ -21,15 +21,33 @@ package org.apache.pulsar.client.impl;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mockStatic;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.DefaultSelectStrategyFactory;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.concurrent.DefaultEventExecutorChooserFactory;
+import io.netty.util.concurrent.RejectedExecutionHandlers;
+import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+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.CompressionType;
@@ -40,6 +58,8 @@ import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.interceptor.ProducerInterceptor;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.protocol.ByteBufPair;
import org.apache.pulsar.common.util.FutureUtil;
@@ -73,9 +93,317 @@ public class ProducerMemoryLeakTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ private static String newTopicName(String namespace) {
+ return BrokerTestUtil.newUniqueName("persistent://" + namespace + "/tp_");
+ }
+
+ private String getBrokerServiceUrl() {
+ return pulsar.getBrokerServiceUrl();
+ }
+
+ @DataProvider
+ public Object[][] pendingMessageAfterSchemaFailure() {
+ return new Object[][] {{false}, {true}};
+ }
+
+ @Test(dataProvider = "pendingMessageAfterSchemaFailure")
+ public void testSchemaFailureCallbackStopsRecovery(boolean enqueueFollowingMessage) throws Exception {
+ try (PulsarClientImpl client = (PulsarClientImpl) newPulsarClient();
+ ProducerImpl<String> producer = (ProducerImpl<String>) client.newProducer(Schema.STRING)
+ .topic(newTopicName(NAMESPACE_NEVER_COMPATIBLE)).enableBatching(false).maxPendingMessages(10)
+ .create()) {
+ producer.send("initial");
+ ClientCnx cnx = producer.getClientCnx();
+ AtomicInteger writesAfterClose = new AtomicInteger();
+ AtomicBoolean closeStarted = new AtomicBoolean();
+ AtomicReference<CompletableFuture<Void>> closeFuture = new AtomicReference<>();
+ cnx.channel().eventLoop().submit(() -> cnx.channel().pipeline().addBefore(cnx.ctx().name(),
+ "count-sends-after-close", new ChannelOutboundHandlerAdapter() {
+ @Override
+ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
+ throws Exception {
+ if (closeStarted.get() && msg instanceof ByteBufPair) {
+ writesAfterClose.incrementAndGet();
+ }
+ super.write(ctx, msg, promise);
+ }
+ })).get(10, TimeUnit.SECONDS);
+ cnx.channel().config().setAutoRead(false);
+ MsgPayloadTouchableMessageBuilder<Boolean> incompatible =
+ new MsgPayloadTouchableMessageBuilder<>(producer, Schema.BOOL);
+ MsgPayloadTouchableMessageBuilder<String> following = newMessage(producer);
+ try {
+ CompletableFuture<MessageId> failedSend = incompatible.value(true).sendAsync();
+ CompletableFuture<MessageId> completion = failedSend.whenComplete((id, error) -> {
+ closeStarted.set(true);
+ closeFuture.set(producer.closeAsync());
+ });
+ CompletableFuture<MessageId> followingSend = enqueueFollowingMessage
+ ? following.value("must not be sent after close").sendAsync() : null;
+ assertEquals(producer.getPendingQueueSize(), enqueueFollowingMessage ? 2 : 1);
+ cnx.channel().config().setAutoRead(true);
+ try {
+ completion.get(10, TimeUnit.SECONDS);
+ fail("Expected the broker to reject the incompatible schema");
+ } catch (ExecutionException error) {
+ assertTrue(error.getCause() instanceof PulsarClientException.IncompatibleSchemaException);
+ }
+ closeFuture.get().get(10, TimeUnit.SECONDS);
+ cnx.channel().eventLoop().submit(() -> { }).get(10, TimeUnit.SECONDS);
+ assertEquals(writesAfterClose.get(), 0, "Recovery must stop after the callback closes the producer");
+ if (followingSend != null) {
+ assertTrue(followingSend.isCompletedExceptionally());
+ }
+ Awaitility.await().untilAsserted(() -> {
+ assertFalse(cnx.channel().isActive());
+ assertEquals(producer.getPendingQueueSize(), 0);
+ assertEquals(producer.availableSendPermitsForTesting(), 10);
+ assertEquals(client.getMemoryLimitController().currentUsage(), 0L);
+ assertEquals(incompatible.payload.refCnt(), 1);
+ if (enqueueFollowingMessage) {
+ assertEquals(following.payload.refCnt(), 1);
+ }
+ });
+ } finally {
+ cnx.channel().config().setAutoRead(true);
+ incompatible.release();
+ if (following.payload != null) {
+ following.release();
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testDeferredOversizedMessageClosesProducerInCallback() throws Exception {
+ try (PulsarClientImpl client = (PulsarClientImpl) newPulsarClient()) {
+ String topic = newTopicName();
+ ProducerBuilderImpl<byte[]> builder = (ProducerBuilderImpl<byte[]>) client.newProducer()
+ .topic(topic).enableBatching(false).maxPendingMessages(10);
+ CompletableFuture<Producer<byte[]>> created = new CompletableFuture<>();
+ AtomicInteger callbacks = new AtomicInteger();
+ try (ProducerImpl<byte[]> producer = new ProducerImpl<>(client, topic, builder.getConf(), created,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ protected ByteBufPair sendMessage(long producerId, long sequenceId, int numMessages,
+ MessageId messageId, MessageMetadata metadata, ByteBuf payload) {
+ ByteBufPair cmd = super.sendMessage(producerId, sequenceId, numMessages,
+ messageId, metadata, payload);
+ // Put the real deferred-schema recovery in the reconnect window before its size check.
+ ClientCnx cnx = getClientCnx();
+ connectionClosed(cnx, Optional.of(60_000L), Optional.empty());
+ cnx.channel().close();
+ return cmd;
+ }
+
+ @Override
+ protected void onSendAcknowledgement(Message<?> message, MessageId messageId, Throwable error) {
+ callbacks.incrementAndGet();
+ super.onSendAcknowledgement(message, messageId, error);
+ }
+ }) {
+ created.get(10, TimeUnit.SECONDS);
+ producer.getConnectionHandler().setMaxMessageSize(1024);
+ ClientCnx cnx = producer.getClientCnx();
+ cnx.channel().config().setAutoRead(false);
+ MsgPayloadTouchableMessageBuilder<String> message =
+ new MsgPayloadTouchableMessageBuilder<>(producer, Schema.STRING);
+ try {
+ CompletableFuture<MessageId> send = message.value("x".repeat(800))
+ .property("large-metadata", "y".repeat(400)).sendAsync();
+ CompletableFuture<MessageId> completion = send.whenComplete((id, error) -> producer.closeAsync());
+ assertEquals(producer.getPendingQueueSize(), 1);
+ cnx.channel().config().setAutoRead(true);
+ try {
+ completion.get(10, TimeUnit.SECONDS);
+ fail("Expected the deferred command to exceed the message size limit");
+ } catch (ExecutionException error) {
+ assertTrue(error.getCause() instanceof PulsarClientException.InvalidMessageException);
+ }
+ // A barrier also waits for recovery cleanup after the callback has completed the future.
+ cnx.channel().eventLoop().submit(() -> { }).get(10, TimeUnit.SECONDS);
+ assertEquals(callbacks.get(), 1);
+ assertEquals(producer.getPendingQueueSize(), 0);
+ assertEquals(producer.availableSendPermitsForTesting(), 10);
+ assertEquals(client.getMemoryLimitController().currentUsage(), 0L);
+ assertEquals(message.payload.refCnt(), 1);
+ } finally {
+ cnx.channel().config().setAutoRead(true);
+ if (message.payload.refCnt() > 0) {
+ message.release();
+ }
+ }
+ }
+ }
+ }
+
+ @DataProvider
+ public Object[][] chunkWriteFailures() {
+ return new Object[][] {{false, 0}, {false, 1}, {false, 2}, {true, 0}, {true, 1}, {true, 2}};
+ }
+
+ @Test(dataProvider = "chunkWriteFailures")
+ public void testChunkedSendStopsAfterRejectedWrite(boolean blockIfQueueFull, int rejectedChunk) throws Exception {
+ AtomicReference<Thread> rejectNextTaskFrom = new AtomicReference<>();
+ // Use a real socket/event loop. Make its task queue reject one write submission from the sender;
+ // rejecting by thread keeps broker responses and cleanup tasks running normally.
+ NioEventLoopGroup eventLoops = new NioEventLoopGroup(1, (Executor) null,
+ DefaultEventExecutorChooserFactory.INSTANCE, SelectorProvider.provider(),
+ DefaultSelectStrategyFactory.INSTANCE, RejectedExecutionHandlers.reject(),
+ capacity -> new ConcurrentLinkedQueue<>() {
+ @Override
+ public boolean offer(Runnable task) {
+ if (rejectNextTaskFrom.compareAndSet(Thread.currentThread(), null)) {
+ return false;
+ }
+ return super.offer(task);
+ }
+ });
+ ClientConfigurationData configuration = new ClientConfigurationData();
+ configuration.setServiceUrl(getBrokerServiceUrl());
+ try (PulsarClientImpl client = PulsarClientImpl.builder()
+ .conf(configuration)
+ .eventLoopGroup(eventLoops)
+ .build()) {
+ String topic = newTopicName();
+ ProducerBuilderImpl<byte[]> builder = (ProducerBuilderImpl<byte[]>) client.newProducer()
+ .topic(topic).enableBatching(false).enableChunking(true).chunkMaxMessageSize(100)
+ .maxPendingMessages(10).blockIfQueueFull(blockIfQueueFull);
+ CompletableFuture<Producer<byte[]>> created = new CompletableFuture<>();
+ AtomicInteger builtChunks = new AtomicInteger();
+ AtomicInteger callbacks = new AtomicInteger();
+ AtomicReference<ProducerImpl.ChunkedMessageCtx> context = new AtomicReference<>();
+ try (ProducerImpl<byte[]> producer = new ProducerImpl<>(client, topic, builder.getConf(), created,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ protected void processOpSendMsg(OpSendMsg op) {
+ builtChunks.incrementAndGet();
+ context.set(op.chunkedMessageCtx);
+ if (op.chunkId == rejectedChunk) {
+ rejectNextTaskFrom.set(Thread.currentThread());
+ }
+ super.processOpSendMsg(op);
+ }
+
+ @Override
+ protected void onSendAcknowledgement(Message<?> message, MessageId messageId, Throwable error) {
+ callbacks.incrementAndGet();
+ super.onSendAcknowledgement(message, messageId, error);
+ }
+ }) {
+ created.get(10, TimeUnit.SECONDS);
+ ClientCnx cnx = producer.getClientCnx();
+ cnx.channel().config().setAutoRead(false);
+ MsgPayloadTouchableMessageBuilder<byte[]> message = newMessage(producer);
+ try {
+ CompletableFuture<MessageId> send = message.value(new byte[250]).sendAsync();
+ try {
+ send.get(10, TimeUnit.SECONDS);
+ fail("Expected the chunk write submission to fail");
+ } catch (ExecutionException expected) {
+ assertTrue(expected.getCause().getCause() instanceof RejectedExecutionException);
+ }
+ assertEquals(builtChunks.get(), rejectedChunk + 1);
+ assertEquals(callbacks.get(), 1);
+ assertEquals(producer.getPendingQueueSize(), rejectedChunk);
+ assertEquals(producer.availableSendPermitsForTesting(), 10 - rejectedChunk);
+ assertEquals(client.getMemoryLimitController().currentUsage(), 0L);
+ cnx.channel().config().setAutoRead(true);
+ producer.close();
+ Awaitility.await().untilAsserted(() -> {
+ assertEquals(context.get().refCnt(), 0);
+ assertEquals(message.payload.refCnt(), 1);
+ });
+ assertEquals(producer.getPendingQueueSize(), 0);
+ assertEquals(producer.availableSendPermitsForTesting(), 10);
+ assertEquals(client.getMemoryLimitController().currentUsage(), 0L);
+ assertEquals(callbacks.get(), 1);
+ } finally {
+ rejectNextTaskFrom.set(null);
+ cnx.channel().config().setAutoRead(true);
+ message.release();
+ }
+ }
+ } finally {
+ eventLoops.shutdownGracefully(0, 0, TimeUnit.SECONDS).sync();
+ }
+ }
+
+ @Test
+ public void testInterruptedChunkedSendReleasesUnbuiltChunks() throws Exception {
+ try (PulsarClientImpl client = (PulsarClientImpl) newPulsarClient()) {
+ String topic = newTopicName();
+ ProducerBuilderImpl<byte[]> builder = (ProducerBuilderImpl<byte[]>) client.newProducer()
+ .topic(topic).enableBatching(false).enableChunking(true).chunkMaxMessageSize(100)
+ .maxPendingMessages(1).blockIfQueueFull(true);
+ CompletableFuture<Producer<byte[]>> created = new CompletableFuture<>();
+ CountDownLatch firstChunkQueued = new CountDownLatch(1);
+ AtomicReference<ProducerImpl.ChunkedMessageCtx> context = new AtomicReference<>();
+ try (ProducerImpl<byte[]> producer = new ProducerImpl<>(client, topic, builder.getConf(), created,
+ -1, Schema.BYTES, null, Optional.empty()) {
+ @Override
+ protected void processOpSendMsg(OpSendMsg op) {
+ context.set(op.chunkedMessageCtx);
+ super.processOpSendMsg(op);
+ firstChunkQueued.countDown();
+ }
+ }) {
+ created.get(10, TimeUnit.SECONDS);
+ ClientCnx cnx = producer.getClientCnx();
+ cnx.channel().config().setAutoRead(false);
+ MsgPayloadTouchableMessageBuilder<byte[]> message = newMessage(producer);
+ CompletableFuture<MessageId> sendResult = new CompletableFuture<>();
+ AtomicBoolean interruptPreserved = new AtomicBoolean();
+ Thread sender = new Thread(() -> {
+ try {
+ CompletableFuture<MessageId> send = message.value(new byte[250]).sendAsync();
+ interruptPreserved.set(Thread.currentThread().isInterrupted());
+ send.whenComplete((id, error) -> {
+ if (error != null) {
+ sendResult.completeExceptionally(error);
+ } else {
+ sendResult.complete(id);
+ }
+ });
+ } catch (Throwable error) {
+ sendResult.completeExceptionally(error);
+ }
+ }, "interrupted-chunk-sender");
+ try {
+ sender.start();
+ assertTrue(firstChunkQueued.await(10, TimeUnit.SECONDS));
+ Awaitility.await().until(() -> sender.getState() == Thread.State.WAITING);
+ sender.interrupt();
+ try {
+ sendResult.get(10, TimeUnit.SECONDS);
+ fail("Expected interruption while acquiring the second chunk's permit");
+ } catch (ExecutionException expected) {
+ assertTrue(expected.getCause().getCause() instanceof InterruptedException);
+ }
+ assertTrue(interruptPreserved.get());
+ cnx.channel().config().setAutoRead(true);
+ producer.close();
+ cnx.channel().eventLoop().submit(() -> { }).get(10, TimeUnit.SECONDS);
+ assertEquals(context.get().refCnt(), 0);
+ assertEquals(producer.availableSendPermitsForTesting(), 1);
+ assertEquals(client.getMemoryLimitController().currentUsage(), 0L);
+ assertEquals(message.payload.refCnt(), 1);
+ } finally {
+ sender.interrupt();
+ sender.join(10_000);
+ assertFalse(sender.isAlive());
+ cnx.channel().config().setAutoRead(true);
+ if (message.payload != null) {
+ message.release();
+ }
+ }
+ }
+ }
+ }
+
@Test
public void testSendQueueIsFull() throws Exception {
- final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp_");
+ final String topicName = newTopicName();
admin.topics().createNonPartitionedTopic(topicName);
ProducerImpl<String> producer = (ProducerImpl<String>) pulsarClient.newProducer(Schema.STRING)
.blockIfQueueFull(false).maxPendingMessages(1)
@@ -126,7 +454,7 @@ public class ProducerMemoryLeakTest extends ProducerConsumerBase {
@Test(dataProvider = "maxMessageSizeAndCompressions")
public void testSendMessageSizeExceeded(int maxMessageSize, CompressionType compressionType) throws Exception {
- final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp_");
+ final String topicName = newTopicName();
admin.topics().createNonPartitionedTopic(topicName);
ProducerImpl<String> producer = (ProducerImpl<String>) pulsarClient.newProducer(Schema.STRING).topic(topicName)
.compressionType(compressionType)
@@ -204,7 +532,7 @@ public class ProducerMemoryLeakTest extends ProducerConsumerBase {
@Test(dataProvider = "maxMessageSizes")
public void testBatchedSendMessageSizeExceeded(int maxMessageSize) throws Exception {
- final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp_");
+ final String topicName = newTopicName();
admin.topics().createNonPartitionedTopic(topicName);
ProducerImpl<String> producer = (ProducerImpl<String>) pulsarClient.newProducer(Schema.STRING).topic(topicName)
.enableBatching(true)
@@ -248,7 +576,7 @@ public class ProducerMemoryLeakTest extends ProducerConsumerBase {
@Test
public void testSendAfterClosedProducer() throws Exception {
- final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp_");
+ final String topicName = newTopicName();
admin.topics().createNonPartitionedTopic(topicName);
ProducerImpl<String> producer =
(ProducerImpl<String>) pulsarClient.newProducer(Schema.STRING).topic(topicName).create();
@@ -277,8 +605,7 @@ public class ProducerMemoryLeakTest extends ProducerConsumerBase {
@Test
public void testBrokenSchema() throws Exception {
- final String topicName = BrokerTestUtil.newUniqueName("persistent://" + NAMESPACE_NEVER_COMPATIBLE
- + "/tp");
+ final String topicName = newTopicName(NAMESPACE_NEVER_COMPATIBLE);
admin.topics().createNonPartitionedTopic(topicName);
ProducerImpl producer =
(ProducerImpl) pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES()).topic(topicName).create();
@@ -331,7 +658,7 @@ public class ProducerMemoryLeakTest extends ProducerConsumerBase {
@Test(dataProvider = "failedInterceptAt")
public void testInterceptorError(String method) throws Exception {
- final String topicName = BrokerTestUtil.newUniqueName("persistent://public/default/tp_");
+ final String topicName = newTopicName();
admin.topics().createNonPartitionedTopic(topicName);
ProducerImpl<String> producer = (ProducerImpl<String>) pulsarClient.newProducer(Schema.STRING).topic(topicName)
.intercept(
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 61e3b5660a23..80b57dcf379c 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
@@ -685,6 +685,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// earlier chunks carry no share): release the full reservation here and no permits -
// the permits of the already-built chunks are released by their own operations.
client.getMemoryLimitController().releaseMemory(uncompressedSize);
+ chunkedMessageCtx.release(totalChunks - chunkId);
return;
}
synchronized (this) {
@@ -720,6 +721,20 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
}
throw t;
}
+ if (chunkedMessageCtx != null && chunkId < totalChunks - 1 && chunkedMessageCtx.sendFailed) {
+ // processOpSendMsg has failed and disposed of this op. The unbuilt chunks still
+ // own the base payload, the memory reservation and their context/permit claims.
+ if (TopicName.get(topic).isPersistent()) {
+ ReferenceCountUtil.safeRelease(compressedPayload);
+ }
+ client.getMemoryLimitController().releaseMemory(uncompressedSize);
+ int remainingChunks = totalChunks - chunkId - 1;
+ if (!conf.isBlockIfQueueFull()) {
+ semaphoreRelease(remainingChunks);
+ }
+ chunkedMessageCtx.release(remainingChunks);
+ return;
+ }
readStartIndex = ((chunkId + 1) * payloadChunkSize);
}
}
@@ -1648,6 +1663,11 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
static class ChunkedMessageCtx extends AbstractReferenceCounted {
protected MessageIdImpl firstChunkMessageId;
protected MessageIdImpl lastChunkMessageId;
+ // Reset in get() before publication; subsequent reads and writes hold the producer lock.
+ // The builder reads this flag only while it still owns shares for unbuilt chunks (refCnt > 0).
+ // Those shares prevent recycling/reuse for another message even if the failed op is recycled
+ // before processOpSendMsg returns. Release them only after the builder has checked this flag.
+ private boolean sendFailed;
public ChunkMessageIdImpl getChunkMessageId() {
return new ChunkMessageIdImpl(firstChunkMessageId, lastChunkMessageId);
@@ -1664,6 +1684,7 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
public static ChunkedMessageCtx get(int totalChunks) {
ChunkedMessageCtx chunkedMessageCtx = RECYCLER.get();
chunkedMessageCtx.setRefCnt(totalChunks);
+ chunkedMessageCtx.sendFailed = false;
return chunkedMessageCtx;
}
@@ -2677,22 +2698,21 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
if (op == null) {
return;
}
+ boolean retainedForWrite = false;
try {
if (op.msg != null && isBatchMessagingEnabled()) {
batchMessageAndSend(false);
}
- if (isMessageSizeExceeded(op)) {
- op.cmd.release();
+ PulsarClientException.InvalidMessageException sizeError = getMessageSizeError(op);
+ if (sizeError != null) {
+ failSendOp(op, sizeError);
return;
}
final State state = getState();
if (state == State.Terminated || state == State.Closed || state == State.ProducerFenced) {
// The producer is in a terminal state and will never reconnect. Fail the message immediately
// rather than leaving it stuck in pendingMessages until sendTimeout.
- // releaseSemaphoreForSendOp() also gives the reserved memory back, so it must not be released again.
- releaseSemaphoreForSendOp(op);
- op.sendComplete(getTerminalException(state));
- releaseOpCmdAndRecycle(op);
+ failSendOp(op, getTerminalException(state));
return;
}
pendingMessages.add(op);
@@ -2712,8 +2732,10 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// If we do have a connection, the message is sent immediately, otherwise we'll try again once a new
// connection is established
op.cmd.retain();
+ retainedForWrite = true;
op.writeEventLoop = cnx.ctx().channel().eventLoop();
cnx.ctx().channel().eventLoop().execute(WriteInEventLoopCallback.create(this, cnx, op));
+ retainedForWrite = false;
stats.updateNumMsgsSent(op.numMessagesInBatch, op.batchSizeByte);
} else {
if (log.isDebugEnabled()) {
@@ -2726,21 +2748,38 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
// shutdown, after the op was queued): take it back out, or the recycled op would dangle in the
// queue and the resend would operate on a pooled instance.
pendingMessages.remove(op);
- releaseSemaphoreForSendOp(op);
- log.warn("[{}] [{}] error while closing out batch -- {}", topic, producerName, t);
- try {
- // The callback runs application future handlers and can throw (every other sendComplete call
- // site guards against this); the releases below must still run.
- op.sendComplete(new PulsarClientException(t, op.sequenceId));
- } catch (Throwable callbackEx) {
- log.warn("[{}] [{}] Got exception while completing the callback for msg {}:", topic,
- producerName, op.sequenceId, callbackEx);
- }
- if (op.writeEventLoop != null) {
+ if (retainedForWrite) {
// The cmd was retained for a write that never got queued: that reference has no owner
// anymore, so drop it on top of the op's own release.
ReferenceCountUtil.safeRelease(op.cmd);
}
+ failSendOp(op, t instanceof PulsarClientException
+ ? (PulsarClientException) t : new PulsarClientException(t, op.sequenceId));
+ }
+ }
+
+ /**
+ * Fails an operation that was never queued or has already been detached from pendingMessages.
+ * Unlike normal acknowledgments and failPendingMessages, this must complete the callback even for
+ * a non-last chunk: its failure stops the builder, so no last-chunk operation will be created to
+ * settle the message future, interceptor payload retain and pending-message metrics.
+ *
+ * <p>Keep this separate from failPendingMessages: that method settles already-queued operations,
+ * invokes chunked-message callbacks only for the last chunk, and returns permits together after
+ * draining its snapshot of the queue. Here the failed op returns its own accounting, while the
+ * builder releases the claims for chunks that will never be created.
+ */
+ private void failSendOp(OpSendMsg op, PulsarClientException exception) {
+ if (op.chunkedMessageCtx != null) {
+ op.chunkedMessageCtx.sendFailed = true;
+ }
+ releaseSemaphoreForSendOp(op);
+ try {
+ op.sendComplete(exception);
+ } catch (Throwable callbackEx) {
+ log.warn("[{}] [{}] Got exception while completing the callback for msg {}:", topic, producerName,
+ op.sequenceId, callbackEx);
+ } finally {
releaseOpCmdAndRecycle(op);
}
}
@@ -2785,7 +2824,23 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
MessageImpl loopStartAt = latestMsgAttemptedRegisteredSchema;
OpSendMsg loopEndDueToSchemaRegisterNeeded = null;
boolean pausedSendingToPreservePublishOrderOnSchemaRegFailure = false;
- while (msgIterator.hasNext()) {
+ while (true) {
+ // Re-check after every iteration that can invoke application callbacks, including the
+ // last one: closing the producer may also drain the queue and invalidate this iterator.
+ if (cnx() != cnx || expectedEpoch != connectionHandler.getEpoch()) {
+ return;
+ }
+ State state = getState();
+ if (state == State.Closing || state == State.Closed
+ || state == State.Terminated || state == State.ProducerFenced) {
+ // Preserve the reconnect cleanup below: closeAsync may have only closed the local
+ // producer while it was Connecting, leaving its registration on the broker intact.
+ cnx.channel().close();
+ return;
+ }
+ if (!msgIterator.hasNext()) {
+ break;
+ }
OpSendMsg op = msgIterator.next();
if (loopStartAt != null) {
if (op.msg == loopStartAt) {
@@ -2866,12 +2921,11 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
if (op.cmd == null) {
checkState(op.rePopulate != null);
op.rePopulate.run();
- if (isMessageSizeExceeded(op)) {
- // isMessageSizeExceeded already released the accounting and completed the callback:
- // take the op out of the queue and release its freshly built command too, or it stays
- // pending until the send timeout fails it - releasing the accounting a second time.
+ PulsarClientException.InvalidMessageException sizeError = getMessageSizeError(op);
+ if (sizeError != null) {
+ // Detach before the callback can close the producer and fail the pending queue again.
msgIterator.remove();
- releaseOpCmdAndRecycle(op);
+ failSendOp(op, sizeError);
continue;
}
}
@@ -2935,19 +2989,17 @@ public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask, Conne
/**
* Check if final message size for non-batch and non-chunked messages is larger than max message size.
*/
- private boolean isMessageSizeExceeded(OpSendMsg op) {
+ private PulsarClientException.InvalidMessageException getMessageSizeError(OpSendMsg op) {
if (op.msg != null && !conf.isChunkingEnabled()) {
int messageSize = op.getMessageHeaderAndPayloadSize();
if (messageSize > getMaxMessageSize()) {
- releaseSemaphoreForSendOp(op);
- op.sendComplete(new PulsarClientException.InvalidMessageException(
+ return new PulsarClientException.InvalidMessageException(
format("The producer %s of the topic %s sends a message with %d bytes that exceeds %d bytes",
producerName, topic, messageSize, getMaxMessageSize()),
- op.sequenceId));
- return true;
+ op.sequenceId);
}
}
- return false;
+ return null;
}
private int getMaxMessageSize() {