Fix zero queue consumer and table view reader

apache · Feb 26, 2025 · effd9c7 · effd9c7
1 parent 29cb448
commit effd9c7
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 5 deletions.
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
@@ -88,6 +88,7 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 
 	if options.EnableZeroQueueConsumer {
 		options.ReceiverQueueSize = 0
+		options.StartMessageIDInclusive = true
 	}
 
 	if options.Interceptors == nil {

diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
@@ -2305,12 +2305,14 @@ func convertToMessageID(id *pb.MessageIdData) *trackingMessageID {
 
 	msgID := &trackingMessageID{
 		messageID: &messageID{
-			ledgerID: int64(*id.LedgerId),
-			entryID:  int64(*id.EntryId),
+			ledgerID:  int64(id.GetLedgerId()),
+			entryID:   int64(id.GetEntryId()),
+			batchIdx:  -1,
+			batchSize: id.GetBatchSize(),
 		},
 	}
-	if id.BatchIndex != nil {
-		msgID.batchIdx = *id.BatchIndex
+	if id.GetBatchSize() > 1 {
+		msgID.batchIdx = id.GetBatchIndex()
 	}
 
 	return msgID

diff --git a/pulsar/consumer_zero_queue_test.go b/pulsar/consumer_zero_queue_test.go
@@ -474,7 +474,6 @@ func TestZeroQueueConsumer_Seek(t *testing.T) {
 		Topic:                   topicName,
 		EnableZeroQueueConsumer: true,
 		SubscriptionName:        "sub-1",
-		StartMessageIDInclusive: true,
 	})
 	assert.Nil(t, err)
 	_, ok := consumer.(*zeroQueueConsumer)