From b5499a0899fed107cb674acabb5633653ae14754 Mon Sep 17 00:00:00 2001 From: VitorNathanG Date: Thu, 2 Jul 2026 15:47:50 -0300 Subject: [PATCH] [Issue 1519][Consumer] Do not deduplicate messages on non-persistent topics Non-persistent topics are not stored in BookKeeper, so the broker assigns every message the same placeholder MessageID (ledgerId=0, entryId=0). The grouping acknowledgment tracker deduplicates incoming messages by MessageID via isDuplicate(); once the first message on such a topic is acknowledged, its (0,0) key is recorded in the tracker and every subsequent message is judged a duplicate and dropped before it reaches Receive(). A consumer that acks each message before receiving the next (for example a synchronous request-reply loop) therefore receives only the first message and then blocks forever. A streaming consumer loses a large, timing-dependent fraction of its messages. Force the immediate ack tracker, whose isDuplicate() always returns false, for non-persistent topics. This matches the Java and C++ clients, which use a dedicated non-persistent ack tracker that never deduplicates. There is nothing to deduplicate against on a non-persistent topic in any case, since those topics never redeliver. The MessageID dedup was introduced in #957, which ported the grouping-ack tracker from the Java client but not its non-persistent counterpart. The Java client had the identical bug and fixed it the same way; see apache/pulsar#1967. Fixes #1519 --- pulsar/ack_grouping_tracker.go | 10 ++++++ pulsar/ack_grouping_tracker_test.go | 19 +++++++++++ pulsar/consumer_partition.go | 16 ++++++++- pulsar/consumer_test.go | 50 +++++++++++++++++++++++++++++ 4 files changed, 94 insertions(+), 1 deletion(-) diff --git a/pulsar/ack_grouping_tracker.go b/pulsar/ack_grouping_tracker.go index bb9059ac06..cc66e6e8c3 100644 --- a/pulsar/ack_grouping_tracker.go +++ b/pulsar/ack_grouping_tracker.go @@ -22,6 +22,7 @@ import ( "sync/atomic" "time" + "github.com/apache/pulsar-client-go/pulsar/internal" pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto" "github.com/bits-and-blooms/bitset" ) @@ -40,6 +41,15 @@ type ackGroupingTracker interface { close() } +func isNonPersistentTopic(topic string) bool { + tn, err := internal.ParseTopicName(topic) + if err != nil { + // On a parse error, keep the default (persistent) behavior. + return false + } + return tn.Domain == "non-persistent" +} + func newAckGroupingTracker(options *AckGroupingOptions, ackIndividual func(id MessageID), ackCumulative func(id MessageID), diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go index 57adf8887a..d7395d4118 100644 --- a/pulsar/ack_grouping_tracker_test.go +++ b/pulsar/ack_grouping_tracker_test.go @@ -287,3 +287,22 @@ func TestTrackerPendingAcks(t *testing.T) { assert.True(t, found) assert.Equal(t, 0, len(ackSet)) // all messages in the batch are acknowledged } + +func TestIsNonPersistentTopic(t *testing.T) { + tests := []struct { + name string + topic string + want bool + }{ + {"non-persistent scheme", "non-persistent://public/default/my-topic", true}, + {"persistent scheme", "persistent://public/default/my-topic", false}, + {"short name defaults to persistent", "my-topic", false}, + {"namespace path defaults to persistent", "public/default/my-topic", false}, + {"unparseable treated as persistent", "://invalid", false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, isNonPersistentTopic(tt.topic)) + }) + } +} diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go index 08e1b36f27..2da4cec444 100644 --- a/pulsar/consumer_partition.go +++ b/pulsar/consumer_partition.go @@ -448,7 +448,21 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon pc.availablePermits = &availablePermits{pc: pc} pc.chunkedMsgCtxMap = newChunkedMsgCtxMap(options.maxPendingChunkedMessage, pc) pc.unAckChunksTracker = newUnAckChunksTracker(pc) - pc.ackGroupingTracker = newAckGroupingTracker(options.ackGroupingOptions, + ackGroupingOptions := options.ackGroupingOptions + if isNonPersistentTopic(pc.topic) { + // Non-persistent topics are not stored in BookKeeper, so the broker + // assigns every message the same placeholder MessageID (ledgerId=0, + // entryId=0). The grouping ack tracker's isDuplicate() dedupes by + // MessageID, so once the first message is acked, every later message + // looks like a duplicate of it and is silently dropped before reaching + // the application. Force the immediate (no-op isDuplicate) tracker for + // non-persistent topics, matching the Java and C++ clients, which use a + // dedicated non-persistent ack tracker that never dedupes. There is + // nothing to dedupe against anyway: non-persistent topics do not + // redeliver. + ackGroupingOptions = &AckGroupingOptions{MaxSize: 1} + } + pc.ackGroupingTracker = newAckGroupingTracker(ackGroupingOptions, func(id MessageID) { pc.sendIndividualAck(id) }, func(id MessageID) { pc.sendCumulativeAck(id) }, func(ids []*pb.MessageIdData) { diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go index 077799e03c..513f45502e 100644 --- a/pulsar/consumer_test.go +++ b/pulsar/consumer_test.go @@ -6508,3 +6508,53 @@ func TestConsumerWithDLQRetryTopicNoGetPartitionedTopicMetadata(t *testing.T) { "GetPartitionedTopicMetadata should not be called with old Retry topic when custom Retry topic is provided") } } + +// TestNonPersistentTopicReceiveAllMessages is a regression test for the bug +// where a consumer on a non-persistent topic received only the first message +// and then silently dropped the rest. See the non-persistent guard in +// newPartitionConsumer. +func TestNonPersistentTopicReceiveAllMessages(t *testing.T) { + client, err := NewClient(ClientOptions{URL: lookupURL}) + assert.Nil(t, err) + defer client.Close() + + topic := newTopicName() + topic = "non-persistent://public/default/" + topic + + consumer, err := client.Subscribe(ConsumerOptions{ + Topic: topic, + SubscriptionName: "sub-non-persistent", + Type: Shared, + }) + assert.Nil(t, err) + defer consumer.Close() + + producer, err := client.CreateProducer(ProducerOptions{ + Topic: topic, + DisableBatching: true, + }) + assert.Nil(t, err) + defer producer.Close() + + // Synchronous send-receive-ack loop: this is the pattern that + // deterministically triggered the "only the first message" bug, because + // each message arrives within the ack-grouping flush window right after the + // previous one was acked. + const total = 10 + for i := 0; i < total; i++ { + _, err := producer.Send(context.Background(), &ProducerMessage{ + Payload: []byte(fmt.Sprintf("msg-%d", i)), + }) + assert.Nil(t, err) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + msg, err := consumer.Receive(ctx) + cancel() + assert.Nil(t, err, "did not receive message %d (dropped as a bogus duplicate?)", i) + if err != nil { + return + } + assert.Equal(t, fmt.Sprintf("msg-%d", i), string(msg.Payload())) + consumer.Ack(msg) + } +}