Repository: apache/pulsar-client-go
Pull request #1084 (2023-08-24 06:07:10)

double check before consumer reconnect
### Motivation

I have a topic with 4 partitions and multiple subscriptions. One of the subscriptions (in keyshared mode) is stuck while consuming from one partition, while the other subscriptions and this particular subscription consuming from other partitions are working fine.
After communicating with the cloud service provider, they believe that it is caused by a consumer not acknowledging the messages it received. However, in my code, I have closed the consumer, so the server should not be waiting indefinitely.
After checking the logs and the code, I believe the situation was as follows:
1. The server notifies the consumer to reconnect and starts sleeping at this line: https://github.com/apache/pulsar-client-go/blob/master/pulsar/consumer_partition.go#L1645.
2. In my code logic, I closed the consumer. At this point, since the connection is closed, the corresponding error message is printed in the log: [Failed to close consumer], err: connection closed
3. The sleep ends, the reconnect is successful, but this new connection is leaked, and the messages within it cannot be consumed.
4. In my code logic, I recreate the consumer, but due to the issue in the previous step, the new consumer is unable to receive any messages.

### Modifications

In the reconnectToBroker method, after the sleep, a double check is performed to confirm the status of the consumer.

### Verifying this change

This change is a trivial rework / code cleanup without any test coverage.

