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/07/22 10:02:38 UTC
(pulsar) 03/15: [fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ (#26135)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 0d7115fd3da0b0dd773c4c1b59cc97e1c4bf28d9
Author: Yike Xiao <km...@live.com>
AuthorDate: Fri Jul 3 21:34:38 2026 +0800
[fix][client] Sync ackSet in client with broker to stop acked messages reaching the DLQ (#26135)
(cherry picked from commit e8df873ddfde4d8b5a38a49e4b8765c693d65ab9)
---
.../pulsar/client/api/DeadLetterTopicTest.java | 71 ++++++++++++++++++++++
.../apache/pulsar/client/impl/ConsumerImpl.java | 22 ++++---
.../client/impl/MessagePayloadContextImpl.java | 9 ++-
.../pulsar/client/impl/ZeroQueueConsumerImpl.java | 2 +-
.../pulsar/client/impl/ConsumerImplTest.java | 57 +++++++++++++++++
5 files changed, 148 insertions(+), 13 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
index 31c92583a907..8ce96bea214a 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DeadLetterTopicTest.java
@@ -1677,5 +1677,76 @@ public class DeadLetterTopicTest extends SharedPulsarBaseTest {
verify(client, times(0)).getPartitionedTopicMetadata(anyString(), anyBoolean(), anyBoolean());
}
+ @Test
+ public void testAckedBatchMessageNotSentToDeadLetterTopicOnFinalRedeliveryRound() throws Exception {
+ final String topic = newTopicName();
+ final int maxRedeliveryCount = 3;
+ final int batchSize = 5;
+ final String subscriptionName = "my-subscription";
+
+ Consumer<byte[]> consumer = pulsarClient.newConsumer(Schema.BYTES)
+ .topic(topic)
+ .subscriptionName(subscriptionName)
+ .subscriptionType(SubscriptionType.Shared)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .enableBatchIndexAcknowledgment(true)
+ .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(maxRedeliveryCount).build())
+ .ackTimeout(1, TimeUnit.SECONDS)
+ .receiverQueueSize(100)
+ .subscribe();
+
+ @Cleanup
+ PulsarClient newPulsarClient = newPulsarClient();
+ Consumer<byte[]> deadLetterConsumer = newPulsarClient.newConsumer(Schema.BYTES)
+ .topic(topic + "-" + subscriptionName + "-DLQ")
+ .subscriptionName(subscriptionName)
+ .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
+ .subscribe();
+
+ Producer<byte[]> producer = pulsarClient.newProducer(Schema.BYTES)
+ .topic(topic)
+ .enableBatching(true)
+ .batchingMaxPublishDelay(1, TimeUnit.SECONDS)
+ .create();
+ List<CompletableFuture<MessageId>> sendFutures = new ArrayList<>();
+ for (int i = 0; i < batchSize; i++) {
+ sendFutures.add(producer.newMessage().value(("message-" + i).getBytes()).sendAsync());
+ }
+ for (CompletableFuture<MessageId> future : sendFutures) {
+ future.get();
+ }
+ producer.close();
+
+ // Batch indices 1 and 2 are deliberately left to time out for the first `maxRedeliveryCount`
+ // rounds, then explicitly acked on the final round (redeliveryCount == maxRedeliveryCount) --
+ // the app's last chance to prevent them from being routed to the DLQ. The other 3 messages in
+ // the batch are acked immediately on the first delivery.
+ // Expected deliveries: (batchSize - 2) once each, plus indices 1 and 2 redelivered on every
+ // round from 0 through maxRedeliveryCount inclusive.
+ final int expectedDeliveries = (batchSize - 2) + 2 * (maxRedeliveryCount + 1);
+ int received = 0;
+ while (received < expectedDeliveries) {
+ Message<byte[]> message = consumer.receive(5, TimeUnit.SECONDS);
+ assertNotNull(message, "consumer should keep receiving messages until the batch settles");
+ received++;
+ MessageIdAdv messageId = (MessageIdAdv) message.getMessageId();
+ int batchIndex = messageId.getBatchIndex();
+ int redeliveryCount = message.getRedeliveryCount();
+ if ((batchIndex == 1 || batchIndex == 2) && redeliveryCount < maxRedeliveryCount) {
+ // Let it time out instead of acking.
+ continue;
+ }
+ consumer.acknowledge(message);
+ }
+
+ // No message should ever be routed to the DLQ, since every message was explicitly acked at or
+ // before its final allowed redelivery round.
+ Message<byte[]> deadLetterMessage = deadLetterConsumer.receive(5, TimeUnit.SECONDS);
+ assertNull(deadLetterMessage, "no message should have been routed to the DLQ, "
+ + "but received: " + deadLetterMessage);
+
+ deadLetterConsumer.close();
+ consumer.close();
+ }
}
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 54ef0db71aa0..0b5ecf7e13e6 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
@@ -23,6 +23,7 @@ import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
import static org.apache.pulsar.common.protocol.Commands.hasChecksum;
import static org.apache.pulsar.common.protocol.Commands.serializeWithSize;
import static org.apache.pulsar.common.util.Runnables.catchingAndLoggingThrowables;
+import static org.apache.pulsar.common.util.SafeCollectionUtils.longArrayToList;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ComparisonChain;
import com.google.common.collect.Iterables;
@@ -134,7 +135,6 @@ import org.apache.pulsar.common.util.Backoff;
import org.apache.pulsar.common.util.CompletableFutureCancellationHandler;
import org.apache.pulsar.common.util.ExceptionHandler;
import org.apache.pulsar.common.util.FutureUtil;
-import org.apache.pulsar.common.util.SafeCollectionUtils;
import org.apache.pulsar.common.util.collections.BitSetRecyclable;
import org.apache.pulsar.common.util.collections.ConcurrentBitSet;
import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
@@ -142,6 +142,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandler.Connection {
+ private static final long[] EMPTY_ACK_SET = new long[0];
private static final int MAX_REDELIVER_UNACKNOWLEDGED = 1000;
final long consumerId;
@@ -1429,11 +1430,11 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
void messageReceived(CommandMessage cmdMessage, ByteBuf headersAndPayload, ClientCnx cnx) {
- List<Long> ackSet = Collections.emptyList();
+ long[] ackSet = EMPTY_ACK_SET;
if (cmdMessage.getAckSetsCount() > 0) {
- ackSet = new ArrayList<>(cmdMessage.getAckSetsCount());
+ ackSet = new long[cmdMessage.getAckSetsCount()];
for (int i = 0; i < cmdMessage.getAckSetsCount(); i++) {
- ackSet.add(cmdMessage.getAckSetAt(i));
+ ackSet[i] = cmdMessage.getAckSetAt(i);
}
}
int redeliveryCount = cmdMessage.getRedeliveryCount();
@@ -1503,7 +1504,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
if (conf.getPayloadProcessor() != null) {
// uncompressedPayload is released in this method so we don't need to call release() again
processPayloadByProcessor(brokerEntryMetadata, msgMetadata,
- uncompressedPayload, msgId, schema, redeliveryCount, ackSet, consumerEpoch);
+ uncompressedPayload, msgId, schema, redeliveryCount, longArrayToList(ackSet), consumerEpoch);
return;
}
@@ -1781,7 +1782,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
}
void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata, MessageMetadata msgMetadata,
- int redeliveryCount, List<Long> ackSet, ByteBuf uncompressedPayload,
+ int redeliveryCount, long[] ackSet, ByteBuf uncompressedPayload,
MessageIdData messageId, ClientCnx cnx, long consumerEpoch,
boolean isEncrypted) {
int batchSize = msgMetadata.getNumMessagesInBatch();
@@ -1796,8 +1797,9 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
BitSet ackSetInMessageId = BatchMessageIdImpl.newAckSet(batchSize);
BitSetRecyclable ackBitSet = null;
- if (ackSet != null && ackSet.size() > 0) {
- ackBitSet = BitSetRecyclable.valueOf(SafeCollectionUtils.longListToArray(ackSet));
+ if (ackSet != null && ackSet.length > 0) {
+ ackBitSet = BitSetRecyclable.valueOf(ackSet);
+ ackSetInMessageId.and(BitSet.valueOf(ackSet));
}
SingleMessageMetadata singleMessageMetadata = new SingleMessageMetadata();
@@ -2688,7 +2690,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
final ByteBuf seek;
if (msgId.getFirstChunkMessageId() != null) {
seek = Commands.newSeek(consumerId, requestId, firstChunkMsgId.getLedgerId(),
- firstChunkMsgId.getEntryId(), new long[0]);
+ firstChunkMsgId.getEntryId(), EMPTY_ACK_SET);
} else {
final long[] ackSetArr;
if (MessageIdAdvUtils.isBatch(msgId)) {
@@ -2698,7 +2700,7 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
ackSetArr = ackSet.toLongArray();
ackSet.recycle();
} else {
- ackSetArr = new long[0];
+ ackSetArr = EMPTY_ACK_SET;
}
seek = Commands.newSeek(consumerId, requestId, msgId.getLedgerId(), msgId.getEntryId(), ackSetArr);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessagePayloadContextImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessagePayloadContextImpl.java
index 954afb571311..0220f1154d16 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessagePayloadContextImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessagePayloadContextImpl.java
@@ -75,8 +75,13 @@ public class MessagePayloadContextImpl implements MessagePayloadContext {
context.consumer = consumer;
context.redeliveryCount = redeliveryCount;
context.ackSetInMessageId = BatchMessageIdImpl.newAckSet(context.getNumMessages());
- context.ackBitSet = (ackSet != null && ackSet.size() > 0)
- ? BitSetRecyclable.valueOf(SafeCollectionUtils.longListToArray(ackSet))
+ boolean isAckSetNotEmpty = ackSet != null && ackSet.size() > 0;
+ long[] ackSetArray = SafeCollectionUtils.longListToArray(ackSet);
+ if (isAckSetNotEmpty) {
+ context.ackSetInMessageId.and(BitSet.valueOf(ackSetArray));
+ }
+ context.ackBitSet = isAckSetNotEmpty
+ ? BitSetRecyclable.valueOf(ackSetArray)
: null;
return context;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
index 02851ac36fc4..26fb1e0c8caa 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ZeroQueueConsumerImpl.java
@@ -200,7 +200,7 @@ public class ZeroQueueConsumerImpl<T> extends ConsumerImpl<T> {
@Override
void receiveIndividualMessagesFromBatch(BrokerEntryMetadata brokerEntryMetadata, MessageMetadata msgMetadata,
- int redeliveryCount, List<Long> ackSet, ByteBuf uncompressedPayload,
+ int redeliveryCount, long[] ackSet, ByteBuf uncompressedPayload,
MessageIdData messageId, ClientCnx cnx, long consumerEpoch,
boolean isEncrypted) {
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 005a4cd5ba64..a427e35a4c45 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
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import static org.apache.pulsar.common.protocol.Commands.DEFAULT_CONSUMER_EPOCH;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.any;
@@ -31,6 +32,10 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertTrue;
import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
@@ -38,17 +43,22 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import lombok.Cleanup;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageIdAdv;
+import org.apache.pulsar.client.api.MessagePayload;
import org.apache.pulsar.client.api.Messages;
import org.apache.pulsar.client.api.PulsarClient;
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.ConsumerConfigurationData;
import org.apache.pulsar.client.impl.conf.TopicConsumerConfigurationData;
import org.apache.pulsar.client.util.ExecutorProvider;
import org.apache.pulsar.client.util.ScheduledExecutorProvider;
+import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.util.Backoff;
import org.awaitility.Awaitility;
import org.testng.Assert;
@@ -336,4 +346,51 @@ public class ConsumerImplTest {
"Hint must reflect the post-enqueue state (pipeline had >=1 message); "
+ "a concurrent drain of the just-enqueued message must not clear it.");
}
+
+ @Test(invocationTimeOut = 1000)
+ public void testGetMessageAtSyncsAckSetInMessageIdWithBrokerAckSet() {
+ // Regression test for MessagePayloadContextImpl#getMessageAt: the BatchMessageIdImpl handed
+ // back to the caller carries a shared ackSetInMessageId bitset that must be seeded from the
+ // broker-reported ackSet, not a fresh "all unacked" bitset. Otherwise indices the broker
+ // already knows are acked would be reported as still-outstanding in the returned MessageId,
+ // which is the same root cause that let acked batch messages leak into the DLQ (see
+ // ConsumerImpl#receiveIndividualMessagesFromBatch and its ackSetInMessageId.and(...) fix).
+ final int batchSize = 3;
+ MessageMetadata messageMetadata = new MessageMetadata()
+ .setProducerName("test-producer")
+ .setSequenceId(0)
+ .setPublishTime(System.currentTimeMillis())
+ .setNumMessagesInBatch(batchSize);
+
+ // Broker reports index 0 as already acked (bit cleared); indices 1 and 2 are still
+ // outstanding (bits set). This mirrors the ackSet the broker attaches on redelivery.
+ BitSet brokerAckSet = new BitSet(batchSize);
+ brokerAckSet.set(1);
+ brokerAckSet.set(2);
+ List<Long> ackSet = Arrays.stream(brokerAckSet.toLongArray()).boxed().collect(Collectors.toList());
+
+ MessageIdImpl messageId = new MessageIdImpl(1L, 2L, -1);
+ MessagePayloadContextImpl context = MessagePayloadContextImpl.get(
+ null, messageMetadata, messageId, consumer, 0, ackSet, DEFAULT_CONSUMER_EPOCH);
+ MessagePayload payload0 = MessagePayloadImpl.create(Unpooled.wrappedBuffer(new byte[]{0}));
+ MessagePayload payload1 = MessagePayloadImpl.create(Unpooled.wrappedBuffer(new byte[]{1}));
+ try {
+ // Index 0 is already acked per the broker, so it must not be redelivered to the app.
+ Assert.assertNull(context.getMessageAt(0, batchSize, payload0, false, Schema.BYTES));
+
+ Message<byte[]> message1 = context.getMessageAt(1, batchSize, payload1, false, Schema.BYTES);
+ Assert.assertNotNull(message1);
+
+ BitSet ackSetInMessageId = ((MessageIdAdv) message1.getMessageId()).getAckSet();
+ Assert.assertFalse(ackSetInMessageId.get(0),
+ "index 0 was already acked by the broker, so the returned MessageId's ackSet "
+ + "must reflect it as acked, not fall back to the default all-unacked state");
+ Assert.assertTrue(ackSetInMessageId.get(1), "index 1 is still outstanding");
+ Assert.assertTrue(ackSetInMessageId.get(2), "index 2 is still outstanding");
+ } finally {
+ payload0.release();
+ payload1.release();
+ context.recycle();
+ }
+ }
}