Repository: apache/pulsar-client-go
Pull request #963 (2023-02-23 03:37:02)

[PIP-165] Auto release idle connections
<!--
### Contribution Checklist
  
  - Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
    Skip *Issue XYZ* if there is no associated github issue for this pull request.
    Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.

  - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
  
  - Each pull request should address only one issue, not mix up code from multiple issues.
  
  - Each commit in the pull request has a meaningful commit message

  - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.

**(The sections below can be removed for hotfixes of typos)**
-->

### Motivation

The go implementation of PIP-165:https://github.com/apache/pulsar/issues/15516


### Modifications

* Add new configuration `ConnectionMaxIdleTime` to `ClientOptions`
* Add a goroutine to `ConnectionPool` to period check and release idle connections.

### Verifying this change


This change added tests

### 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)
  - 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


shibd: @RobertIndie 

It seems that the newly added unit test has a timeout.

```shell
time="2023-03-01T10:41:21Z" level=info msg="Connecting to broker" remote_addr="pulsar://localhost:6650"
time="2023-03-01T10:41:21Z" level=info msg="TCP connection established" local_addr="127.0.0.1:42264" remote_addr="pulsar://localhost:6650"
time="2023-03-01T10:41:21Z" level=info msg="Connection is ready" local_addr="127.0.0.1:42264" remote_addr="pulsar://localhost:6650"
    client_impl_test.go:1224: 
        	Error Trace:	/pulsar/pulsar-client-go/pulsar/client_impl_test.go:1224
        	Error:      	Expected nil, but got: &errors.errorString{s:"request timed out"}
        	Test:       	TestAutoCloseIdleConnection
coverage: 29.7% of statements
panic: test timed out after 20m0s
```

Patch:
diff --git a/distribution/LICENSE.bin.txt b/distribution/LICENSE.bin.txt
index e37200f..79e2c52 100644
--- a/distribution/LICENSE.bin.txt
+++ b/distribution/LICENSE.bin.txt
@@ -218,9 +218,6 @@ MIT License
   * github.com/kr/pretty v0.2.0 -- distribution/license/LICENSE-pretty.txt
   * github.com/sirupsen/logrus v1.4.1 -- distribution/license/LICENSE-logrus.txt
 
-GNU GENERAL PUBLIC LICENSE Version 3
-  * github.com/beefsack/go-rate v0.0.0-20180408011153-efa7637bb9b6 -- distribution/license/LICENSE-go-rate.txt
-
 BSD License
   * github.com/DataDog/zstd v1.4.6-0.20210211175136-c6db21d202f4 -- distribution/license/LICENSE-zstd.txt
   * github.com/klauspost/compress v1.9.2 -- distribution/license/LICENSE-compress.txt
