Skip to content

[improve][client] Best-effort retry for individual/batch-index acks on send failure when ackReceiptEnabled=false #25525

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
nodece merged 3 commits into from
Apr 24, 2026
Merged

[improve][client] Best-effort retry for individual/batch-index acks on send failure when ackReceiptEnabled=false #25525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
09024b2
[improve][client] Best-effort retry for acks on send failure when ack…
nodece Apr 15, 2026
c22da98
[improve][client] Skip cumulative ack restore on send failure
nodece Apr 15, 2026
be6d53b
Remove unused import
nodece Apr 15, 2026
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
30 changes: 13 additions & 17 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1080,29 +1080,25 @@ CompletableFuture<ProducerResponse> sendRequestWithId(ByteBuf cmd, long requestI
}

private <T> void sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId,
RequestType requestType, boolean flush,
TimedCompletableFuture<T> future) {
RequestType requestType, boolean flush,
TimedCompletableFuture<T> future) {
pendingRequests.put(requestId, future);
if (flush) {
ctx.writeAndFlush(requestMessage).addListener(writeFuture -> {
if (!writeFuture.isSuccess()) {
if (pendingRequests.remove(requestId, future) && !future.isDone()) {
log.warn()
.attr("send", requestType.getDescription())
.exceptionMessage(writeFuture.cause())
.log("Failed to send to broker");
future.completeExceptionally(writeFuture.cause());
}
(flush ? ctx.writeAndFlush(requestMessage) : ctx.write(requestMessage)).addListener(writeFuture -> {
if (!writeFuture.isSuccess()) {
if (pendingRequests.remove(requestId, future) && !future.isDone()) {
log.warn()
.attr("send", requestType.getDescription())
.exceptionMessage(writeFuture.cause())
.log("Failed to send to broker");
future.completeExceptionally(writeFuture.cause());
}
});
} else {
ctx.write(requestMessage, ctx().voidPromise());
}
}
});
requestTimeoutQueue.add(new RequestTime(requestId, requestType));
}

private <T> CompletableFuture<T> sendRequestAndHandleTimeout(ByteBuf requestMessage, long requestId,
RequestType requestType, boolean flush) {
RequestType requestType, boolean flush) {
TimedCompletableFuture<T> future = new TimedCompletableFuture<>();
sendRequestAndHandleTimeout(requestMessage, requestId, requestType, flush, future);
return future;
Expand Down
75 changes: 65 additions & 10 deletions ...src/main/java/org/apache/pulsar/client/impl/PersistentAcknowledgmentsGroupingTracker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
Expand Down Expand Up @@ -279,6 +280,11 @@ private CompletableFuture<Void> doIndividualAckAsync(MessageIdAdv messageId) {
return CompletableFuture.completedFuture(null);
}

@VisibleForTesting
int getPendingIndividualAcksSize() {
return pendingIndividualAcks.size();
}

private CompletableFuture<Void> doIndividualBatchAck(MessageIdAdv batchMessageId,
Map<String, Long> properties) {
if (acknowledgementGroupTimeMicros == 0 || (properties != null && !properties.isEmpty())) {
Expand Down Expand Up @@ -405,7 +411,7 @@ private CompletableFuture<Void> doImmediateBatchIndexAck(MessageIdAdv msgId, int
}

CompletableFuture<Void> completableFuture = newMessageAckCommandAndWrite(cnx, consumer.consumerId,
msgId.getLedgerId(), msgId.getEntryId(), bitSet, ackType, properties, true, null, null);
msgId.getLedgerId(), msgId.getEntryId(), bitSet, ackType, properties, true, null, null, null);
bitSet.recycle();
return completableFuture;
}
Expand Down Expand Up @@ -440,13 +446,15 @@ private void flushAsync(ClientCnx cnx) {
newMessageAckCommandAndWrite(cnx, consumer.consumerId, messageId.getLedgerId(), messageId.getEntryId(),
lastCumulativeAckToFlush.getBitSetRecyclable(), AckType.Cumulative,
Collections.emptyMap(), false,
(TimedCompletableFuture<Void>) this.currentCumulativeAckFuture, null);
(TimedCompletableFuture<Void>) this.currentCumulativeAckFuture, null, null);
this.consumer.unAckedChunkedMessageIdSequenceMap.remove(messageId);
}

// Flush all individual acks
List<Triple<Long, Long, ConcurrentBitSet>> entriesToAck =
new ArrayList<>(pendingIndividualAcks.size() + pendingIndividualBatchIndexAcks.size());
List<MessageIdAdv> individualAcksToFlush = new ArrayList<>(pendingIndividualAcks.size());
Map<MessageIdAdv, MessageIdImpl[]> chunkedMessageIdsToRestore = new HashMap<>();
if (!pendingIndividualAcks.isEmpty()) {
if (Commands.peerSupportsMultiMessageAcknowledgment(cnx.getRemoteEndpointProtocolVersion())) {
// We can send 1 single protobuf command with all individual acks
Expand All @@ -455,11 +463,13 @@ private void flushAsync(ClientCnx cnx) {
if (msgId == null) {
break;
}
individualAcksToFlush.add(msgId);

// if messageId is checked then all the chunked related to that msg also processed so, ack all of
// them
MessageIdImpl[] chunkMsgIds = this.consumer.unAckedChunkedMessageIdSequenceMap.get(msgId);
if (chunkMsgIds != null && chunkMsgIds.length > 1) {
chunkedMessageIdsToRestore.put(msgId, chunkMsgIds);
for (MessageIdImpl cMsgId : chunkMsgIds) {
if (cMsgId != null) {
entriesToAck.add(Triple.of(cMsgId.getLedgerId(), cMsgId.getEntryId(), null));
Expand All @@ -478,21 +488,24 @@ private void flushAsync(ClientCnx cnx) {
if (msgId == null) {
break;
}
individualAcksToFlush.add(msgId);
newMessageAckCommandAndWrite(cnx, consumer.consumerId, msgId.getLedgerId(), msgId.getEntryId(),
null, AckType.Individual, Collections.emptyMap(), false,
null, null);
null, null, () -> restoreIndividualAck(msgId, null));
shouldFlush = true;
}
}
}

List<Map.Entry<MessageIdAdv, ConcurrentBitSet>> batchIndexAcksToFlush = new ArrayList<>();
while (true) {
Map.Entry<MessageIdAdv, ConcurrentBitSet> entry =
pendingIndividualBatchIndexAcks.pollFirstEntry();
if (entry == null) {
// The entry has been removed in a different thread
break;
}
batchIndexAcksToFlush.add(entry);
entriesToAck.add(Triple.of(
entry.getKey().getLedgerId(), entry.getKey().getEntryId(), entry.getValue()));
}
Expand All @@ -501,7 +514,9 @@ private void flushAsync(ClientCnx cnx) {

newMessageAckCommandAndWrite(cnx, consumer.consumerId, 0L, 0L,
null, AckType.Individual, null, true,
(TimedCompletableFuture<Void>) currentIndividualAckFuture, entriesToAck);
(TimedCompletableFuture<Void>) currentIndividualAckFuture, entriesToAck,
() -> restoreIndividualAndBatchIndexAcks(individualAcksToFlush, chunkedMessageIdsToRestore,
batchIndexAcksToFlush));
shouldFlush = true;
}

Expand Down Expand Up @@ -546,19 +561,19 @@ private CompletableFuture<Void> newImmediateAckAndFlush(long consumerId, Message
}
}
completableFuture = newMessageAckCommandAndWrite(cnx, consumer.consumerId, 0L, 0L,
null, ackType, null, true, null, entriesToAck);
null, ackType, null, true, null, entriesToAck, null);
} else {
// if don't support multi message ack, it also support ack receipt, so we should not think about the
// ack receipt in this logic
for (MessageIdImpl cMsgId : chunkMsgIds) {
newMessageAckCommandAndWrite(cnx, consumerId, cMsgId.getLedgerId(), cMsgId.getEntryId(),
bitSet, ackType, map, true, null, null);
bitSet, ackType, map, true, null, null, null);
}
completableFuture = CompletableFuture.completedFuture(null);
}
} else {
completableFuture = newMessageAckCommandAndWrite(cnx, consumerId, msgId.getLedgerId(), msgId.getEntryId(),
bitSet, ackType, map, true, null, null);
bitSet, ackType, map, true, null, null, null);
}
return completableFuture;
}
Expand All @@ -568,7 +583,8 @@ private CompletableFuture<Void> newMessageAckCommandAndWrite(
long entryId, BitSetRecyclable ackSet, AckType ackType,
Map<String, Long> properties, boolean flush,
TimedCompletableFuture<Void> timedCompletableFuture,
List<Triple<Long, Long, ConcurrentBitSet>> entriesToAck) {
List<Triple<Long, Long, ConcurrentBitSet>> entriesToAck,
Runnable writeFailureCallback) {
if (consumer.isAckReceiptEnabled()) {
final long requestId = consumer.getClient().newRequestId();
final ByteBuf cmd;
Expand Down Expand Up @@ -610,14 +626,53 @@ private CompletableFuture<Void> newMessageAckCommandAndWrite(
cmd = Commands.newMultiMessageAck(consumerId, entriesToAck, -1);
}
if (flush) {
cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
if (writeFailureCallback == null) {
cnx.ctx().writeAndFlush(cmd, cnx.ctx().voidPromise());
} else {
cnx.ctx().writeAndFlush(cmd).addListener(writeFuture -> {
if (!writeFuture.isSuccess()) {
writeFailureCallback.run();
}
});
}
} else {
cnx.ctx().write(cmd, cnx.ctx().voidPromise());
if (writeFailureCallback == null) {
cnx.ctx().write(cmd, cnx.ctx().voidPromise());
} else {
cnx.ctx().write(cmd).addListener(writeFuture -> {
if (!writeFuture.isSuccess()) {
writeFailureCallback.run();
}
});
}
}
return CompletableFuture.completedFuture(null);
}
}

