Repository: apache/pulsar-client-go
Pull request #1091 (2023-09-11 04:53:46)

[fix] Fix comment for ConnectionMaxIdleTime
### Motivation

The default value of `ConnectionMaxIdleTime` is said to be 60 seconds in the comment,
https://github.com/apache/pulsar-client-go/blob/2a15a251d25bd2a6276051d71873e9e106cc1e85/pulsar/client.go#L155-L157

but it actually seems to be 180 seconds. 60 seconds is the minimum value, not the default value.
https://github.com/apache/pulsar-client-go/blob/2a15a251d25bd2a6276051d71873e9e106cc1e85/pulsar/client_impl.go#L39-L40

### Verifying this change

- [x] Make sure that the change passes the CI checks.


Patch:
diff --git a/pulsar/client.go b/pulsar/client.go
index d6f18c9..d9ac2a7 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -153,7 +153,7 @@ type ClientOptions struct {
 	MetricsRegisterer prometheus.Registerer
 
 	// Release the connection if it is not used for more than ConnectionMaxIdleTime.
-	// Default is 60 seconds, negative such as -1 to disable.
+	// Default is 180 seconds, minimum is 60 seconds. Negative such as -1 to disable.
 	ConnectionMaxIdleTime time.Duration
 
 	EnableTransaction bool
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()
+}


Test patch:
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())
+}