Skip to content

[fix][client] Defer op cmd release to the write event loop on send timeout #26456

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lhotari merged 1 commit into from
Sep 10, 2026
+474 −21
Merged

[fix][client] Defer op cmd release to the write event loop on send timeout #26456

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 84 additions & 21 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandler;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoop;
import io.netty.util.AbstractReferenceCounted;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
Expand All @@ -59,6 +60,7 @@
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
Expand Down Expand Up @@ -1176,8 +1178,18 @@ public void run() {
.log("Sending message");

try {
cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
op.updateSentTimestamp();
// Stale guard: the op may have been re-sent on another connection or disposed while this callback
// was queued behind a disconnect storm. Only write if this loop is still the op's write loop and
// op.cmd still identifies our cmd; otherwise drop only the reference this callback took.
if (op.writeEventLoop == cnx.ctx().channel().eventLoop() && op.cmd == cmd) {
cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
// Re-check before mutating: the op may have been disposed concurrently on another loop.
if (op.cmd == cmd) {
op.updateSentTimestamp();
}
} else {
cmd.release();
}
} finally {
recycle();
}
Expand Down Expand Up @@ -1428,8 +1440,7 @@ protected void ackReceived(ClientCnx cnx, long sequenceId, long highestSequenceI
.log("Got exception while completing the callback for msg");
}
}
ReferenceCountUtil.safeRelease(op.cmd);
op.recycle();
releaseOpCmdAndRecycle(op);
}

protected long getHighestSequenceId(OpSendMsg op) {
Expand Down Expand Up @@ -1488,8 +1499,7 @@ protected synchronized void recoverChecksumError(ClientCnx cnx, long sequenceId)
.exception(t)
.log("Got exception while completing the callback for msg");
}
ReferenceCountUtil.safeRelease(op.cmd);
op.recycle();
releaseOpCmdAndRecycle(op);
return;
} else {
log.debug()
Expand Down Expand Up @@ -1521,8 +1531,7 @@ protected synchronized void recoverNotAllowedError(long sequenceId, String error
.exception(t)
.log("Got exception while completing the callback for msg");
}
ReferenceCountUtil.safeRelease(op.cmd);
op.recycle();
releaseOpCmdAndRecycle(op);
}
}

Expand Down Expand Up @@ -1612,10 +1621,21 @@ protected static final class OpSendMsg {
LatencyHistogram rpcLatencyHistogram;
MessageImpl<?> msg;
List<MessageImpl<?>> msgs;
ByteBufPair cmd;
// Volatile so a stale WriteInEventLoopCallback on an older connection's event loop can never observe
// cmd == this after the disposal has released it; the disposal also clears this field before releasing.
volatile ByteBufPair cmd;
SendCallback callback;
Runnable rePopulate;
ChunkedMessageCtx chunkedMessageCtx;
/**
* The connection event loop that this op's {@link #cmd} was last handed to for writing, or
* {@code null} if the {@link #cmd} has never been handed to a connection.
*
* <p>When a timeout or recovery path fails this op from another thread (e.g. the send-timeout timer
* while the producer is in the reconnect window), the cmd release and op recycle are deferred to this
* loop so they are serialized after any in-flight write.
*/
volatile EventLoop writeEventLoop;
long uncompressedSize;
long sequenceId;
long createdAt;
Expand All @@ -1635,6 +1655,7 @@ void initialize() {
cmd = null;
callback = null;
rePopulate = null;
writeEventLoop = null;
sequenceId = -1L;
createdAt = -1L;
firstSentAt = -1L;
Expand Down Expand Up @@ -2406,8 +2427,7 @@ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
}

client.getMemoryLimitController().releaseMemory(op.uncompressedSize);
ReferenceCountUtil.safeRelease(op.cmd);
op.recycle();
releaseOpCmdAndRecycle(op);
}