diff --git a/pulsar/client.go b/pulsar/client.go
index 75b363d..bc3f4f5 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -143,6 +143,10 @@ type ClientOptions struct {
 	// Default prometheus.DefaultRegisterer
 	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.
+	ConnectionMaxIdleTime time.Duration
+
 	EnableTransaction bool
 
 	// Limit of client memory usage (in byte). The 64M default can guarantee a high producer throughput.
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 7d90922..7c8fcc9 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -34,6 +34,8 @@ const (
 	defaultOperationTimeout  = 30 * time.Second
 	defaultKeepAliveInterval = 30 * time.Second
 	defaultMemoryLimitBytes  = 64 * 1024 * 1024
+	defaultConnMaxIdleTime   = 180 * time.Second
+	minConnMaxIdleTime       = 60 * time.Second
 )
 
 type client struct {
@@ -56,6 +58,16 @@ func newClient(options ClientOptions) (Client, error) {
 		logger = log.NewLoggerWithLogrus(logrus.StandardLogger())
 	}
 
+	connectionMaxIdleTime := options.ConnectionMaxIdleTime
+	if connectionMaxIdleTime == 0 {
+		connectionMaxIdleTime = defaultConnMaxIdleTime
+	} else if connectionMaxIdleTime > 0 && connectionMaxIdleTime < minConnMaxIdleTime {
+		return nil, newError(InvalidConfiguration, fmt.Sprintf("Connection max idle time should be at least %f "+
+			"seconds", minConnMaxIdleTime.Seconds()))
+	} else {
+		logger.Debugf("Disable auto release idle connections")
+	}
+
 	if options.URL == "" {
 		return nil, newError(InvalidConfiguration, "URL is required for client")
 	}
@@ -143,7 +155,7 @@ func newClient(options ClientOptions) (Client, error) {
 
 	c := &client{
 		cnxPool: internal.NewConnectionPool(tlsConfig, authProvider, connectionTimeout, keepAliveInterval,
-			maxConnectionsPerHost, logger, metrics),
+			maxConnectionsPerHost, logger, metrics, connectionMaxIdleTime),
 		log:      logger,
 		metrics:  metrics,
 		memLimit: internal.NewMemoryLimitController(memLimitBytes),
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index 8ee1822..7a86574 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -383,7 +383,7 @@ func (c *consumer) internalTopicSubscribeToPartitions() error {
 				metadata:                    metadata,
 				subProperties:               subProperties,
 				replicateSubscriptionState:  c.options.ReplicateSubscriptionState,
-				startMessageID:              trackingMessageID{},
+				startMessageID:              nil,
 				subscriptionMode:            durable,
 				readCompacted:               c.options.ReadCompacted,
 				interceptors:                c.options.Interceptors,
@@ -531,8 +531,8 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
 	if delay < 0 {
 		delay = 0
 	}
-	msgID, ok := c.messageID(msg.ID())
-	if !ok {
+	msgID := c.messageID(msg.ID())
+	if msgID == nil {
 		return
 	}
 	props := make(map[string]string)
@@ -581,13 +581,13 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
 
 func (c *consumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
-		mid, ok := c.messageID(msg.ID())
-		if !ok {
+		mid := c.messageID(msg.ID())
+		if mid == nil {
 			return
 		}
 
 		if mid.consumer != nil {
-			mid.consumer.NackID(msg.ID())
+			mid.NackByMsg(msg)
 			return
 		}
 		c.consumers[mid.partitionIdx].NackMsg(msg)
@@ -743,11 +743,11 @@ func toProtoInitialPosition(p SubscriptionInitialPosition) pb.CommandSubscribe_I
 	return pb.CommandSubscribe_Latest
 }
 
-func (c *consumer) messageID(msgID MessageID) (trackingMessageID, bool) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+func (c *consumer) messageID(msgID MessageID) *trackingMessageID {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
-		return trackingMessageID{}, false
+		return nil
 	}
 
 	partition := int(mid.partitionIdx)
@@ -755,10 +755,10 @@ func (c *consumer) messageID(msgID MessageID) (trackingMessageID, bool) {
 	if partition < 0 || partition >= len(c.consumers) {
 		c.log.Warnf("invalid partition index %d expected a partition between [0-%d]",
 			partition, len(c.consumers))
-		return trackingMessageID{}, false
+		return nil
 	}
 
-	return mid, true
+	return mid
 }
 
 func addMessageCryptoIfMissing(client *client, options *ConsumerOptions, topics interface{}) error {
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index 452915a..c0fcaef 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -125,8 +125,8 @@ func (c *multiTopicConsumer) Ack(msg Message) error {
 
 // AckID the consumption of a single message, identified by its MessageID
 func (c *multiTopicConsumer) AckID(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type in multi_consumer")
 	}
@@ -152,8 +152,8 @@ func (c *multiTopicConsumer) AckCumulative(msg Message) error {
 // AckIDCumulative the reception of all the messages in the stream up to (and including)
 // the provided message, identified by its MessageID
 func (c *multiTopicConsumer) AckIDCumulative(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type in multi_consumer")
 	}
@@ -203,8 +203,8 @@ func (c *multiTopicConsumer) ReconsumeLaterWithCustomProperties(msg Message, cus
 func (c *multiTopicConsumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
 		msgID := msg.ID()
-		mid, ok := toTrackingMessageID(msgID)
-		if !ok {
+		mid := toTrackingMessageID(msgID)
+		if mid == nil {
 			c.log.Warnf("invalid message id type %T", msgID)
 			return
 		}
@@ -221,8 +221,8 @@ func (c *multiTopicConsumer) Nack(msg Message) {
 }
 
 func (c *multiTopicConsumer) NackID(msgID MessageID) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return
 	}
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 95a8d32..0f7af3d 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -41,10 +41,6 @@ import (
 	uAtomic "go.uber.org/atomic"
 )
 
-var (
-	lastestMessageID = LatestMessageID()
-)
-
 type consumerState int
 
 const (
@@ -98,7 +94,7 @@ type partitionConsumerOpts struct {
 	metadata                    map[string]string
 	subProperties               map[string]string
 	replicateSubscriptionState  bool
-	startMessageID              trackingMessageID
+	startMessageID              *trackingMessageID
 	startMessageIDInclusive     bool
 	subscriptionMode            subscriptionMode
 	readCompacted               bool
@@ -149,13 +145,13 @@ type partitionConsumer struct {
 	queueSize       int32
 	queueCh         chan []*message
 	startMessageID  atomicMessageID
-	lastDequeuedMsg trackingMessageID
+	lastDequeuedMsg *trackingMessageID
 
 	eventsCh        chan interface{}
 	connectedCh     chan struct{}
 	connectClosedCh chan connectionClosed
 	closeCh         chan struct{}
-	clearQueueCh    chan func(id trackingMessageID)
+	clearQueueCh    chan func(id *trackingMessageID)
 
 	nackTracker *negativeAcksTracker
 	dlq         *dlqRouter
@@ -217,17 +213,17 @@ func (p *availablePermits) reset() {
 
 // atomicMessageID is a wrapper for trackingMessageID to make get and set atomic
 type atomicMessageID struct {
-	msgID trackingMessageID
+	msgID *trackingMessageID
 	sync.RWMutex
 }
 
-func (a *atomicMessageID) get() trackingMessageID {
+func (a *atomicMessageID) get() *trackingMessageID {
 	a.RLock()
 	defer a.RUnlock()
 	return a.msgID
 }
 
-func (a *atomicMessageID) set(msgID trackingMessageID) {
+func (a *atomicMessageID) set(msgID *trackingMessageID) {
 	a.Lock()
 	defer a.Unlock()
 	a.msgID = msgID
@@ -303,7 +299,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		messageCh:            messageCh,
 		connectClosedCh:      make(chan connectionClosed, 10),
 		closeCh:              make(chan struct{}),
-		clearQueueCh:         make(chan func(id trackingMessageID)),
+		clearQueueCh:         make(chan func(id *trackingMessageID)),
 		compressionProviders: sync.Map{},
 		dlq:                  dlq,
 		metrics:              metrics,
@@ -347,7 +343,8 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 	pc.log.Info("Created consumer")
 	pc.setConsumerState(consumerReady)
 
-	if pc.options.startMessageIDInclusive && pc.startMessageID.get().equal(lastestMessageID.(messageID)) {
+	startingMessageID := pc.startMessageID.get()
+	if pc.options.startMessageIDInclusive && startingMessageID != nil && startingMessageID.equal(latestMessageID) {
 		msgID, err := pc.requestGetLastMessageID()
 		if err != nil {
 			pc.nackTracker.Close()
@@ -418,10 +415,10 @@ func (pc *partitionConsumer) internalUnsubscribe(unsub *unsubscribeRequest) {
 	pc.setConsumerState(consumerClosed)
 }
 
-func (pc *partitionConsumer) getLastMessageID() (trackingMessageID, error) {
+func (pc *partitionConsumer) getLastMessageID() (*trackingMessageID, error) {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to redeliver closing or closed consumer")
-		return trackingMessageID{}, errors.New("failed to redeliver closing or closed consumer")
+		return nil, errors.New("failed to redeliver closing or closed consumer")
 	}
 	req := &getLastMsgIDRequest{doneCh: make(chan struct{})}
 	pc.eventsCh <- req
@@ -436,10 +433,10 @@ func (pc *partitionConsumer) internalGetLastMessageID(req *getLastMsgIDRequest)
 	req.msgID, req.err = pc.requestGetLastMessageID()
 }
 
-func (pc *partitionConsumer) requestGetLastMessageID() (trackingMessageID, error) {
+func (pc *partitionConsumer) requestGetLastMessageID() (*trackingMessageID, error) {
 	if state := pc.getConsumerState(); state == consumerClosed || state == consumerClosing {
 		pc.log.WithField("state", state).Error("Failed to getLastMessageID closing or closed consumer")
-		return trackingMessageID{}, errors.New("failed to getLastMessageID closing or closed consumer")
+		return nil, errors.New("failed to getLastMessageID closing or closed consumer")
 	}
 
 	requestID := pc.client.rpcClient.NewRequestID()
@@ -451,7 +448,7 @@ func (pc *partitionConsumer) requestGetLastMessageID() (trackingMessageID, error
 		pb.BaseCommand_GET_LAST_MESSAGE_ID, cmdGetLastMessageID)
 	if err != nil {
 		pc.log.WithError(err).Error("Failed to get last message id")
-		return trackingMessageID{}, err
+		return nil, err
 	}
 	id := res.Response.GetLastMessageIdResponse.GetLastMessageId()
 	return convertToMessageID(id), nil
@@ -463,16 +460,16 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 		return errors.New("consumer state is closed")
 	}
 
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		return pc.unAckChunksTracker.ack(cmid)
 	}
 
-	trackingID, ok := toTrackingMessageID(msgID)
-	if !ok {
+	trackingID := toTrackingMessageID(msgID)
+	if trackingID == nil {
 		return errors.New("failed to convert trackingMessageID")
 	}
 
-	if !trackingID.Undefined() && trackingID.ack() {
+	if trackingID != nil && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
 		pc.metrics.ProcessingTime.Observe(float64(time.Now().UnixNano()-trackingID.receivedTime.UnixNano()) / 1.0e9)
 	} else if !pc.options.enableBatchIndexAck {
@@ -481,10 +478,10 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 
 	var ackReq *ackRequest
 	if withResponse {
-		ackReq := pc.sendIndividualAck(&trackingID)
+		ackReq := pc.sendIndividualAck(trackingID)
 		<-ackReq.doneCh
 	} else {
-		pc.ackGroupingTracker.add(&trackingID)
+		pc.ackGroupingTracker.add(trackingID)
 	}
 	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
 	if ackReq == nil {
@@ -526,15 +523,12 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 	}
 
 	// chunk message id will be converted to tracking message id
-	trackingID, ok := toTrackingMessageID(msgID)
-	if !ok {
+	trackingID := toTrackingMessageID(msgID)
+	if trackingID == nil {
 		return errors.New("failed to convert trackingMessageID")
 	}
-	if trackingID.Undefined() {
-		return nil
-	}
 
-	var msgIDToAck trackingMessageID
+	var msgIDToAck *trackingMessageID
 	if trackingID.ackCumulative() || pc.options.enableBatchIndexAck {
 		msgIDToAck = trackingID
 	} else if !trackingID.tracker.hasPrevBatchAcked() {
@@ -551,15 +545,15 @@ func (pc *partitionConsumer) internalAckIDCumulative(msgID MessageID, withRespon
 
 	var ackReq *ackRequest
 	if withResponse {
-		ackReq := pc.sendCumulativeAck(&msgIDToAck)
+		ackReq := pc.sendCumulativeAck(msgIDToAck)
 		<-ackReq.doneCh
 	} else {
-		pc.ackGroupingTracker.addCumulative(&msgIDToAck)
+		pc.ackGroupingTracker.addCumulative(msgIDToAck)
 	}
 
 	pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
 
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		pc.unAckChunksTracker.remove(cmid)
 	}
 
@@ -580,13 +574,13 @@ func (pc *partitionConsumer) sendCumulativeAck(msgID MessageID) *ackRequest {
 }
 
 func (pc *partitionConsumer) NackID(msgID MessageID) {
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		pc.unAckChunksTracker.nack(cmid)
 		return
 	}
 
-	trackingID, ok := toTrackingMessageID(msgID)
-	if !ok {
+	trackingID := toTrackingMessageID(msgID)
+	if trackingID == nil {
 		return
 	}
 
@@ -674,9 +668,9 @@ func (pc *partitionConsumer) Seek(msgID MessageID) error {
 	req := &seekRequest{
 		doneCh: make(chan struct{}),
 	}
-	if cmid, ok := toChunkedMessageID(msgID); ok {
+	if cmid, ok := msgID.(*chunkMessageID); ok {
 		req.msgID = cmid.firstChunkID
-	} else if tmid, ok := toTrackingMessageID(msgID); ok {
+	} else if tmid := toTrackingMessageID(msgID); tmid != nil {
 		req.msgID = tmid.messageID
 	} else {
 		// will never reach
@@ -695,7 +689,7 @@ func (pc *partitionConsumer) internalSeek(seek *seekRequest) {
 	defer close(seek.doneCh)
 	seek.err = pc.requestSeek(seek.msgID)
 }
-func (pc *partitionConsumer) requestSeek(msgID messageID) error {
+func (pc *partitionConsumer) requestSeek(msgID *messageID) error {
 	if err := pc.requestSeekWithoutClear(msgID); err != nil {
 		return err
 	}
@@ -703,7 +697,7 @@ func (pc *partitionConsumer) requestSeek(msgID messageID) error {
 	return nil
 }
 
-func (pc *partitionConsumer) requestSeekWithoutClear(msgID messageID) error {
+func (pc *partitionConsumer) requestSeekWithoutClear(msgID *messageID) error {
 	state := pc.getConsumerState()
 	if state == consumerClosing || state == consumerClosed {
 		pc.log.WithField("state", state).Error("failed seek by consumer is closing or has closed")
@@ -1063,7 +1057,7 @@ func (pc *partitionConsumer) processMessageChunk(compressedPayload internal.Buff
 	numChunks := msgMeta.GetNumChunksFromMsg()
 	totalChunksSize := int(msgMeta.GetTotalChunkMsgSize())
 	chunkID := msgMeta.GetChunkId()
-	msgID := messageID{
+	msgID := &messageID{
 		ledgerID:     int64(pbMsgID.GetLedgerId()),
 		entryID:      int64(pbMsgID.GetEntryId()),
 		batchIdx:     -1,
@@ -1105,12 +1099,12 @@ func (pc *partitionConsumer) processMessageChunk(compressedPayload internal.Buff
 	return ctx.chunkedMsgBuffer
 }
 
-func (pc *partitionConsumer) messageShouldBeDiscarded(msgID trackingMessageID) bool {
-	if pc.startMessageID.get().Undefined() {
+func (pc *partitionConsumer) messageShouldBeDiscarded(msgID *trackingMessageID) bool {
+	if pc.startMessageID.get() == nil {
 		return false
 	}
 	// if we start at latest message, we should never discard
-	if pc.options.startMessageID.equal(latestMessageID) {
+	if pc.options.startMessageID != nil && pc.options.startMessageID.equal(latestMessageID) {
 		return false
 	}
 
@@ -1263,7 +1257,7 @@ func (pc *partitionConsumer) dispatcher() {
 		case clearQueueCb := <-pc.clearQueueCh:
 			// drain the message queue on any new connection by sending a
 			// special nil message to the channel so we know when to stop dropping messages
-			var nextMessageInQueue trackingMessageID
+			var nextMessageInQueue *trackingMessageID
 			go func() {
 				pc.queueCh <- nil
 			}()
@@ -1272,8 +1266,8 @@ func (pc *partitionConsumer) dispatcher() {
 				// the queue has been drained
 				if m == nil {
 					break
-				} else if nextMessageInQueue.Undefined() {
-					nextMessageInQueue, _ = toTrackingMessageID(m[0].msgID)
+				} else if nextMessageInQueue == nil {
+					nextMessageInQueue = toTrackingMessageID(m[0].msgID)
 				}
 			}
 
@@ -1311,13 +1305,13 @@ type redeliveryRequest struct {
 
 type getLastMsgIDRequest struct {
 	doneCh chan struct{}
-	msgID  trackingMessageID
+	msgID  *trackingMessageID
 	err    error
 }
 
 type seekRequest struct {
 	doneCh chan struct{}
-	msgID  messageID
+	msgID  *messageID
 	err    error
 }
 
@@ -1578,15 +1572,15 @@ func (pc *partitionConsumer) grabConn() error {
 	}
 }
 
-func (pc *partitionConsumer) clearQueueAndGetNextMessage() trackingMessageID {
+func (pc *partitionConsumer) clearQueueAndGetNextMessage() *trackingMessageID {
 	if pc.getConsumerState() != consumerReady {
-		return trackingMessageID{}
+		return nil
 	}
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
-	var msgID trackingMessageID
+	var msgID *trackingMessageID
 
-	pc.clearQueueCh <- func(id trackingMessageID) {
+	pc.clearQueueCh <- func(id *trackingMessageID) {
 		msgID = id
 		wg.Done()
 	}
@@ -1599,16 +1593,16 @@ func (pc *partitionConsumer) clearQueueAndGetNextMessage() trackingMessageID {
  * Clear the internal receiver queue and returns the message id of what was the 1st message in the queue that was
  * not seen by the application
  */
-func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
+func (pc *partitionConsumer) clearReceiverQueue() *trackingMessageID {
 	nextMessageInQueue := pc.clearQueueAndGetNextMessage()
 
-	if pc.startMessageID.get().Undefined() {
+	if pc.startMessageID.get() == nil {
 		return pc.startMessageID.get()
 	}
 
-	if !nextMessageInQueue.Undefined() {
+	if nextMessageInQueue != nil {
 		return getPreviousMessage(nextMessageInQueue)
-	} else if !pc.lastDequeuedMsg.Undefined() {
+	} else if pc.lastDequeuedMsg != nil {
 		// If the queue was empty we need to restart from the message just after the last one that has been dequeued
 		// in the past
 		return pc.lastDequeuedMsg
@@ -1618,10 +1612,10 @@ func (pc *partitionConsumer) clearReceiverQueue() trackingMessageID {
 	}
 }
 
-func getPreviousMessage(mid trackingMessageID) trackingMessageID {
+func getPreviousMessage(mid *trackingMessageID) *trackingMessageID {
 	if mid.batchIdx >= 0 {
-		return trackingMessageID{
-			messageID: messageID{
+		return &trackingMessageID{
+			messageID: &messageID{
 				ledgerID:     mid.ledgerID,
 				entryID:      mid.entryID,
 				batchIdx:     mid.batchIdx - 1,
@@ -1634,8 +1628,8 @@ func getPreviousMessage(mid trackingMessageID) trackingMessageID {
 	}
 
 	// Get on previous message in previous entry
-	return trackingMessageID{
-		messageID: messageID{
+	return &trackingMessageID{
+		messageID: &messageID{
 			ledgerID:     mid.ledgerID,
 			entryID:      mid.entryID - 1,
 			batchIdx:     mid.batchIdx,
@@ -1734,8 +1728,8 @@ func (pc *partitionConsumer) _getConn() internal.Connection {
 	return pc.conn.Load().(internal.Connection)
 }
 
-func convertToMessageIDData(msgID trackingMessageID) *pb.MessageIdData {
-	if msgID.Undefined() {
+func convertToMessageIDData(msgID *trackingMessageID) *pb.MessageIdData {
+	if msgID == nil {
 		return nil
 	}
 
@@ -1745,13 +1739,13 @@ func convertToMessageIDData(msgID trackingMessageID) *pb.MessageIdData {
 	}
 }
 
-func convertToMessageID(id *pb.MessageIdData) trackingMessageID {
+func convertToMessageID(id *pb.MessageIdData) *trackingMessageID {
 	if id == nil {
-		return trackingMessageID{}
+		return nil
 	}
 
-	msgID := trackingMessageID{
-		messageID: messageID{
+	msgID := &trackingMessageID{
+		messageID: &messageID{
 			ledgerID: int64(*id.LedgerId),
 			entryID:  int64(*id.EntryId),
 		},
@@ -1767,7 +1761,7 @@ type chunkedMsgCtx struct {
 	totalChunks      int32
 	chunkedMsgBuffer internal.Buffer
 	lastChunkedMsgID int32
-	chunkedMsgIDs    []messageID
+	chunkedMsgIDs    []*messageID
 	receivedTime     int64
 
 	mu sync.Mutex
@@ -1778,12 +1772,12 @@ func newChunkedMsgCtx(numChunksFromMsg int32, totalChunkMsgSize int) *chunkedMsg
 		totalChunks:      numChunksFromMsg,
 		chunkedMsgBuffer: internal.NewBuffer(totalChunkMsgSize),
 		lastChunkedMsgID: -1,
-		chunkedMsgIDs:    make([]messageID, numChunksFromMsg),
+		chunkedMsgIDs:    make([]*messageID, numChunksFromMsg),
 		receivedTime:     time.Now().Unix(),
 	}
 }
 
-func (c *chunkedMsgCtx) append(chunkID int32, msgID messageID, partPayload internal.Buffer) {
+func (c *chunkedMsgCtx) append(chunkID int32, msgID *messageID, partPayload internal.Buffer) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.chunkedMsgIDs[chunkID] = msgID
@@ -1791,20 +1785,20 @@ func (c *chunkedMsgCtx) append(chunkID int32, msgID messageID, partPayload inter
 	c.lastChunkedMsgID = chunkID
 }
 
-func (c *chunkedMsgCtx) firstChunkID() messageID {
+func (c *chunkedMsgCtx) firstChunkID() *messageID {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.chunkedMsgIDs) == 0 {
-		return messageID{}
+		return nil
 	}
 	return c.chunkedMsgIDs[0]
 }
 
-func (c *chunkedMsgCtx) lastChunkID() messageID {
+func (c *chunkedMsgCtx) lastChunkID() *messageID {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if len(c.chunkedMsgIDs) == 0 {
-		return messageID{}
+		return nil
 	}
 	return c.chunkedMsgIDs[len(c.chunkedMsgIDs)-1]
 }
@@ -1814,9 +1808,13 @@ func (c *chunkedMsgCtx) discard(pc *partitionConsumer) {
 	defer c.mu.Unlock()
 
 	for _, mid := range c.chunkedMsgIDs {
+		if mid == nil {
+			continue
+		}
 		pc.log.Info("Removing chunk message-id", mid.String())
-		tmid, _ := toTrackingMessageID(mid)
-		pc.AckID(tmid)
+		if tmid := toTrackingMessageID(mid); tmid != nil {
+			pc.AckID(tmid)
+		}
 	}
 }
 
@@ -1935,40 +1933,41 @@ func (c *chunkedMsgCtxMap) Close() {
 }
 
 type unAckChunksTracker struct {
-	chunkIDs map[chunkMessageID][]messageID
+	// TODO: use hash code of chunkMessageID as the key
+	chunkIDs map[chunkMessageID][]*messageID
 	pc       *partitionConsumer
 	mu       sync.Mutex
 }
 
 func newUnAckChunksTracker(pc *partitionConsumer) *unAckChunksTracker {
 	return &unAckChunksTracker{
-		chunkIDs: make(map[chunkMessageID][]messageID),
+		chunkIDs: make(map[chunkMessageID][]*messageID),
 		pc:       pc,
 	}
 }
 
-func (u *unAckChunksTracker) add(cmid chunkMessageID, ids []messageID) {
+func (u *unAckChunksTracker) add(cmid *chunkMessageID, ids []*messageID) {
 	u.mu.Lock()
 	defer u.mu.Unlock()
 
-	u.chunkIDs[cmid] = ids
+	u.chunkIDs[*cmid] = ids
 }
 
-func (u *unAckChunksTracker) get(cmid chunkMessageID) []messageID {
+func (u *unAckChunksTracker) get(cmid *chunkMessageID) []*messageID {
 	u.mu.Lock()
 	defer u.mu.Unlock()
 
-	return u.chunkIDs[cmid]
+	return u.chunkIDs[*cmid]
 }
 
-func (u *unAckChunksTracker) remove(cmid chunkMessageID) {
+func (u *unAckChunksTracker) remove(cmid *chunkMessageID) {
 	u.mu.Lock()
 	defer u.mu.Unlock()
 
-	delete(u.chunkIDs, cmid)
+	delete(u.chunkIDs, *cmid)
 }
 
-func (u *unAckChunksTracker) ack(cmid chunkMessageID) error {
+func (u *unAckChunksTracker) ack(cmid *chunkMessageID) error {
 	ids := u.get(cmid)
 	for _, id := range ids {
 		if err := u.pc.AckID(id); err != nil {
@@ -1979,7 +1978,7 @@ func (u *unAckChunksTracker) ack(cmid chunkMessageID) error {
 	return nil
 }
 
-func (u *unAckChunksTracker) nack(cmid chunkMessageID) {
+func (u *unAckChunksTracker) nack(cmid *chunkMessageID) {
 	ids := u.get(cmid)
 	for _, id := range ids {
 		u.pc.NackID(id)
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index d890c67..fdfecec 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -174,8 +174,8 @@ func (c *regexConsumer) ReconsumeLaterWithCustomProperties(msg Message, customPr
 
 // AckID the consumption of a single message, identified by its MessageID
 func (c *regexConsumer) AckID(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type")
 	}
@@ -201,8 +201,8 @@ func (c *regexConsumer) AckCumulative(msg Message) error {
 // AckIDCumulative the reception of all the messages in the stream up to (and including)
 // the provided message, identified by its MessageID
 func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type")
 	}
@@ -222,8 +222,8 @@ func (c *regexConsumer) AckIDCumulative(msgID MessageID) error {
 func (c *regexConsumer) Nack(msg Message) {
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
 		msgID := msg.ID()
-		mid, ok := toTrackingMessageID(msgID)
-		if !ok {
+		mid := toTrackingMessageID(msgID)
+		if mid == nil {
 			c.log.Warnf("invalid message id type %T", msgID)
 			return
 		}
@@ -240,8 +240,8 @@ func (c *regexConsumer) Nack(msg Message) {
 }
 
 func (c *regexConsumer) NackID(msgID MessageID) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return
 	}
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 39db8e1..68ddecd 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -39,7 +39,7 @@ type messageID struct {
 	batchSize    int32
 }
 
-var latestMessageID = messageID{
+var latestMessageID = &messageID{
 	ledgerID:     math.MaxInt64,
 	entryID:      math.MaxInt64,
 	batchIdx:     -1,
@@ -47,7 +47,7 @@ var latestMessageID = messageID{
 	batchSize:    0,
 }
 
-var earliestMessageID = messageID{
+var earliestMessageID = &messageID{
 	ledgerID:     -1,
 	entryID:      -1,
 	batchIdx:     -1,
@@ -56,18 +56,14 @@ var earliestMessageID = messageID{
 }
 
 type trackingMessageID struct {
-	messageID
+	*messageID
 
 	tracker      *ackTracker
 	consumer     acker
 	receivedTime time.Time
 }
 
-func (id trackingMessageID) Undefined() bool {
-	return id == trackingMessageID{}
-}
-
-func (id trackingMessageID) Ack() error {
+func (id *trackingMessageID) Ack() error {
 	if id.consumer == nil {
 		return errors.New("consumer is nil in trackingMessageID")
 	}
@@ -78,7 +74,7 @@ func (id trackingMessageID) Ack() error {
 	return nil
 }
 
-func (id trackingMessageID) AckWithResponse() error {
+func (id *trackingMessageID) AckWithResponse() error {
 	if id.consumer == nil {
 		return errors.New("consumer is nil in trackingMessageID")
 	}
@@ -89,37 +85,37 @@ func (id trackingMessageID) AckWithResponse() error {
 	return nil
 }
 
-func (id trackingMessageID) Nack() {
+func (id *trackingMessageID) Nack() {
 	if id.consumer == nil {
 		return
 	}
 	id.consumer.NackID(id)
 }
 
-func (id trackingMessageID) NackByMsg(msg Message) {
+func (id *trackingMessageID) NackByMsg(msg Message) {
 	if id.consumer == nil {
 		return
 	}
 	id.consumer.NackMsg(msg)
 }
 
-func (id trackingMessageID) ack() bool {
+func (id *trackingMessageID) ack() bool {
 	if id.tracker != nil && id.batchIdx > -1 {
 		return id.tracker.ack(int(id.batchIdx))
 	}
 	return true
 }
 
-func (id trackingMessageID) ackCumulative() bool {
+func (id *trackingMessageID) ackCumulative() bool {
 	if id.tracker != nil && id.batchIdx > -1 {
 		return id.tracker.ackCumulative(int(id.batchIdx))
 	}
 	return true
 }
 
-func (id trackingMessageID) prev() trackingMessageID {
-	return trackingMessageID{
-		messageID: messageID{
+func (id *trackingMessageID) prev() *trackingMessageID {
+	return &trackingMessageID{
+		messageID: &messageID{
 			ledgerID:     id.ledgerID,
 			entryID:      id.entryID - 1,
 			partitionIdx: id.partitionIdx,
@@ -129,11 +125,11 @@ func (id trackingMessageID) prev() trackingMessageID {
 	}
 }
 
-func (id messageID) isEntryIDValid() bool {
+func (id *messageID) isEntryIDValid() bool {
 	return id.entryID >= 0
 }
 
-func (id messageID) greater(other messageID) bool {
+func (id *messageID) greater(other *messageID) bool {
 	if id.ledgerID != other.ledgerID {
 		return id.ledgerID > other.ledgerID
 	}
@@ -145,17 +141,17 @@ func (id messageID) greater(other messageID) bool {
 	return id.batchIdx > other.batchIdx
 }
 
-func (id messageID) equal(other messageID) bool {
+func (id *messageID) equal(other *messageID) bool {
 	return id.ledgerID == other.ledgerID &&
 		id.entryID == other.entryID &&
 		id.batchIdx == other.batchIdx
 }
 
-func (id messageID) greaterEqual(other messageID) bool {
+func (id *messageID) greaterEqual(other *messageID) bool {
 	return id.equal(other) || id.greater(other)
 }
 
-func (id messageID) Serialize() []byte {
+func (id *messageID) Serialize() []byte {
 	msgID := &pb.MessageIdData{
 		LedgerId:   proto.Uint64(uint64(id.ledgerID)),
 		EntryId:    proto.Uint64(uint64(id.entryID)),
@@ -167,27 +163,27 @@ func (id messageID) Serialize() []byte {
 	return data
 }
 
-func (id messageID) LedgerID() int64 {
+func (id *messageID) LedgerID() int64 {
 	return id.ledgerID
 }
 
-func (id messageID) EntryID() int64 {
+func (id *messageID) EntryID() int64 {
 	return id.entryID
 }
 
-func (id messageID) BatchIdx() int32 {
+func (id *messageID) BatchIdx() int32 {
 	return id.batchIdx
 }
 
-func (id messageID) PartitionIdx() int32 {
+func (id *messageID) PartitionIdx() int32 {
 	return id.partitionIdx
 }
 
-func (id messageID) BatchSize() int32 {
+func (id *messageID) BatchSize() int32 {
 	return id.batchSize
 }
 
-func (id messageID) String() string {
+func (id *messageID) String() string {
 	return fmt.Sprintf("%d:%d:%d", id.ledgerID, id.entryID, id.partitionIdx)
 }
 
@@ -208,7 +204,7 @@ func deserializeMessageID(data []byte) (MessageID, error) {
 }
 
 func newMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32, batchSize int32) MessageID {
-	return messageID{
+	return &messageID{
 		ledgerID:     ledgerID,
 		entryID:      entryID,
 		batchIdx:     batchIdx,
@@ -218,9 +214,9 @@ func newMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx in
 }
 
 func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32, batchSize int32,
-	tracker *ackTracker) trackingMessageID {
-	return trackingMessageID{
-		messageID: messageID{
+	tracker *ackTracker) *trackingMessageID {
+	return &trackingMessageID{
+		messageID: &messageID{
 			ledgerID:     ledgerID,
 			entryID:      entryID,
 			batchIdx:     batchIdx,
@@ -232,31 +228,23 @@ func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partiti
 	}
 }
 
-func toTrackingMessageID(msgID MessageID) (trackingMessageID, bool) {
-	if mid, ok := msgID.(messageID); ok {
-		return trackingMessageID{
+func toTrackingMessageID(msgID MessageID) *trackingMessageID {
+	if mid, ok := msgID.(*messageID); ok {
+		return &trackingMessageID{
 			messageID:    mid,
 			receivedTime: time.Now(),
-		}, true
-	} else if mid, ok := msgID.(trackingMessageID); ok {
-		return mid, true
-	} else if cmid, ok := msgID.(chunkMessageID); ok {
-		return trackingMessageID{
+		}
+	} else if mid, ok := msgID.(*trackingMessageID); ok {
+		return mid
+	} else if cmid, ok := msgID.(*chunkMessageID); ok {
+		return &trackingMessageID{
 			messageID:    cmid.messageID,
 			receivedTime: cmid.receivedTime,
 			consumer:     cmid.consumer,
-		}, true
+		}
 	} else {
-		return trackingMessageID{}, false
-	}
-}
-
-func toChunkedMessageID(msgID MessageID) (chunkMessageID, bool) {
-	cid, ok := msgID.(chunkMessageID)
-	if ok {
-		return cid, true
+		return nil
 	}
-	return chunkMessageID{}, false
 }
 
 func timeFromUnixTimestampMillis(timestamp uint64) time.Time {
@@ -449,16 +437,16 @@ func (t *ackTracker) toAckSet() []int64 {
 }
 
 type chunkMessageID struct {
-	messageID
+	*messageID
 
-	firstChunkID messageID
+	firstChunkID *messageID
 	receivedTime time.Time
 
 	consumer acker
 }
 
-func newChunkMessageID(firstChunkID messageID, lastChunkID messageID) chunkMessageID {
-	return chunkMessageID{
+func newChunkMessageID(firstChunkID *messageID, lastChunkID *messageID) *chunkMessageID {
+	return &chunkMessageID{
 		messageID:    lastChunkID,
 		firstChunkID: firstChunkID,
 		receivedTime: time.Now(),
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 09623e5..55d04f3 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -170,6 +170,8 @@ type connection struct {
 	metrics        *Metrics
 
 	keepAliveInterval time.Duration
+
+	lastActive time.Time
 }
 
 // connectionOptions defines configurations for creating connection.
@@ -927,6 +929,52 @@ func (c *connection) UnregisterListener(id uint64) {
 	delete(c.listeners, id)
 }
 
+func (c *connection) ResetLastActive() {
+	c.Lock()
+	defer c.Unlock()
+	c.lastActive = time.Now()
+}
+
+func (c *connection) isIdle() bool {
+	{
+		c.pendingLock.Lock()
+		defer c.pendingLock.Unlock()
+		if len(c.pendingReqs) != 0 {
+			return false
+		}
+	}
+
+	{
+		c.listenersLock.RLock()
+		defer c.listenersLock.RUnlock()
+		if len(c.listeners) != 0 {
+			return false
+		}
+	}
+
+	{
+		c.consumerHandlersLock.Lock()
+		defer c.consumerHandlersLock.Unlock()
+		if len(c.consumerHandlers) != 0 {
+			return false
+		}
+	}
+
+	if len(c.incomingRequestsCh) != 0 || len(c.writeRequestsCh) != 0 {
+		return false
+	}
+	return true
+}
+
+func (c *connection) CheckIdle(maxIdleTime time.Duration) bool {
+	// We don't need to lock here because this method should only be
+	// called in a single goroutine of the connectionPool
+	if !c.isIdle() {
+		c.lastActive = time.Now()
+	}
+	return time.Since(c.lastActive) > maxIdleTime
+}
+
 // Close closes the connection by
 // closing underlying socket connection and closeCh.
 // This also triggers callbacks to the ConnectionClosed listeners.
diff --git a/pulsar/internal/connection_pool.go b/pulsar/internal/connection_pool.go
index 4f6d656..bf1e297 100644
--- a/pulsar/internal/connection_pool.go
+++ b/pulsar/internal/connection_pool.go
@@ -47,6 +47,7 @@ type connectionPool struct {
 	maxConnectionsPerHost int32
 	roundRobinCnt         int32
 	keepAliveInterval     time.Duration
+	closeCh               chan struct{}
 
 	metrics *Metrics
 	log     log.Logger
@@ -60,8 +61,9 @@ func NewConnectionPool(
 	keepAliveInterval time.Duration,
 	maxConnectionsPerHost int,
 	logger log.Logger,
-	metrics *Metrics) ConnectionPool {
-	return &connectionPool{
+	metrics *Metrics,
+	connectionMaxIdleTime time.Duration) ConnectionPool {
+	p := &connectionPool{
 		connections:           make(map[string]*connection),
 		tlsOptions:            tlsOptions,
 		auth:                  auth,
@@ -70,7 +72,10 @@ func NewConnectionPool(
 		keepAliveInterval:     keepAliveInterval,
 		log:                   logger,
 		metrics:               metrics,
+		closeCh:               make(chan struct{}),
 	}
+	go p.checkAndCleanIdleConnections(connectionMaxIdleTime)
+	return p
 }
 
 func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.URL) (Connection, error) {
@@ -109,6 +114,7 @@ func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.U
 		p.Unlock()
 		conn.start()
 	} else {
+		conn.ResetLastActive()
 		// we already have a connection
 		p.Unlock()
 	}
@@ -119,6 +125,7 @@ func (p *connectionPool) GetConnection(logicalAddr *url.URL, physicalAddr *url.U
 
 func (p *connectionPool) Close() {
 	p.Lock()
+	close(p.closeCh)
 	for k, c := range p.connections {
 		delete(p.connections, k)
 		c.Close()
@@ -134,3 +141,25 @@ func (p *connectionPool) getMapKey(addr *url.URL) string {
 	idx := cnt % p.maxConnectionsPerHost
 	return fmt.Sprint(addr.Host, '-', idx)
 }
+
+func (p *connectionPool) checkAndCleanIdleConnections(maxIdleTime time.Duration) {
+	if maxIdleTime < 0 {
+		return
+	}
+	for {
+		select {
+		case <-p.closeCh:
+			return
+		case <-time.After(maxIdleTime):
+			p.Lock()
+			for k, c := range p.connections {
+				if c.CheckIdle(maxIdleTime) {
+					c.log.Debugf("Closed connection due to inactivity.")
+					delete(p.connections, k)
+					c.Close()
+				}
+			}
+			p.Unlock()
+		}
+	}
+}
diff --git a/pulsar/negative_acks_tracker.go b/pulsar/negative_acks_tracker.go
index 79ed694..58f5676 100644
--- a/pulsar/negative_acks_tracker.go
+++ b/pulsar/negative_acks_tracker.go
@@ -65,7 +65,7 @@ func newNegativeAcksTracker(rc redeliveryConsumer, delay time.Duration,
 	return t
 }
 
-func (t *negativeAcksTracker) Add(msgID messageID) {
+func (t *negativeAcksTracker) Add(msgID *messageID) {
 	// Always clear up the batch index since we want to track the nack
 	// for the entire batch
 	batchMsgID := messageID{
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index c3a0aa9..744df79 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -1163,7 +1163,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 				if sr.totalChunks > 1 {
 					if sr.chunkID == 0 {
 						sr.chunkRecorder.setFirstChunkID(
-							messageID{
+							&messageID{
 								int64(response.MessageId.GetLedgerId()),
 								int64(response.MessageId.GetEntryId()),
 								-1,
@@ -1172,7 +1172,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 							})
 					} else if sr.chunkID == sr.totalChunks-1 {
 						sr.chunkRecorder.setLastChunkID(
-							messageID{
+							&messageID{
 								int64(response.MessageId.GetLedgerId()),
 								int64(response.MessageId.GetEntryId()),
 								-1,
@@ -1180,7 +1180,7 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 								0,
 							})
 						// use chunkMsgID to set msgID
-						msgID = sr.chunkRecorder.chunkedMsgID
+						msgID = &sr.chunkRecorder.chunkedMsgID
 					}
 				}
 
@@ -1375,11 +1375,11 @@ func newChunkRecorder() *chunkRecorder {
 	}
 }
 
-func (c *chunkRecorder) setFirstChunkID(msgID messageID) {
+func (c *chunkRecorder) setFirstChunkID(msgID *messageID) {
 	c.chunkedMsgID.firstChunkID = msgID
 }
 
-func (c *chunkRecorder) setLastChunkID(msgID messageID) {
+func (c *chunkRecorder) setLastChunkID(msgID *messageID) {
 	c.chunkedMsgID.messageID = msgID
 }
 
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 079754b..68dd084 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -37,7 +37,7 @@ type reader struct {
 	client              *client
 	pc                  *partitionConsumer
 	messageCh           chan ConsumerMessage
-	lastMessageInBroker trackingMessageID
+	lastMessageInBroker *trackingMessageID
 	log                 log.Logger
 	metrics             *internal.LeveledMetrics
 }
@@ -51,8 +51,8 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		return nil, newError(InvalidConfiguration, "StartMessageID is required")
 	}
 
-	startMessageID, ok := toTrackingMessageID(options.StartMessageID)
-	if !ok {
+	startMessageID := toTrackingMessageID(options.StartMessageID)
+	if startMessageID == nil {
 		// a custom type satisfying MessageID may not be a messageID or trackingMessageID
 		// so re-create messageID using its data
 		deserMsgID, err := deserializeMessageID(options.StartMessageID.Serialize())
@@ -60,8 +60,8 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 			return nil, err
 		}
 		// de-serialized MessageID is a messageID
-		startMessageID = trackingMessageID{
-			messageID:    deserMsgID.(messageID),
+		startMessageID = &trackingMessageID{
+			messageID:    deserMsgID.(*messageID),
 			receivedTime: time.Now(),
 		}
 	}
@@ -148,7 +148,7 @@ func (r *reader) Next(ctx context.Context) (Message, error) {
 			// Acknowledge message immediately because the reader is based on non-durable subscription. When it reconnects,
 			// it will specify the subscription position anyway
 			msgID := cm.Message.ID()
-			if mid, ok := toTrackingMessageID(msgID); ok {
+			if mid := toTrackingMessageID(msgID); mid != nil {
 				r.pc.lastDequeuedMsg = mid
 				r.pc.AckID(mid)
 				return cm.Message, nil
@@ -161,7 +161,7 @@ func (r *reader) Next(ctx context.Context) (Message, error) {
 }
 
 func (r *reader) HasNext() bool {
-	if !r.lastMessageInBroker.Undefined() && r.hasMoreMessages() {
+	if r.lastMessageInBroker != nil && r.hasMoreMessages() {
 		return true
 	}
 
@@ -180,7 +180,7 @@ func (r *reader) HasNext() bool {
 }
 
 func (r *reader) hasMoreMessages() bool {
-	if !r.pc.lastDequeuedMsg.Undefined() {
+	if r.pc.lastDequeuedMsg != nil {
 		return r.lastMessageInBroker.isEntryIDValid() && r.lastMessageInBroker.greater(r.pc.lastDequeuedMsg.messageID)
 	}
 
@@ -200,29 +200,29 @@ func (r *reader) Close() {
 	r.metrics.ReadersClosed.Inc()
 }
 
-func (r *reader) messageID(msgID MessageID) (trackingMessageID, bool) {
-	mid, ok := toTrackingMessageID(msgID)
-	if !ok {
+func (r *reader) messageID(msgID MessageID) *trackingMessageID {
+	mid := toTrackingMessageID(msgID)
+	if mid == nil {
 		r.log.Warnf("invalid message id type %T", msgID)
-		return trackingMessageID{}, false
+		return nil
 	}
 
 	partition := int(mid.partitionIdx)
 	// did we receive a valid partition index?
 	if partition < 0 {
 		r.log.Warnf("invalid partition index %d expected", partition)
-		return trackingMessageID{}, false
+		return nil
 	}
 
-	return mid, true
+	return mid
 }
 
 func (r *reader) Seek(msgID MessageID) error {
 	r.Lock()
 	defer r.Unlock()
 
-	mid, ok := r.messageID(msgID)
-	if !ok {
+	mid := r.messageID(msgID)
+	if mid == nil {
 		return nil
 	}
 


Test patch:
diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go
index d7903e8..e7a6725 100644
--- a/pulsar/ack_grouping_tracker_test.go
+++ b/pulsar/ack_grouping_tracker_test.go
@@ -184,14 +184,14 @@ func TestTimedTrackerCumulativeAck(t *testing.T) {
 func TestTimedTrackerIsDuplicate(t *testing.T) {
 	tracker := newAckGroupingTracker(nil, func(id MessageID) {}, func(id MessageID) {})
 
-	tracker.add(messageID{batchIdx: 0, batchSize: 3})
-	tracker.add(messageID{batchIdx: 2, batchSize: 3})
-	assert.True(t, tracker.isDuplicate(messageID{batchIdx: 0, batchSize: 3}))
-	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 1, batchSize: 3}))
-	assert.True(t, tracker.isDuplicate(messageID{batchIdx: 2, batchSize: 3}))
+	tracker.add(&messageID{batchIdx: 0, batchSize: 3})
+	tracker.add(&messageID{batchIdx: 2, batchSize: 3})
+	assert.True(t, tracker.isDuplicate(&messageID{batchIdx: 0, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 1, batchSize: 3}))
+	assert.True(t, tracker.isDuplicate(&messageID{batchIdx: 2, batchSize: 3}))
 
 	tracker.flush()
-	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 0, batchSize: 3}))
-	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 1, batchSize: 3}))
-	assert.False(t, tracker.isDuplicate(messageID{batchIdx: 2, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 0, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 1, batchSize: 3}))
+	assert.False(t, tracker.isDuplicate(&messageID{batchIdx: 2, batchSize: 3}))
 }
diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index b8efe9a..bb28371 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"log"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -1123,3 +1124,117 @@ func TestServiceUrlTLSWithTLSTransportWithBasicAuth(t *testing.T) {
 func TestWebServiceUrlTLSWithTLSTransportWithBasicAuth(t *testing.T) {
 	testTLSTransportWithBasicAuth(t, webServiceURLTLS)
 }
+
+func TestConfigureConnectionMaxIdleTime(t *testing.T) {
+	_, err := NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: 1 * time.Second,
+	})
+
+	assert.Error(t, err, "Should be failed when the connectionMaxIdleTime is less than minConnMaxIdleTime")
+
+	cli, err := NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: -1, // Disabled
+	})
+
+	assert.Nil(t, err)
+	cli.Close()
+
+	cli, err = NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: 60 * time.Second,
+	})
+
+	assert.Nil(t, err)
+	cli.Close()
+}
+
+func testSendAndReceive(t *testing.T, producer Producer, consumer Consumer) {
+	// send 10 messages
+	for i := 0; i < 10; i++ {
+		if _, err := producer.Send(context.Background(), &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		}); err != nil {
+			log.Fatal(err)
+		}
+	}
+
+	// receive 10 messages
+	for i := 0; i < 10; i++ {
+		msg, err := consumer.Receive(context.Background())
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		expectMsg := fmt.Sprintf("hello-%d", i)
+		assert.Equal(t, []byte(expectMsg), msg.Payload())
+		// ack message
+		err = consumer.Ack(msg)
+		if err != nil {
+			return
+		}
+	}
+}
+
+func TestAutoCloseIdleConnection(t *testing.T) {
+	cli, err := NewClient(ClientOptions{
+		URL:                   serviceURL,
+		ConnectionMaxIdleTime: -1, // Disable auto release connections first, we will enable it manually later
+	})
+
+	assert.Nil(t, err)
+
+	topic := "TestAutoCloseIdleConnection"
+
+	// create consumer
+	consumer1, err := cli.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+
+	// create producer
+	producer1, err := cli.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+
+	testSendAndReceive(t, producer1, consumer1)
+
+	pool := cli.(*client).cnxPool
+
+	producer1.Close()
+	consumer1.Close()
+
+	assert.NotEqual(t, 0, internal.GetConnectionsCount(&pool))
+
+	internal.StartCleanConnectionsTask(&pool, 2*time.Second) // Enable auto idle connections release manually
+
+	time.Sleep(6 * time.Second) // Need to wait at least 3 * ConnectionMaxIdleTime
+
+	assert.Equal(t, 0, internal.GetConnectionsCount(&pool))
+
+	// create consumer
+	consumer2, err := cli.Subscribe(ConsumerOptions{
+		Topic:            topic,
+		SubscriptionName: "my-sub",
+	})
+	assert.Nil(t, err)
+
+	// create producer
+	producer2, err := cli.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: false,
+	})
+	assert.Nil(t, err)
+
+	// Ensure the client still works
+	testSendAndReceive(t, producer2, consumer2)
+
+	producer2.Close()
+	consumer2.Close()
+
+	cli.Close()
+}
diff --git a/pulsar/consumer_partition_test.go b/pulsar/consumer_partition_test.go
index b9a9a02..16c4399 100644
--- a/pulsar/consumer_partition_test.go
+++ b/pulsar/consumer_partition_test.go
@@ -48,11 +48,11 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 	// ensure the tracker was set on the message id
 	messages := <-pc.queueCh
 	for _, m := range messages {
-		assert.Nil(t, m.ID().(trackingMessageID).tracker)
+		assert.Nil(t, m.ID().(*trackingMessageID).tracker)
 	}
 
 	// ack the message id
-	pc.AckID(messages[0].msgID.(trackingMessageID))
+	pc.AckID(messages[0].msgID.(*trackingMessageID))
 
 	select {
 	case <-eventsCh:
@@ -86,11 +86,12 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 	// ensure the tracker was set on the message id
 	messages := <-pc.queueCh
 	for _, m := range messages {
-		assert.Nil(t, m.ID().(trackingMessageID).tracker)
+		assert.Nil(t, m.ID().(*trackingMessageID).tracker)
 	}
 
 	// ack the message id
-	pc.AckID(messages[0].msgID.(trackingMessageID))
+	err := pc.AckID(messages[0].msgID.(*trackingMessageID))
+	assert.Nil(t, err)
 
 	select {
 	case <-eventsCh:
@@ -120,12 +121,13 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 	// ensure the tracker was set on the message id
 	messages := <-pc.queueCh
 	for _, m := range messages {
-		assert.NotNil(t, m.ID().(trackingMessageID).tracker)
+		assert.NotNil(t, m.ID().(*trackingMessageID).tracker)
 	}
 
 	// ack all message ids except the last one
 	for i := 0; i < 9; i++ {
-		pc.AckID(messages[i].msgID.(trackingMessageID))
+		err := pc.AckID(messages[i].msgID.(*trackingMessageID))
+		assert.Nil(t, err)
 	}
 
 	select {
@@ -135,7 +137,8 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 	}
 
 	// ack last message
-	pc.AckID(messages[9].msgID.(trackingMessageID))
+	err := pc.AckID(messages[9].msgID.(*trackingMessageID))
+	assert.Nil(t, err)
 
 	select {
 	case <-eventsCh:
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index de90c0e..21fa7d0 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -910,7 +910,7 @@ func TestConsumerNoBatchCumulativeAck(t *testing.T) {
 
 		if i == N/2-1 {
 			// cumulative acks the first half of messages
-			consumer.AckCumulative(msg)
+			assert.Nil(t, consumer.AckCumulative(msg))
 		}
 	}
 
@@ -931,7 +931,7 @@ func TestConsumerNoBatchCumulativeAck(t *testing.T) {
 		assert.Nil(t, err)
 		assert.Equal(t, fmt.Sprintf("msg-content-%d", i), string(msg.Payload()))
 
-		consumer.Ack(msg)
+		assert.Nil(t, consumer.Ack(msg))
 	}
 }
 
diff --git a/pulsar/impl_message_test.go b/pulsar/impl_message_test.go
index 413a39f..6a21171 100644
--- a/pulsar/impl_message_test.go
+++ b/pulsar/impl_message_test.go
@@ -31,11 +31,11 @@ func TestMessageId(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, id2)
 
-	assert.Equal(t, int64(1), id2.(messageID).ledgerID)
-	assert.Equal(t, int64(2), id2.(messageID).entryID)
-	assert.Equal(t, int32(3), id2.(messageID).batchIdx)
-	assert.Equal(t, int32(4), id2.(messageID).partitionIdx)
-	assert.Equal(t, int32(5), id2.(messageID).batchSize)
+	assert.Equal(t, int64(1), id2.(*messageID).ledgerID)
+	assert.Equal(t, int64(2), id2.(*messageID).entryID)
+	assert.Equal(t, int32(3), id2.(*messageID).batchIdx)
+	assert.Equal(t, int32(4), id2.(*messageID).partitionIdx)
+	assert.Equal(t, int32(5), id2.(*messageID).batchSize)
 
 	id, err = DeserializeMessageID(nil)
 	assert.Error(t, err)
@@ -110,7 +110,7 @@ func TestAckingMessageIDBatchOne(t *testing.T) {
 
 func TestAckingMessageIDBatchTwo(t *testing.T) {
 	tracker := newAckTracker(2)
-	ids := []trackingMessageID{
+	ids := []*trackingMessageID{
 		newTrackingMessageID(1, 1, 0, 0, 0, tracker),
 		newTrackingMessageID(1, 1, 1, 0, 0, tracker),
 	}
@@ -121,7 +121,7 @@ func TestAckingMessageIDBatchTwo(t *testing.T) {
 
 	// try reverse order
 	tracker = newAckTracker(2)
-	ids = []trackingMessageID{
+	ids = []*trackingMessageID{
 		newTrackingMessageID(1, 1, 0, 0, 0, tracker),
 		newTrackingMessageID(1, 1, 1, 0, 0, tracker),
 	}
diff --git a/pulsar/impl_message_bench_test.go b/pulsar/internal/helper.go
similarity index 61%
rename from pulsar/impl_message_bench_test.go
rename to pulsar/internal/helper.go
index 4b6ca10..3bca1ee 100644
--- a/pulsar/impl_message_bench_test.go
+++ b/pulsar/internal/helper.go
@@ -15,35 +15,19 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package pulsar
+package internal
 
-import (
-	"testing"
-)
+import "time"
 
-var (
-	usedByProducer messageID
-	usedByConsumer trackingMessageID
-)
+// These method should only be used by tests
 
-func producerCall(id messageID) messageID {
-	id.entryID++
-	return id
+func StartCleanConnectionsTask(p *ConnectionPool, connectionMaxIdleTime time.Duration) {
+	go (*p).(*connectionPool).checkAndCleanIdleConnections(connectionMaxIdleTime)
 }
 
-func consumerCall(id trackingMessageID) trackingMessageID {
-	id.entryID++
-	return id
-}
-
-func BenchmarkProducerCall(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		usedByProducer = producerCall(usedByProducer)
-	}
-}
-
-func BenchmarkConsumerCall(b *testing.B) {
-	for i := 0; i < b.N; i++ {
-		usedByConsumer = consumerCall(usedByConsumer)
-	}
+func GetConnectionsCount(p *ConnectionPool) int {
+	pool := (*p).(*connectionPool)
+	pool.Lock()
+	defer pool.Unlock()
+	return len(pool.connections)
 }
diff --git a/pulsar/message_chunking_test.go b/pulsar/message_chunking_test.go
index b3d64af..aac87c7 100644
--- a/pulsar/message_chunking_test.go
+++ b/pulsar/message_chunking_test.go
@@ -454,10 +454,10 @@ func TestChunkSize(t *testing.T) {
 		})
 		assert.NoError(t, err)
 		if size <= payloadChunkSize {
-			_, ok := msgID.(messageID)
+			_, ok := msgID.(*messageID)
 			assert.Equal(t, true, ok)
 		} else {
-			_, ok := msgID.(chunkMessageID)
+			_, ok := msgID.(*chunkMessageID)
 			assert.Equal(t, true, ok)
 		}
 	}
diff --git a/pulsar/negative_acks_tracker_test.go b/pulsar/negative_acks_tracker_test.go
index 5faa947..12d33de 100644
--- a/pulsar/negative_acks_tracker_test.go
+++ b/pulsar/negative_acks_tracker_test.go
@@ -81,13 +81,13 @@ func TestNacksTracker(t *testing.T) {
 	nmc := newNackMockedConsumer(nil)
 	nacks := newNegativeAcksTracker(nmc, testNackDelay, nil, log.DefaultNopLogger())
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 1,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 2,
 		entryID:  2,
 		batchIdx: 1,
@@ -114,25 +114,25 @@ func TestNacksWithBatchesTracker(t *testing.T) {
 	nmc := newNackMockedConsumer(nil)
 	nacks := newNegativeAcksTracker(nmc, testNackDelay, nil, log.DefaultNopLogger())
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 1,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 2,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 3,
 	})
 
-	nacks.Add(messageID{
+	nacks.Add(&messageID{
 		ledgerID: 2,
 		entryID:  2,
 		batchIdx: 1,
@@ -194,7 +194,7 @@ func (msg *mockMessage1) Payload() []byte {
 }
 
 func (msg *mockMessage1) ID() MessageID {
-	return messageID{
+	return &messageID{
 		ledgerID: 1,
 		entryID:  1,
 		batchIdx: 1,
@@ -270,7 +270,7 @@ func (msg *mockMessage2) Payload() []byte {
 }
 
 func (msg *mockMessage2) ID() MessageID {
-	return messageID{
+	return &messageID{
 		ledgerID: 2,
 		entryID:  2,
 		batchIdx: 1,
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index f86d01a..a6f5e39 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -325,7 +325,7 @@ func TestFlushInProducer(t *testing.T) {
 		assert.Nil(t, err)
 		msgCount++
 
-		msgID := msg.ID().(trackingMessageID)
+		msgID := msg.ID().(*trackingMessageID)
 		// Since messages are batched, they will be sharing the same ledgerId/entryId
 		if ledgerID == -1 {
 			ledgerID = msgID.ledgerID
@@ -742,7 +742,7 @@ func TestBatchDelayMessage(t *testing.T) {
 	var delayMsgID int64
 	ch := make(chan struct{}, 2)
 	producer.SendAsync(ctx, delayMsg, func(id MessageID, producerMessage *ProducerMessage, err error) {
-		atomic.StoreInt64(&delayMsgID, id.(messageID).entryID)
+		atomic.StoreInt64(&delayMsgID, id.(*messageID).entryID)
 		ch <- struct{}{}
 	})
 	delayMsgPublished := false
@@ -758,13 +758,13 @@ func TestBatchDelayMessage(t *testing.T) {
 	}
 	var noDelayMsgID int64
 	producer.SendAsync(ctx, noDelayMsg, func(id MessageID, producerMessage *ProducerMessage, err error) {
-		atomic.StoreInt64(&noDelayMsgID, id.(messageID).entryID)
+		atomic.StoreInt64(&noDelayMsgID, id.(*messageID).entryID)
 	})
 	for i := 0; i < 2; i++ {
 		msg, err := consumer.Receive(context.Background())
 		assert.Nil(t, err, "unexpected error occurred when recving message from topic")
 
-		switch msg.ID().(trackingMessageID).entryID {
+		switch msg.ID().(*trackingMessageID).entryID {
 		case atomic.LoadInt64(&noDelayMsgID):
 			assert.LessOrEqual(t, time.Since(msg.PublishTime()).Nanoseconds(), int64(batchingDelay*2))
 		case atomic.LoadInt64(&delayMsgID):