Repository: apache/pulsar-client-go
Pull request #1087 (2023-09-01 08:05:32)

[improve][client] Implement GetLastMSgID for Reader
### Motivation
Implement the GetLastMSgID API for Reader.

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test coverage.

*(or)*

This change is already covered by existing tests, such as *(please describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for end-to-end deployment with large payloads (10MB)*
  - *Extended integration test for recovery after broker failure*

### 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): (yes / no)
  - The public API: (yes / no)
  - The schema: (yes / no / don't know)
  - The default values of configurations: (yes / no)
  - The wire protocol: (yes / no)

### Documentation

  - Does this pull request introduce a new feature? (yes / no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - 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


liangyepianzhou: Leave a question: Do we have a plan to implement a multiple-topic-reader?
BewareMyPower: > Do we have a plan to implement a multiple-topic reader?

Not yet. But I think we can do that.

In addition, please add tests for the exposed API.

Patch:
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index c283f5a..801eab3 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -20,6 +20,7 @@ package pulsar
 import (
 	"fmt"
 	"net/url"
+	"sync"
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/auth"
@@ -47,6 +48,7 @@ type client struct {
 	metrics       *internal.Metrics
 	tcClient      *transactionCoordinatorClient
 	memLimit      internal.MemoryLimitController
+	closeOnce     sync.Once
 
 	log log.Logger
 }
@@ -266,7 +268,9 @@ func (c *client) TopicPartitions(topic string) ([]string, error) {
 }
 
 func (c *client) Close() {
-	c.handlers.Close()
-	c.cnxPool.Close()
-	c.lookupService.Close()
+	c.closeOnce.Do(func() {
+		c.handlers.Close()
+		c.cnxPool.Close()
+		c.lookupService.Close()
+	})
 }
diff --git a/pulsar/reader.go b/pulsar/reader.go
index d58d06f..5e1a73b 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -134,4 +134,7 @@ type Reader interface {
 	//            the message publish time where to reposition the subscription
 	//
 	SeekByTime(time time.Time) error
+
+	// GetLastMessageID get the last message id available for consume.
+	GetLastMessageID() (MessageID, error)
 }
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 5a2128a..ffc92de 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -244,3 +244,7 @@ func (r *reader) SeekByTime(time time.Time) error {
 
 	return r.pc.SeekByTime(time)
 }
+
+func (r *reader) GetLastMessageID() (MessageID, error) {
+	return r.pc.getLastMessageID()
+}
diff --git a/pulsaradmin/pkg/admin/auth/oauth2.go b/pulsaradmin/pkg/admin/auth/oauth2.go
index eab64ca..2864eae 100644
--- a/pulsaradmin/pkg/admin/auth/oauth2.go
+++ b/pulsaradmin/pkg/admin/auth/oauth2.go
@@ -19,9 +19,12 @@ package auth
 
 import (
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"path/filepath"
 
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
+
 	"github.com/99designs/keyring"
 	"github.com/apache/pulsar-client-go/oauth2"
 	"github.com/apache/pulsar-client-go/oauth2/cache"
@@ -243,8 +246,9 @@ func makeKeyring() (keyring.Keyring, error) {
 		ServiceName:              serviceName,
 		KeychainName:             keyChainName,
 		KeychainTrustApplication: true,
-		FileDir:                  filepath.Join("~/.config/pulsar", "credentials"),
-		FilePasswordFunc:         keyringPrompt,
+		FileDir: filepath.Join(fmt.Sprintf(
+			"%s/.config/pulsar", utils.GetConfigPath()), "credentials"),
+		FilePasswordFunc: keyringPrompt,
 	})
 }
 
diff --git a/pulsaradmin/pkg/utils/utils.go b/pulsaradmin/pkg/utils/utils.go
index 7b23b15..fc0897c 100644
--- a/pulsaradmin/pkg/utils/utils.go
+++ b/pulsaradmin/pkg/utils/utils.go
@@ -19,6 +19,7 @@ package utils
 
 import (
 	"fmt"
+	"os"
 	"reflect"
 )
 
@@ -36,3 +37,12 @@ func IsNilFixed(i interface{}) bool {
 	}
 	return false
 }
+
+func GetConfigPath() string {
+	// pulsar client config dir path, for example: PULSAR_CLIENT_CONF_PATH: /home/pulsar
+	// configuration file path is: /home/pulsar/.config/pulsar
+	if envConf, ok := os.LookupEnv("PULSAR_CLIENT_CONF_PATH"); ok {
+		return envConf
+	}
+	return HomeDir()
+}


Test patch:
diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index bb28371..78dc1ca 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -1238,3 +1238,10 @@ func TestAutoCloseIdleConnection(t *testing.T) {
 
 	cli.Close()
 }
+
+func TestMultipleCloseClient(t *testing.T) {
+	client, err := NewClient(ClientOptions{URL: serviceURL})
+	assert.Nil(t, err)
+	client.Close()
+	client.Close()
+}
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index 0a1b2a1..ec10f8f 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -901,3 +901,45 @@ func TestReaderWithBackoffPolicy(t *testing.T) {
 	partitionConsumerImp.reconnectToBroker()
 	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
 }
+
+func TestReaderGetLastMessageID(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.Nil(t, err)
+	topic := newTopicName()
+	ctx := context.Background()
+	schema := NewStringSchema(nil)
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+		Schema:          schema,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	var lastMsgID MessageID
+	// send 10 messages
+	for i := 0; i < 10; i++ {
+		msgID, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.NoError(t, err)
+		assert.NotNil(t, msgID)
+		lastMsgID = msgID
+	}
+
+	reader, err := client.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: EarliestMessageID(),
+	})
+	assert.Nil(t, err)
+	getLastMessageID, err := reader.GetLastMessageID()
+	if err != nil {
+		return
+	}
+
+	assert.Equal(t, lastMsgID.LedgerID(), getLastMessageID.LedgerID())
+	assert.Equal(t, lastMsgID.EntryID(), getLastMessageID.EntryID())
+}