semaphoreRelease(releaseCount.get());
Expand All @@ -2418,12 +2438,55 @@ synchronized void failPendingMessages(ClientCnx cnx, PulsarClientException ex) {
} else {
// If we have a connection, we schedule the callback and recycle on the event loop thread to avoid any
// race condition since we also write the message on the socket from this thread
cnx.ctx().channel().eventLoop().execute(() -> {
synchronized (ProducerImpl.this) {
failPendingMessages(null, ex);
}
});
try {
cnx.ctx().channel().eventLoop().execute(() -> {
synchronized (ProducerImpl.this) {
failPendingMessages(null, ex);
}
});
} catch (RejectedExecutionException e) {
// The connection's event loop is shutting down; the deferred task would be dropped. Fail inline
// instead (per-op cmd release is still deferred to each op's write event loop).
log.warn()
.exception(e)
.log("Connection event loop is shutting down while failing pending messages; failing inline");
failPendingMessages(null, ex);
}
}
}

/**
* Releases the op's cmd and recycles the op, deferred to the event loop the cmd was last handed to for writing
* so the buffers are never released on another thread while a write may still be reading them (e.g. the
* send-timeout timer thread during the reconnect window). If the op was never handed to a connection, the
* cleanup runs inline.
*/
private void releaseOpCmdAndRecycle(OpSendMsg op) {
final EventLoop writeEventLoop = op.writeEventLoop;
if (writeEventLoop != null) {
try {
writeEventLoop.execute(() -> releaseOpCmd(op));
return;
} catch (RejectedExecutionException e) {
// Event loop shutting down: release inline (the write either never ran, or the channel now owns
// the buffer reference it held).
log.warn()
.exception(e)
.log("Write event loop is shutting down while releasing an in-flight op cmd");
}
}
releaseOpCmd(op);
}

/**
* Detaches {@code op.cmd} before releasing it so a stale write callback can never pass its guard and write a
* released (possibly re-pooled) buffer, then recycles the op.
*/
private void releaseOpCmd(OpSendMsg op) {
final ByteBufPair cmd = op.cmd;
op.cmd = null;
ReferenceCountUtil.safeRelease(cmd);
op.recycle();
}

/**
Expand Down Expand Up @@ -2556,8 +2619,7 @@ protected synchronized void processOpSendMsg(OpSendMsg op) {
// releaseSemaphoreForSendOp() also gives the reserved memory back, so it must not be released again.
releaseSemaphoreForSendOp(op);
op.sendComplete(getTerminalException(state));
ReferenceCountUtil.safeRelease(op.cmd);
op.recycle();
releaseOpCmdAndRecycle(op);
return;
}
pendingMessages.add(op);
Expand All @@ -2577,6 +2639,7 @@ protected synchronized void processOpSendMsg(OpSendMsg op) {
// 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();
op.writeEventLoop = cnx.ctx().channel().eventLoop();
cnx.ctx().channel().eventLoop().execute(WriteInEventLoopCallback.create(this, cnx, op));
stats.updateNumMsgsSent(op.numMessagesInBatch, op.batchSizeByte);
} else {
Expand Down Expand Up @@ -2664,7 +2727,7 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl<?> latestMsg
+ " but not, which is not expected.");
releaseSemaphoreForSendOp(op);
msgIterator.remove();
op.recycle();
releaseOpCmdAndRecycle(op);
continue;
}
} else if (op.msg == latestMsgAttemptedRegisteredSchema && failedIncompatibleSchema
Expand Down Expand Up @@ -2700,7 +2763,6 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl<?> latestMsg
op.rePopulate.run();
}
msgIterator.remove();
ReferenceCountUtil.safeRelease(op.cmd);
try {
// Need to protect ourselves from any exception being thrown in the future handler from the
// application
Expand All @@ -2711,7 +2773,7 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl<?> latestMsg
.log("Got exception while completing the failed publishing");
}
releaseSemaphoreForSendOp(op);
op.recycle();
releaseOpCmdAndRecycle(op);
continue;
} else if (op.msg.getSchemaState() == None) {
// Event 1-1.
Expand All @@ -2734,6 +2796,7 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl<?> latestMsg
stripChecksum(op);
}
op.cmd.retain();
op.writeEventLoop = cnx.ctx().channel().eventLoop();
log.debug()
.attr("sequenceId", op.sequenceId)
.log("Re-Sending message");
Expand Down
Loading
Loading