private void restoreIndividualAck(MessageIdAdv messageId, @Nullable MessageIdImpl[] chunkMsgIds) {
pendingIndividualAcks.add(messageId);
restoreChunkedMessageIds(messageId, chunkMsgIds);
}

private void restoreIndividualAndBatchIndexAcks(List<MessageIdAdv> messageIds,
Map<MessageIdAdv, MessageIdImpl[]> chunkMsgIds,
List<Map.Entry<MessageIdAdv, ConcurrentBitSet>> batchIndexAcks) {
pendingIndividualAcks.addAll(messageIds);
chunkMsgIds.forEach(this::restoreChunkedMessageIds);
batchIndexAcks.forEach(entry -> pendingIndividualBatchIndexAcks.merge(entry.getKey(), entry.getValue(),
(currentValue, valueToRestore) -> {
currentValue.and(valueToRestore);
return currentValue;
}));
}

private void restoreChunkedMessageIds(MessageIdAdv messageId, @Nullable MessageIdImpl[] chunkMsgIds) {
if (chunkMsgIds != null) {
consumer.unAckedChunkedMessageIdSequenceMap.putIfAbsent(messageId, chunkMsgIds);
}
}

public Optional<Lock> acquireReadLock() {
Optional<Lock> optionalLock = Optional.ofNullable(consumer.isAckReceiptEnabled() ? lock.readLock() : null);
optionalLock.ifPresent(Lock::lock);
Expand Down
93 changes: 91 additions & 2 deletions ...ient/src/test/java/org/apache/pulsar/client/impl/AcknowledgementsGroupingTrackerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;

import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
Expand All @@ -28,9 +29,13 @@
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import java.util.ArrayList;
import java.util.BitSet;
import java.util.Collections;
Expand All @@ -56,6 +61,8 @@ public class AcknowledgementsGroupingTrackerTest {
private ConsumerImpl<?> consumer;
private EventLoopGroup eventLoopGroup;
private AtomicBoolean returnCnx = new AtomicBoolean(true);
private ChannelHandlerContext successCtx;
private AtomicBoolean failAckCommandSend = new AtomicBoolean(false);

@BeforeClass
public void setup() throws NoSuchFieldException, IllegalAccessException {
Expand All @@ -70,9 +77,9 @@ public void setup() throws NoSuchFieldException, IllegalAccessException {
doReturn(new ConsumerStatsRecorderImpl()).when(consumer).getStats();
doReturn(UnAckedMessageTracker.UNACKED_MESSAGE_TRACKER_DISABLED)
.when(consumer).getUnAckedMessageTracker();
ChannelHandlerContext ctx = ClientTestFixtures.mockChannelHandlerContext();
successCtx = ClientTestFixtures.mockChannelHandlerContext();
doAnswer(invocation -> returnCnx.get() ? cnx : null).when(consumer).getClientCnx();
doReturn(ctx).when(cnx).ctx();
doReturn(successCtx).when(cnx).ctx();
}

@DataProvider(name = "isNeedReceipt")
Expand Down Expand Up @@ -323,6 +330,60 @@ public void testAckTrackerMultiAck(boolean isNeedReceipt) {
tracker.close();
}

@Test
public void testFlushRetainsPendingIndividualAckOnSendFailureWithoutAckReceipt() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
conf.setAckReceiptEnabled(false);
doReturn(false).when(consumer).isAckReceiptEnabled();
PersistentAcknowledgmentsGroupingTracker tracker =
new PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);

MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
tracker.addAcknowledgment(msg1, AckType.Individual, Collections.emptyMap());
assertEquals(tracker.getPendingIndividualAcksSize(), 1);

doReturn(createFailedChannelHandlerContext()).when(cnx).ctx();

tracker.flush();

assertTrue(tracker.isDuplicate(msg1));
assertEquals(tracker.getPendingIndividualAcksSize(), 1);

doReturn(successCtx).when(cnx).ctx();

tracker.flush();

assertFalse(tracker.isDuplicate(msg1));
assertEquals(tracker.getPendingIndividualAcksSize(), 0);
tracker.close();
}

@Test
public void testFlushFailsAckFutureOnSendFailureWithAckReceipt() throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
conf.setAcknowledgementsGroupTimeMicros(TimeUnit.SECONDS.toMicros(10));
conf.setAckReceiptEnabled(true);
doReturn(true).when(consumer).isAckReceiptEnabled();
PersistentAcknowledgmentsGroupingTracker tracker =
new PersistentAcknowledgmentsGroupingTracker(consumer, conf, eventLoopGroup);

MessageIdImpl msg1 = new MessageIdImpl(5, 1, 0);
CompletableFuture<Void> ackFuture =
tracker.addAcknowledgment(msg1, AckType.Individual, Collections.emptyMap());
assertEquals(tracker.getPendingIndividualAcksSize(), 1);

failAckCommandSend.set(true);
tracker.flush();

assertTrue(ackFuture.isCompletedExceptionally());
assertFalse(tracker.isDuplicate(msg1));
assertEquals(tracker.getPendingIndividualAcksSize(), 0);

failAckCommandSend.set(false);
tracker.close();
}

