Repository: apache/pulsar-client-go
Pull request #1221 (2024-05-22 14:09:47)

[Improve] PIP-313 Add GetLastMessageIDs API
### Motivation
To keep consistent with the Java client.

Releted PR: https://github.com/apache/pulsar/pull/20040


### Modifications

- Add `GetLastMessageIDs`api for consumer.

### Verifying this change

- [x] Make sure that the change passes the CI checks.


### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (no)
  - The public API: (yes)
  - The schema: (no)
  - The default values of configurations: (no)
  - The wire protocol: (yes)

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (GoDocs)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation


crossoverJie: @RobertIndie The integration test failed, maybe we can try it again, it passes in my own repository.

https://github.com/crossoverJie/pulsar-client-go/actions/runs/9253419225/job/25463657165

Patch:
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index 3161af5..0b70a16 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -278,6 +278,11 @@ type Consumer interface {
 	// where more than one consumer are currently connected.
 	UnsubscribeForce() error
 
+	// GetLastMessageIDs get all the last message id of the topics the consumer subscribed.
+	//
+	// The list of MessageID instances of all the topics that the consumer subscribed
+	GetLastMessageIDs() ([]TopicMessageID, error)
+
 	// Receive a single message.
 	// This calls blocks until a message is available.
 	Receive(context.Context) (Message, error)
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index b22eeaa..71ac323 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -472,6 +472,19 @@ func (c *consumer) unsubscribe(force bool) error {
 	return nil
 }
 
+func (c *consumer) GetLastMessageIDs() ([]TopicMessageID, error) {
+	ids := make([]TopicMessageID, 0)
+	for _, pc := range c.consumers {
+		id, err := pc.getLastMessageID()
+		tm := &topicMessageID{topic: pc.topic, track: id}
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, tm)
+	}
+	return ids, nil
+}
+
 func (c *consumer) Receive(ctx context.Context) (message Message, err error) {
 	for {
 		select {
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index eaf42ad..020e38a 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -109,6 +109,18 @@ func (c *multiTopicConsumer) UnsubscribeForce() error {
 	return errs
 }
 
+func (c *multiTopicConsumer) GetLastMessageIDs() ([]TopicMessageID, error) {
+	ids := make([]TopicMessageID, 0)
+	for _, c := range c.consumers {
+		id, err := c.GetLastMessageIDs()
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, id...)
+	}
+	return ids, nil
+}
+
 func (c *multiTopicConsumer) Receive(ctx context.Context) (message Message, err error) {
 	for {
 		select {
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index 87acfe3..58cfa80 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -148,6 +148,18 @@ func (c *regexConsumer) UnsubscribeForce() error {
 	return errs
 }
 
+func (c *regexConsumer) GetLastMessageIDs() ([]TopicMessageID, error) {
+	ids := make([]TopicMessageID, 0)
+	for _, c := range c.consumers {
+		id, err := c.GetLastMessageIDs()
+		if err != nil {
+			return nil, err
+		}
+		ids = append(ids, id...)
+	}
+	return ids, nil
+}
+
 func (c *regexConsumer) Receive(ctx context.Context) (message Message, err error) {
 	for {
 		select {
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 89a709f..478b1af 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -491,3 +491,40 @@ func (id chunkMessageID) Serialize() []byte {
 	data, _ := proto.Marshal(msgID)
 	return data
 }
+
+type topicMessageID struct {
+	track *trackingMessageID
+	topic string
+}
+
+func (t *topicMessageID) Serialize() []byte {
+	return t.track.Serialize()
+}
+
+func (t *topicMessageID) LedgerID() int64 {
+	return t.track.LedgerID()
+}
+
+func (t *topicMessageID) EntryID() int64 {
+	return t.track.EntryID()
+}
+
+func (t *topicMessageID) BatchIdx() int32 {
+	return t.track.BatchIdx()
+}
+
+func (t *topicMessageID) PartitionIdx() int32 {
+	return t.track.PartitionIdx()
+}
+
+func (t *topicMessageID) BatchSize() int32 {
+	return t.track.BatchSize()
+}
+
+func (t *topicMessageID) String() string {
+	return t.track.String()
+}
+
+func (t *topicMessageID) Topic() string {
+	return t.topic
+}
diff --git a/pulsar/message.go b/pulsar/message.go
index 83afd3f..4e8d60d 100644
--- a/pulsar/message.go
+++ b/pulsar/message.go
@@ -166,6 +166,12 @@ type MessageID interface {
 	String() string
 }
 
+// TopicMessageID defined the correspondence between topic and MessageID.
+type TopicMessageID interface {
+	MessageID
+	Topic() string
+}
+
 // DeserializeMessageID reconstruct a MessageID object from its serialized representation
 func DeserializeMessageID(data []byte) (MessageID, error) {
 	return deserializeMessageID(data)


Test patch:
diff --git a/pulsar/consumer_multitopic_test.go b/pulsar/consumer_multitopic_test.go
index 58ad095..7c6b898 100644
--- a/pulsar/consumer_multitopic_test.go
+++ b/pulsar/consumer_multitopic_test.go
@@ -22,6 +22,10 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/apache/pulsar-client-go/pulsaradmin"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+
 	"github.com/stretchr/testify/assert"
 )
 
@@ -137,3 +141,80 @@ func TestMultiTopicConsumerForceUnsubscribe(t *testing.T) {
 	err = consumer.UnsubscribeForce()
 	assert.Error(t, err)
 }
+
+func TestMultiTopicGetLastMessageIDs(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic1Partition, topic2Partition, topic3Partition := 1, 2, 3
+
+	topic1 := newTopicName()
+	err = createPartitionedTopic(topic1, topic1Partition)
+	assert.Nil(t, err)
+
+	topic2 := newTopicName()
+	err = createPartitionedTopic(topic2, topic2Partition)
+	assert.Nil(t, err)
+
+	topic3 := newTopicName()
+	err = createPartitionedTopic(topic3, topic3Partition)
+	assert.Nil(t, err)
+
+	topics := []string{topic1, topic2, topic3}
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topics:           topics,
+		SubscriptionName: "my-sub",
+		Type:             Shared,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	// produce messages
+	totalMessage := 30
+	for i, topic := range topics {
+		p, err := client.CreateProducer(ProducerOptions{
+			Topic:           topic,
+			DisableBatching: true,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = genMessages(p, totalMessage, func(idx int) string {
+			return fmt.Sprintf("topic-%d-hello-%d", i+1, idx)
+		})
+		p.Close()
+		if err != nil {
+			assert.Nil(t, err)
+		}
+	}
+
+	// create admin
+	admin, err := pulsaradmin.NewClient(&config.Config{})
+	assert.Nil(t, err)
+
+	topicMessageIDs, err := consumer.GetLastMessageIDs()
+	assert.Nil(t, err)
+	assert.Equal(t, topic1Partition+topic2Partition+topic3Partition, len(topicMessageIDs))
+	for _, id := range topicMessageIDs {
+		if strings.Contains(id.Topic(), topic1) {
+			assert.Equal(t, int(id.EntryID()), totalMessage/topic1Partition-1)
+		} else if strings.Contains(id.Topic(), topic2) {
+			assert.Equal(t, int(id.EntryID()), totalMessage/topic2Partition-1)
+		} else if strings.Contains(id.Topic(), topic3) {
+			assert.Equal(t, int(id.EntryID()), totalMessage/topic3Partition-1)
+		}
+
+		topicName, err := utils.GetTopicName(id.Topic())
+		assert.Nil(t, err)
+		messages, err := admin.Subscriptions().GetMessagesByID(*topicName, id.LedgerID(), id.EntryID())
+		assert.Nil(t, err)
+		assert.Equal(t, 1, len(messages))
+
+	}
+
+}
diff --git a/pulsar/consumer_regex_test.go b/pulsar/consumer_regex_test.go
index 6bc3f70..28ba4f7 100644
--- a/pulsar/consumer_regex_test.go
+++ b/pulsar/consumer_regex_test.go
@@ -25,6 +25,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsaradmin"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+
 	"github.com/stretchr/testify/assert"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
@@ -443,3 +447,67 @@ func cloneConsumers(rc *regexConsumer) map[string]Consumer {
 	}
 	return consumers
 }
+
+func TestRegexTopicGetLastMessageIDs(t *testing.T) {
+
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	partition := 1
+	topic1 := fmt.Sprintf("regex-topic-%v", time.Now().Nanosecond())
+	topic2 := fmt.Sprintf("regex-topic-%v", time.Now().Nanosecond())
+	err = createPartitionedTopic(topic1, partition)
+	assert.Nil(t, err)
+	err = createPartitionedTopic(topic2, partition)
+	assert.Nil(t, err)
+	topics := []string{topic1, topic2}
+
+	// create consumer
+	topicsPattern := "persistent://public/default/regex-topic-.*"
+	consumer, err := client.Subscribe(ConsumerOptions{
+		TopicsPattern:    topicsPattern,
+		SubscriptionName: "my-sub",
+		Type:             Shared,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	// produce messages
+	totalMessage := 20
+	for i, topic := range topics {
+		p, err := client.CreateProducer(ProducerOptions{
+			Topic:           topic,
+			DisableBatching: true,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		err = genMessages(p, totalMessage, func(idx int) string {
+			return fmt.Sprintf("topic-%d-hello-%d", i+1, idx)
+		})
+		p.Close()
+		if err != nil {
+			assert.Nil(t, err)
+		}
+	}
+
+	// create admin
+	admin, err := pulsaradmin.NewClient(&config.Config{})
+	assert.Nil(t, err)
+
+	topicMessageIDs, err := consumer.GetLastMessageIDs()
+	assert.Nil(t, err)
+	assert.Equal(t, len(topics), len(topicMessageIDs))
+	for _, id := range topicMessageIDs {
+		assert.Equal(t, int(id.EntryID()), totalMessage/partition-1)
+		topicName, err := utils.GetTopicName(id.Topic())
+		assert.Nil(t, err)
+		messages, err := admin.Subscriptions().GetMessagesByID(*topicName, id.LedgerID(), id.EntryID())
+		assert.Nil(t, err)
+		assert.Equal(t, 1, len(messages))
+	}
+}
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index b3b797f..54c91ec 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -30,6 +30,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsaradmin"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
@@ -4478,3 +4482,119 @@ func TestConsumerForceUnSubscribe(t *testing.T) {
 	assert.Error(t, err)
 
 }
+
+func TestConsumerGetLastMessageIDs(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	partition := 1
+	topic := "my-topic"
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+		Type:             Shared,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	ctx := context.Background()
+	// send messages
+	totalMessage := 10
+	for i := 0; i < totalMessage; i++ {
+		if _, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		}); err != nil {
+			assert.Nil(t, err)
+		}
+	}
+
+	// create admin
+	admin, err := pulsaradmin.NewClient(&config.Config{})
+	assert.Nil(t, err)
+
+	messageIDs, err := consumer.GetLastMessageIDs()
+	assert.Nil(t, err)
+	assert.Equal(t, partition, len(messageIDs))
+
+	id := messageIDs[0]
+	topicName, err := utils.GetTopicName(id.Topic())
+	assert.Nil(t, err)
+	messages, err := admin.Subscriptions().GetMessagesByID(*topicName, id.LedgerID(), id.EntryID())
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(messages))
+
+}
+
+func TestPartitionConsumerGetLastMessageIDs(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	topic := newTopicName()
+	partition := 3
+	err = createPartitionedTopic(topic, partition)
+	assert.Nil(t, err)
+
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	// create consumer
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+		Type:             Shared,
+	})
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	ctx := context.Background()
+	totalMessage := 30
+	// send messages
+	for i := 0; i < totalMessage; i++ {
+		if _, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		}); err != nil {
+			assert.Nil(t, err)
+		}
+	}
+
+	// create admin
+	admin, err := pulsaradmin.NewClient(&config.Config{})
+	assert.Nil(t, err)
+
+	topicMessageIDs, err := consumer.GetLastMessageIDs()
+	assert.Nil(t, err)
+	assert.Equal(t, partition, len(topicMessageIDs))
+	for _, id := range topicMessageIDs {
+		assert.Equal(t, int(id.EntryID()), totalMessage/partition-1)
+
+		topicName, err := utils.GetTopicName(id.Topic())
+		assert.Nil(t, err)
+		messages, err := admin.Subscriptions().GetMessagesByID(*topicName, id.LedgerID(), id.EntryID())
+		assert.Nil(t, err)
+		assert.Equal(t, 1, len(messages))
+
+	}
+
+}
diff --git a/pulsar/internal/pulsartracing/consumer_interceptor_test.go b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
index 92ce231..05953cf 100644
--- a/pulsar/internal/pulsartracing/consumer_interceptor_test.go
+++ b/pulsar/internal/pulsartracing/consumer_interceptor_test.go
@@ -110,3 +110,8 @@ func (c *mockConsumer) SeekByTime(time time.Time) error {
 func (c *mockConsumer) Name() string {
 	return ""
 }
+
+func (c *mockConsumer) GetLastMessageIDs() ([]pulsar.TopicMessageID, error) {
+	ids := make([]pulsar.TopicMessageID, 0)
+	return ids, nil
+}