### Does this pull request potentially affect one of the following parts:

  - Dependencies (does it add or upgrade a dependency): (no)
  - The public API: (no)
  - The schema: (no)
  - The default values of configurations: (no)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (no)


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/client_impl.go b/pulsar/client_impl.go
index c283f5a..7daf6f6 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"
@@ -30,7 +31,6 @@ import (
 )
 
 const (
-	defaultConnectionTimeout           = 10 * time.Second
 	defaultOperationTimeout            = 30 * time.Second
 	defaultKeepAliveInterval           = 30 * time.Second
 	defaultMemoryLimitBytes            = 64 * 1024 * 1024
@@ -47,6 +47,7 @@ type client struct {
 	metrics       *internal.Metrics
 	tcClient      *transactionCoordinatorClient
 	memLimit      internal.MemoryLimitController
+	closeOnce     sync.Once
 
 	log log.Logger
 }
@@ -115,10 +116,10 @@ func newClient(options ClientOptions) (Client, error) {
 		return nil, err
 	}
 
+	// the default timeout respects Go's default timeout which is no timeout
+	// Missing user specified timeout renders 0 values that matches
+	// net.Dailer's default if time.Duration value is not initialized
 	connectionTimeout := options.ConnectionTimeout
-	if connectionTimeout.Nanoseconds() == 0 {
-		connectionTimeout = defaultConnectionTimeout
-	}
 
 	operationTimeout := options.OperationTimeout
 	if operationTimeout.Nanoseconds() == 0 {
@@ -266,7 +267,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/consumer_partition.go b/pulsar/consumer_partition.go
index 364dae7..2ba8427 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -1644,6 +1644,13 @@ func (pc *partitionConsumer) reconnectToBroker() {
 		pc.log.Info("Reconnecting to broker in ", delayReconnectTime)
 		time.Sleep(delayReconnectTime)
 
+		// double check
+		if pc.getConsumerState() != consumerReady {
+			// Consumer is already closing
+			pc.log.Info("consumer state not ready, exit reconnect")
+			return
+		}
+
 		err := pc.grabConn()
 		if err == nil {
 			// Successfully reconnected
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index e2ae7ac..840ecc4 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -256,7 +256,11 @@ func (c *connection) connect() bool {
 
 	if c.tlsOptions == nil {
 		// Clear text connection
-		cnx, err = net.DialTimeout("tcp", c.physicalAddr.Host, c.connectionTimeout)
+		if c.connectionTimeout.Nanoseconds() > 0 {
+			cnx, err = net.DialTimeout("tcp", c.physicalAddr.Host, c.connectionTimeout)
+		} else {
+			cnx, err = net.Dial("tcp", c.physicalAddr.Host)
+		}
 	} else {
 		// TLS connection
 		tlsConfig, err = c.getTLSConfig()
@@ -265,6 +269,8 @@ func (c *connection) connect() bool {
 			return false
 		}
 
+		// time.Duration is initialized to 0 by default, net.Dialer's default timeout is no timeout
+		// therefore if c.connectionTimeout is 0, it means no timeout
 		d := &net.Dialer{Timeout: c.connectionTimeout}
 		cnx, err = tls.DialWithDialer(d, "tcp", c.physicalAddr.Host, tlsConfig)
 	}
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index d3f61ef..ebd292e 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -551,13 +551,13 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 		deliverAt = time.Now().Add(msg.DeliverAfter)
 	}
 
-	mm := p.genMetadata(msg, uncompressedSize, deliverAt)
-
 	// set default ReplicationClusters when DisableReplication
 	if msg.DisableReplication {
 		msg.ReplicationClusters = []string{"__local__"}
 	}
 
+	mm := p.genMetadata(msg, uncompressedSize, deliverAt)
+
 	sendAsBatch := !p.options.DisableBatching &&
 		msg.ReplicationClusters == nil &&
 		deliverAt.UnixNano() < 0
@@ -1291,6 +1291,8 @@ func (p *partitionProducer) internalClose(req *closeProducer) {
 		return
 	}
 
+	defer close(p.dataChan)
+	defer close(p.cmdChan)
 	p.log.Info("Closing producer")
 
 	id := p.client.rpcClient.NewRequestID()
@@ -1304,6 +1306,7 @@ func (p *partitionProducer) internalClose(req *closeProducer) {
 	} else {
 		p.log.Info("Closed producer")
 	}
+	p.failPendingMessages()
 
 	if p.batchBuilder != nil {
 		if err = p.batchBuilder.Close(); err != nil {
@@ -1316,6 +1319,63 @@ func (p *partitionProducer) internalClose(req *closeProducer) {
 	p.batchFlushTicker.Stop()
 }
 
+func (p *partitionProducer) failPendingMessages() {
+	curViewItems := p.pendingQueue.ReadableSlice()
+	viewSize := len(curViewItems)
+	if viewSize <= 0 {
+		return
+	}
+	p.log.Infof("Failing %d messages on closing producer", viewSize)
+	lastViewItem := curViewItems[viewSize-1].(*pendingItem)
+
+	// iterate at most viewSize items
+	for i := 0; i < viewSize; i++ {
+		item := p.pendingQueue.CompareAndPoll(
+			func(m interface{}) bool {
+				return m != nil
+			})
+
+		if item == nil {
+			return
+		}
+
+		pi := item.(*pendingItem)
+		pi.Lock()
+
+		for _, i := range pi.sendRequests {
+			sr := i.(*sendRequest)
+			if sr.msg != nil {
+				size := len(sr.msg.Payload)
+				p.releaseSemaphoreAndMem(sr.reservedMem)
+				p.metrics.MessagesPending.Dec()
+				p.metrics.BytesPending.Sub(float64(size))
+				p.log.WithError(errProducerClosed).
+					WithField("size", size).
+					WithField("properties", sr.msg.Properties)
+			}
+
+			if sr.callback != nil {
+				sr.callbackOnce.Do(func() {
+					runCallback(sr.callback, nil, sr.msg, errProducerClosed)
+				})
+			}
+			if sr.transaction != nil {
+				sr.transaction.endSendOrAckOp(nil)
+			}
+		}
+
+		// flag the sending has completed with error, flush make no effect
+		pi.Complete(errProducerClosed)
+		pi.Unlock()
+
+		// finally reached the last view item, current iteration ends
+		if pi == lastViewItem {
+			p.log.Infof("%d messages complete failed", viewSize)
+			return
+		}
+	}
+}
+
 func (p *partitionProducer) LastSequenceID() int64 {
 	return atomic.LoadInt64(&p.lastSequenceID)
 }
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..4bd1546 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"
@@ -70,10 +73,15 @@ func NewAuthenticationOAuth2(issuer oauth2.Issuer, store store.Store) (*OAuth2Pr
 
 // NewAuthenticationOAuth2WithDefaultFlow uses memory to save the grant
 func NewAuthenticationOAuth2WithDefaultFlow(issuer oauth2.Issuer, keyFile string) (Provider, error) {
-	st := store.NewMemoryStore()
-	flow, err := oauth2.NewDefaultClientCredentialsFlow(oauth2.ClientCredentialsFlowOptions{
+	return NewAuthenticationOAuth2WithFlow(issuer, oauth2.ClientCredentialsFlowOptions{
 		KeyFile: keyFile,
 	})
+}
+
+func NewAuthenticationOAuth2WithFlow(
+	issuer oauth2.Issuer, flowOptions oauth2.ClientCredentialsFlowOptions) (Provider, error) {
+	st := store.NewMemoryStore()
+	flow, err := oauth2.NewDefaultClientCredentialsFlow(flowOptions)
 	if err != nil {
 		return nil, err
 	}
@@ -243,8 +251,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/producer_test.go b/pulsar/producer_test.go
index a9a7b81..29ffa78 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -30,7 +30,9 @@ import (
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
 	plog "github.com/apache/pulsar-client-go/pulsar/log"
@@ -2268,3 +2270,123 @@ func TestProducerSendWithContext(t *testing.T) {
 	//  producer.Send should fail and return err context.Canceled
 	assert.True(t, errors.Is(err, context.Canceled))
 }
+
+func TestFailPendingMessageWithClose(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+	testProducer, err := client.CreateProducer(ProducerOptions{
+		Topic:                   newTopicName(),
+		DisableBlockIfQueueFull: false,
+		BatchingMaxPublishDelay: 100000,
+		BatchingMaxMessages:     1000,
+	})
+
+	assert.NoError(t, err)
+	assert.NotNil(t, testProducer)
+	for i := 0; i < 3; i++ {
+		testProducer.SendAsync(context.Background(), &ProducerMessage{
+			Payload: make([]byte, 1024),
+		}, func(id MessageID, message *ProducerMessage, e error) {
+			if e != nil {
+				assert.Equal(t, errProducerClosed, e)
+			}
+		})
+	}
+	partitionProducerImp := testProducer.(*producer).producers[0].(*partitionProducer)
+	partitionProducerImp.pendingQueue.Put(&pendingItem{})
+	testProducer.Close()
+	assert.Equal(t, 0, partitionProducerImp.pendingQueue.Size())
+}
+
+type pendingQueueWrapper struct {
+	pendingQueue   internal.BlockingQueue
+	writtenBuffers *[]internal.Buffer
+}
+
+func (pqw *pendingQueueWrapper) Put(item interface{}) {
+	pi := item.(*pendingItem)
+	writerIdx := pi.buffer.WriterIndex()
+	buf := internal.NewBuffer(int(writerIdx))
+	buf.Write(pi.buffer.Get(0, writerIdx))
+	*pqw.writtenBuffers = append(*pqw.writtenBuffers, buf)
+	pqw.pendingQueue.Put(item)
+}
+
+func (pqw *pendingQueueWrapper) Take() interface{} {
+	return pqw.pendingQueue.Take()
+}
+
+func (pqw *pendingQueueWrapper) Poll() interface{} {
+	return pqw.pendingQueue.Poll()
+}
+
+func (pqw *pendingQueueWrapper) CompareAndPoll(compare func(interface{}) bool) interface{} {
+	return pqw.pendingQueue.CompareAndPoll(compare)
+}
+
+func (pqw *pendingQueueWrapper) Peek() interface{} {
+	return pqw.pendingQueue.Peek()
+}
+
+func (pqw *pendingQueueWrapper) PeekLast() interface{} {
+	return pqw.pendingQueue.PeekLast()
+}
+
+func (pqw *pendingQueueWrapper) Size() int {
+	return pqw.pendingQueue.Size()
+}
+
+func (pqw *pendingQueueWrapper) ReadableSlice() []interface{} {
+	return pqw.pendingQueue.ReadableSlice()
+}
+
+func TestDisableReplication(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	testProducer, err := client.CreateProducer(ProducerOptions{
+		Topic:           newTopicName(),
+		DisableBatching: true,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, testProducer)
+	defer testProducer.Close()
+
+	writtenBuffers := make([]internal.Buffer, 0)
+	pqw := &pendingQueueWrapper{
+		pendingQueue:   internal.NewBlockingQueue(1000),
+		writtenBuffers: &writtenBuffers,
+	}
+
+	partitionProducerImp := testProducer.(*producer).producers[0].(*partitionProducer)
+	partitionProducerImp.pendingQueue = pqw
+
+	ID, err := testProducer.Send(context.Background(), &ProducerMessage{
+		Payload:            []byte("disable-replication"),
+		DisableReplication: true,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, ID)
+
+	assert.Equal(t, 1, len(writtenBuffers))
+	buf := writtenBuffers[0]
+
+	buf.Skip(4)                        // TOTAL_SIZE
+	cmdSize := buf.ReadUint32()        // CMD_SIZE
+	buf.Skip(cmdSize)                  // CMD
+	buf.Skip(2)                        // MAGIC_NUMBER
+	buf.Skip(4)                        // CHECKSUM
+	metadataSize := buf.ReadUint32()   // METADATA_SIZE
+	metadata := buf.Read(metadataSize) // METADATA
+
+	var msgMetadata pb.MessageMetadata
+	err = proto.Unmarshal(metadata, &msgMetadata)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"__local__"}, msgMetadata.GetReplicateTo())
+}
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())
+}