@Test(dataProvider = "isNeedReceipt")
public void testBatchAckTrackerMultiAck(boolean isNeedReceipt) throws Exception {
ConsumerConfigurationData<?> conf = new ConsumerConfigurationData<>();
Expand Down Expand Up @@ -463,12 +524,40 @@ public ClientCnxTest(ClientConfigurationData conf, EventLoopGroup eventLoopGroup

@Override
public CompletableFuture<Void> newAckForReceipt(ByteBuf request, long requestId) {
if (failAckCommandSend.get()) {
return CompletableFuture.failedFuture(new RuntimeException("ack send failed"));
}
return CompletableFuture.completedFuture(null);
}

@Override
public void newAckForReceiptWithFuture(ByteBuf request, long requestId,
TimedCompletableFuture<Void> future) {
if (failAckCommandSend.get()) {
future.completeExceptionally(new RuntimeException("ack send failed"));
}
}
}

private ChannelHandlerContext createFailedChannelHandlerContext() {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
ChannelFuture listenerFuture = mock(ChannelFuture.class);
ChannelFuture failedFuture = mock(ChannelFuture.class);
when(failedFuture.isSuccess()).thenReturn(false);
when(failedFuture.cause()).thenReturn(new RuntimeException("ack send failed"));
doAnswer(invocation -> {
GenericFutureListener<Future<Void>> listener = invocation.getArgument(0);
listener.operationComplete(failedFuture);
return listenerFuture;
}).when(listenerFuture).addListener(any());
doAnswer(invocation -> {
ReferenceCountUtil.release(invocation.getArgument(0));
return listenerFuture;
}).when(ctx).write(any());
doAnswer(invocation -> {
ReferenceCountUtil.release(invocation.getArgument(0));
return listenerFuture;
}).when(ctx).writeAndFlush(any());
return ctx;
}
}
1 change: 1 addition & 0 deletions ...-client/src/test/java/org/apache/pulsar/client/impl/ClientCnxRequestTimeoutQueueTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@ void setupClientCnx() throws Exception {
ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
Channel channel = mock(Channel.class);
when(ctx.writeAndFlush(any())).thenAnswer(args -> mock(ChannelFuture.class));
when(ctx.write(any())).thenAnswer(args -> mock(ChannelFuture.class));
when(ctx.channel()).thenReturn(channel);
when(channel.remoteAddress()).thenReturn(new InetSocketAddress(1234));
cnx.channelActive(ctx);
Expand Down
5 changes: 5 additions & 0 deletions pulsar-client/src/test/java/org/apache/pulsar/client/impl/ClientTestFixtures.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,11 @@ public static ChannelHandlerContext mockChannelHandlerContext() {
}).when(listenerFuture).addListener(any());

// handle write and writeAndFlush methods so that the input message is released
doAnswer(invocation -> {
Object msg = invocation.getArgument(0);
ReferenceCountUtil.release(msg);
return listenerFuture;
}).when(ctx).write(any());
doAnswer(invocation -> {
Object msg = invocation.getArgument(0);
ReferenceCountUtil.release(msg);
Expand Down
Loading