Repository: apache/pulsar-client-go
Pull request #972 (2023-02-28 15:36:54)

[refactor]: Refactor the toTrackingMessageID()
### Motivation

`toTrackingMessageID()` is a function that is widely used in `consumer` implementation. It can convert an interface type  `MessageID` into an struct type `trackingMessageID`. In addition, the second return value also plays a role in checking the `MessageID` type. In other words, it indicates that `MessageID` **cannot** be a user-defined type. From the perspective of code readability, `toTrackingMessageID()` should not do both.

**Note**: After #968 , `toTrackingMessageID()` returns only a pointer now. The role of original `ok` is replaced by nil pointer  now. However, the main content discussed in this PR has not changed.

For example.

https://github.com/apache/pulsar-client-go/blob/e2ea255052e8a527091791ef368851d885ee2d45/pulsar/consumer_regex.go#L176-L181

This example is the correct usage. The `ok` returned by `toTrackingMessageID()` is used to reject user-defined `MessageID`.

https://github.com/apache/pulsar-client-go/blob/e2ea255052e8a527091791ef368851d885ee2d45/pulsar/consumer_partition.go#L470-L473

This example is a bit vague. The actual effect here is the same as the previous example. But it return an error `failed to convert trackingMessageID` which is confusing. 

https://github.com/apache/pulsar-client-go/blob/e2ea255052e8a527091791ef368851d885ee2d45/pulsar/consumer_partition.go#L1816-L1820

In this case. We just want to convert `MessageID` into `trackingMessageID`. We do not care what it really is because it's not possible an invalid `MessageID` implementation.

So, original `toTrackingMessageID()` needs to require a careful look to use it correctly. I think it would be better to split it into two different method. `toTrackingMessageID()` just do the struct conversion, which it's more clearly. And when the new messageID type is added, we can just modify the `checkMessageIDType`.

### Modifications

- Refactor the `toTrackingMessageID()`
- Add the `checkMessageIDType()` to check whether `MessageID` is user-defined.

### Verifying this change

- [x] Make sure that the change passes the CI checks.

### Documentation

  - Does this pull request introduce a new feature? no


Gleiphir2769: /pulsarbot run-failure-checks
Gleiphir2769: > /pulsarbot run-failure-checks

Still doesn't work...
BewareMyPower: The flaky test is related to https://github.com/apache/pulsar-client-go/issues/971
Gleiphir2769: > The flaky test is related to #971

@BewareMyPower yes, rerun CI failed again. Looks like CI is stucked in `TestConsumerSeekByTimeOnPartitionedTopic` now.

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/auth/athenz.go b/pulsar/auth/athenz.go
index f490be9..35b3cc4 100644
--- a/pulsar/auth/athenz.go
+++ b/pulsar/auth/athenz.go
@@ -32,28 +32,33 @@ import (
 )
 
 const (
-	minExpire            = 2 * time.Hour
-	maxExpire            = 24 * time.Hour
-	AthenzRoleAuthHeader = "Athenz-Role-Auth"
+	minExpire         = 2 * time.Hour
+	maxExpire         = 24 * time.Hour
+	defaultKeyID      = "0"
+	defaultRoleHeader = "Athenz-Role-Auth"
 )
 
 type athenzAuthProvider struct {
-	providerDomain     string
-	tenantDomain       string
-	tenantService      string
-	privateKey         string
-	keyID              string
-	principalHeader    string
-	ztsURL             string
-	tokenBuilder       zms.TokenBuilder
-	roleToken          zts.RoleToken
-	zmsNewTokenBuilder func(domain, name string, privateKeyPEM []byte, keyVersion string) (zms.TokenBuilder, error)
-	ztsNewRoleToken    func(tok zms.Token, domain string, opts zts.RoleTokenOptions) zts.RoleToken
+	providerDomain          string
+	tenantDomain            string
+	tenantService           string
+	privateKey              string
+	keyID                   string
+	x509CertChain           string
+	caCert                  string
+	principalHeader         string
+	roleHeader              string
+	ztsURL                  string
+	tokenBuilder            zms.TokenBuilder
+	roleToken               zts.RoleToken
+	zmsNewTokenBuilder      func(domain, name string, privateKeyPEM []byte, keyVersion string) (zms.TokenBuilder, error)
+	ztsNewRoleToken         func(tok zms.Token, domain string, opts zts.RoleTokenOptions) zts.RoleToken
+	ztsNewRoleTokenFromCert func(certFile, keyFile, domain string, opts zts.RoleTokenOptions) zts.RoleToken
 
 	T http.RoundTripper
 }
 
-type privateKeyURI struct {
+type parsedURI struct {
 	Scheme                   string
 	MediaTypeAndEncodingType string
 	Data                     string
@@ -67,7 +72,10 @@ func NewAuthenticationAthenzWithParams(params map[string]string) (Provider, erro
 		params["tenantService"],
 		params["privateKey"],
 		params["keyId"],
+		params["x509CertChain"],
+		params["caCert"],
 		params["principalHeader"],
+		params["roleHeader"],
 		params["ztsUrl"],
 	), nil
 }
@@ -78,68 +86,97 @@ func NewAuthenticationAthenz(
 	tenantService string,
 	privateKey string,
 	keyID string,
+	x509CertChain string,
+	caCert string,
 	principalHeader string,
+	roleHeader string,
 	ztsURL string) Provider {
-	var fixedKeyID string
-	if keyID == "" {
-		fixedKeyID = "0"
-	} else {
+	fixedKeyID := defaultKeyID
+	if keyID != "" {
 		fixedKeyID = keyID
 	}
+
+	fixedRoleHeader := defaultRoleHeader
+	if roleHeader != "" {
+		fixedRoleHeader = roleHeader
+	}
+
 	ztsNewRoleToken := func(tok zms.Token, domain string, opts zts.RoleTokenOptions) zts.RoleToken {
 		return zts.RoleToken(zts.NewRoleToken(tok, domain, opts))
 	}
 
+	ztsNewRoleTokenFromCert := func(certFile, keyFile, domain string, opts zts.RoleTokenOptions) zts.RoleToken {
+		return zts.RoleToken(zts.NewRoleTokenFromCert(certFile, keyFile, domain, opts))
+	}
+
 	return &athenzAuthProvider{
-		providerDomain:     providerDomain,
-		tenantDomain:       tenantDomain,
-		tenantService:      tenantService,
-		privateKey:         privateKey,
-		keyID:              fixedKeyID,
-		principalHeader:    principalHeader,
-		ztsURL:             strings.TrimSuffix(ztsURL, "/"),
-		zmsNewTokenBuilder: zms.NewTokenBuilder,
-		ztsNewRoleToken:    ztsNewRoleToken,
+		providerDomain:          providerDomain,
+		tenantDomain:            tenantDomain,
+		tenantService:           tenantService,
+		privateKey:              privateKey,
+		keyID:                   fixedKeyID,
+		x509CertChain:           x509CertChain,
+		caCert:                  caCert,
+		principalHeader:         principalHeader,
+		roleHeader:              fixedRoleHeader,
+		ztsURL:                  strings.TrimSuffix(ztsURL, "/"),
+		zmsNewTokenBuilder:      zms.NewTokenBuilder,
+		ztsNewRoleToken:         ztsNewRoleToken,
+		ztsNewRoleTokenFromCert: ztsNewRoleTokenFromCert,
 	}
 }
 
 func (p *athenzAuthProvider) Init() error {
-	uriSt := parseURI(p.privateKey)
-	var keyData []byte
+	if p.providerDomain == "" || p.privateKey == "" || p.ztsURL == "" {
+		return errors.New("missing required parameters")
+	}
 
-	if uriSt.Scheme == "data" {
-		if uriSt.MediaTypeAndEncodingType != "application/x-pem-file;base64" {
-			return errors.New("Unsupported mediaType or encodingType: " + uriSt.MediaTypeAndEncodingType)
+	var roleToken zts.RoleToken
+	opts := zts.RoleTokenOptions{
+		BaseZTSURL: p.ztsURL + "/zts/v1",
+		MinExpire:  minExpire,
+		MaxExpire:  maxExpire,
+		AuthHeader: p.principalHeader,
+	}
+
+	if p.x509CertChain != "" {
+		// use Copper Argos
+		certURISt := parseURI(p.x509CertChain)
+		keyURISt := parseURI(p.privateKey)
+
+		if certURISt.Scheme != "file" || keyURISt.Scheme != "file" {
+			return errors.New("x509CertChain and privateKey must be specified as file paths")
+		}
+
+		if p.caCert != "" {
+			caCertData, err := loadPEM(p.caCert)
+			if err != nil {
+				return err
+			}
+			opts.CACert = caCertData
 		}
-		key, err := base64.StdEncoding.DecodeString(uriSt.Data)
+
+		roleToken = p.ztsNewRoleTokenFromCert(certURISt.Path, keyURISt.Path, p.providerDomain, opts)
+	} else {
+		if p.tenantDomain == "" || p.tenantService == "" {
+			return errors.New("missing required parameters")
+		}
+
+		keyData, err := loadPEM(p.privateKey)
 		if err != nil {
 			return err
 		}
-		keyData = key
-	} else if uriSt.Scheme == "file" {
-		key, err := os.ReadFile(uriSt.Path)
+
+		tb, err := p.zmsNewTokenBuilder(p.tenantDomain, p.tenantService, keyData, p.keyID)
 		if err != nil {
 			return err
 		}
-		keyData = key
-	} else {
-		return errors.New("Unsupported URI Scheme: " + uriSt.Scheme)
-	}
+		p.tokenBuilder = tb
 
-	tb, err := p.zmsNewTokenBuilder(p.tenantDomain, p.tenantService, keyData, p.keyID)
-	if err != nil {
-		return err
+		roleToken = p.ztsNewRoleToken(p.tokenBuilder.Token(), p.providerDomain, opts)
 	}
-	p.tokenBuilder = tb
 
-	roleToken := p.ztsNewRoleToken(p.tokenBuilder.Token(), p.providerDomain, zts.RoleTokenOptions{
-		BaseZTSURL: p.ztsURL + "/zts/v1",
-		MinExpire:  minExpire,
-		MaxExpire:  maxExpire,
-		AuthHeader: p.principalHeader,
-	})
 	p.roleToken = roleToken
-
 	return nil
 }
 
@@ -164,8 +201,8 @@ func (p *athenzAuthProvider) Close() error {
 	return nil
 }
 
-func parseURI(uri string) privateKeyURI {
-	var uriSt privateKeyURI
+func parseURI(uri string) parsedURI {
+	var uriSt parsedURI
 	// scheme mediatype[;base64] path file
 	const expression = `^(?:([^:/?#]+):)(?:([;/\\\-\w]*),)?(?:/{0,2}((?:[^?#/]*/)*))?([^?#]*)`
 
@@ -177,17 +214,47 @@ func parseURI(uri string) privateKeyURI {
 		uriSt.MediaTypeAndEncodingType = groups[2]
 		uriSt.Data = groups[4]
 		uriSt.Path = groups[3] + groups[4]
+	} else {
+		// consider a file path specified instead of a URI
+		uriSt.Scheme = "file"
+		uriSt.Path = uri
 	}
 
 	return uriSt
 }
 
+func loadPEM(uri string) ([]byte, error) {
+	uriSt := parseURI(uri)
+	var pemData []byte
+
+	if uriSt.Scheme == "data" {
+		if uriSt.MediaTypeAndEncodingType != "application/x-pem-file;base64" {
+			return nil, errors.New("Unsupported mediaType or encodingType: " + uriSt.MediaTypeAndEncodingType)
+		}
+		pem, err := base64.StdEncoding.DecodeString(uriSt.Data)
+		if err != nil {
+			return nil, err
+		}
+		pemData = pem
+	} else if uriSt.Scheme == "file" {
+		pem, err := os.ReadFile(uriSt.Path)
+		if err != nil {
+			return nil, err
+		}
+		pemData = pem
+	} else {
+		return nil, errors.New("Unsupported URI Scheme: " + uriSt.Scheme)
+	}
+
+	return pemData, nil
+}
+
 func (p *athenzAuthProvider) RoundTrip(req *http.Request) (*http.Response, error) {
 	tok, err := p.roleToken.RoleTokenValue()
 	if err != nil {
 		return nil, err
 	}
-	req.Header.Add(AthenzRoleAuthHeader, tok)
+	req.Header.Add(p.roleHeader, tok)
 	return p.T.RoundTrip(req)
 }
 
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 d16f719..d19e522 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -531,10 +531,17 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
 	if delay < 0 {
 		delay = 0
 	}
+
+	if !checkMessageIDType(msg.ID()) {
+		c.log.Warnf("invalid message id type %T", msg.ID())
+		return
+	}
+
 	msgID := c.messageID(msg.ID())
 	if msgID == nil {
 		return
 	}
+
 	props := make(map[string]string)
 	for k, v := range msg.Properties() {
 		props[k] = v
@@ -580,6 +587,10 @@ func (c *consumer) ReconsumeLaterWithCustomProperties(msg Message, customPropert
 }
 
 func (c *consumer) Nack(msg Message) {
+	if !checkMessageIDType(msg.ID()) {
+		c.log.Warnf("invalid message id type %T", msg.ID())
+		return
+	}
 	if c.options.EnableDefaultNackBackoffPolicy || c.options.NackBackoffPolicy != nil {
 		mid := c.messageID(msg.ID())
 		if mid == nil {
@@ -587,7 +598,7 @@ func (c *consumer) Nack(msg Message) {
 		}
 
 		if mid.consumer != nil {
-			mid.consumer.NackID(msg.ID())
+			mid.NackByMsg(msg)
 			return
 		}
 		c.consumers[mid.partitionIdx].NackMsg(msg)
@@ -745,10 +756,6 @@ func toProtoInitialPosition(p SubscriptionInitialPosition) pb.CommandSubscribe_I
 
 func (c *consumer) messageID(msgID MessageID) *trackingMessageID {
 	mid := toTrackingMessageID(msgID)
-	if mid == nil {
-		c.log.Warnf("invalid message id type %T", msgID)
-		return nil
-	}
 
 	partition := int(mid.partitionIdx)
 	// did we receive a valid partition index?
diff --git a/pulsar/consumer_multitopic.go b/pulsar/consumer_multitopic.go
index c0fcaef..8108c29 100644
--- a/pulsar/consumer_multitopic.go
+++ b/pulsar/consumer_multitopic.go
@@ -125,11 +125,11 @@ 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 := toTrackingMessageID(msgID)
-	if mid == nil {
+	if !checkMessageIDType(msgID) {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type in multi_consumer")
 	}
+	mid := toTrackingMessageID(msgID)
 
 	if mid.consumer == nil {
 		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
@@ -152,11 +152,11 @@ 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 := toTrackingMessageID(msgID)
-	if mid == nil {
+	if !checkMessageIDType(msgID) {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return errors.New("invalid message id type in multi_consumer")
 	}
+	mid := toTrackingMessageID(msgID)
 
 	if mid.consumer == nil {
 		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
@@ -203,11 +203,11 @@ 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 := toTrackingMessageID(msgID)
-		if mid == nil {
+		if !checkMessageIDType(msgID) {
 			c.log.Warnf("invalid message id type %T", msgID)
 			return
 		}
+		mid := toTrackingMessageID(msgID)
 
 		if mid.consumer == nil {
 			c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
@@ -221,11 +221,11 @@ func (c *multiTopicConsumer) Nack(msg Message) {
 }
 
 func (c *multiTopicConsumer) NackID(msgID MessageID) {
-	mid := toTrackingMessageID(msgID)
-	if mid == nil {
+	if !checkMessageIDType(msgID) {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return
 	}
+	mid := toTrackingMessageID(msgID)
 
 	if mid.consumer == nil {
 		c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 0f7af3d..a3dac19 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -465,9 +465,6 @@ func (pc *partitionConsumer) ackID(msgID MessageID, withResponse bool) error {
 	}
 
 	trackingID := toTrackingMessageID(msgID)
-	if trackingID == nil {
-		return errors.New("failed to convert trackingMessageID")
-	}
 
 	if trackingID != nil && trackingID.ack() {
 		pc.metrics.AcksCounter.Inc()
@@ -501,18 +498,34 @@ func (pc *partitionConsumer) sendIndividualAck(msgID MessageID) *ackRequest {
 }
 
 func (pc *partitionConsumer) AckIDWithResponse(msgID MessageID) error {
+	if !checkMessageIDType(msgID) {
+		pc.log.Errorf("invalid message id type %T", msgID)
+		return fmt.Errorf("invalid message id type %T", msgID)
+	}
 	return pc.ackID(msgID, true)
 }
 
 func (pc *partitionConsumer) AckID(msgID MessageID) error {
+	if !checkMessageIDType(msgID) {
+		pc.log.Errorf("invalid message id type %T", msgID)
+		return fmt.Errorf("invalid message id type %T", msgID)
+	}
 	return pc.ackID(msgID, false)
 }
 
 func (pc *partitionConsumer) AckIDCumulative(msgID MessageID) error {
+	if !checkMessageIDType(msgID) {
+		pc.log.Errorf("invalid message id type %T", msgID)
+		return fmt.Errorf("invalid message id type %T", msgID)
+	}
 	return pc.internalAckIDCumulative(msgID, false)
 }
 
 func (pc *partitionConsumer) AckIDWithResponseCumulative(msgID MessageID) error {
+	if !checkMessageIDType(msgID) {
+		pc.log.Errorf("invalid message id type %T", msgID)
+		return fmt.Errorf("invalid message id type %T", msgID)
+	}
 	return pc.internalAckIDCumulative(msgID, true)
 }
 
@@ -574,15 +587,17 @@ func (pc *partitionConsumer) sendCumulativeAck(msgID MessageID) *ackRequest {
 }
 
 func (pc *partitionConsumer) NackID(msgID MessageID) {
+	if !checkMessageIDType(msgID) {
+		pc.log.Warnf("invalid message id type %T", msgID)
+		return
+	}
+
 	if cmid, ok := msgID.(*chunkMessageID); ok {
 		pc.unAckChunksTracker.nack(cmid)
 		return
 	}
 
 	trackingID := toTrackingMessageID(msgID)
-	if trackingID == nil {
-		return
-	}
 
 	pc.nackTracker.Add(trackingID.messageID)
 	pc.metrics.NacksCounter.Inc()
@@ -665,16 +680,20 @@ func (pc *partitionConsumer) Seek(msgID MessageID) error {
 		pc.log.WithField("state", state).Error("Failed to seek by closing or closed consumer")
 		return errors.New("failed to seek by closing or closed consumer")
 	}
+
+	if !checkMessageIDType(msgID) {
+		pc.log.Errorf("invalid message id type %T", msgID)
+		return fmt.Errorf("invalid message id type %T", msgID)
+	}
+
 	req := &seekRequest{
 		doneCh: make(chan struct{}),
 	}
 	if cmid, ok := msgID.(*chunkMessageID); ok {
 		req.msgID = cmid.firstChunkID
-	} else if tmid := toTrackingMessageID(msgID); tmid != nil {
-		req.msgID = tmid.messageID
 	} else {
-		// will never reach
-		return errors.New("unhandled messageID type")
+		tmid := toTrackingMessageID(msgID)
+		req.msgID = tmid.messageID
 	}
 
 	pc.ackGroupingTracker.flushAndClean()
@@ -1812,9 +1831,8 @@ func (c *chunkedMsgCtx) discard(pc *partitionConsumer) {
 			continue
 		}
 		pc.log.Info("Removing chunk message-id", mid.String())
-		if tmid := toTrackingMessageID(mid); tmid != nil {
-			pc.AckID(tmid)
-		}
+		tmid := toTrackingMessageID(mid)
+		pc.AckID(tmid)
 	}
 }
 
diff --git a/pulsar/consumer_regex.go b/pulsar/consumer_regex.go
index fdfecec..2520af5 100644
--- a/pulsar/consumer_regex.go
+++ b/pulsar/consumer_regex.go
@@ -174,12 +174,13 @@ 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 := toTrackingMessageID(msgID)
-	if mid == nil {
+	if !checkMessageIDType(msgID) {
 		c.log.Warnf("invalid message id type %T", msgID)
-		return errors.New("invalid message id type")
+		return fmt.Errorf("invalid message id type %T", msgID)
 	}
 
+	mid := toTrackingMessageID(msgID)
+
 	if mid.consumer == nil {
 		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
 		return errors.New("consumer is nil in consumer_regex")
@@ -201,12 +202,13 @@ 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 := toTrackingMessageID(msgID)
-	if mid == nil {
+	if !checkMessageIDType(msgID) {
 		c.log.Warnf("invalid message id type %T", msgID)
-		return errors.New("invalid message id type")
+		return fmt.Errorf("invalid message id type %T", msgID)
 	}
 
+	mid := toTrackingMessageID(msgID)
+
 	if mid.consumer == nil {
 		c.log.Warnf("unable to ack messageID=%+v can not determine topic", msgID)
 		return errors.New("unable to ack message because consumer is nil")
@@ -222,11 +224,11 @@ 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 := toTrackingMessageID(msgID)
-		if mid == nil {
+		if !checkMessageIDType(msgID) {
 			c.log.Warnf("invalid message id type %T", msgID)
 			return
 		}
+		mid := toTrackingMessageID(msgID)
 
 		if mid.consumer == nil {
 			c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
@@ -240,12 +242,13 @@ func (c *regexConsumer) Nack(msg Message) {
 }
 
 func (c *regexConsumer) NackID(msgID MessageID) {
-	mid := toTrackingMessageID(msgID)
-	if mid == nil {
+	if !checkMessageIDType(msgID) {
 		c.log.Warnf("invalid message id type %T", msgID)
 		return
 	}
 
+	mid := toTrackingMessageID(msgID)
+
 	if mid.consumer == nil {
 		c.log.Warnf("unable to nack messageID=%+v can not determine topic", msgID)
 		return
diff --git a/pulsar/impl_message.go b/pulsar/impl_message.go
index 68ddecd..9c56070 100644
--- a/pulsar/impl_message.go
+++ b/pulsar/impl_message.go
@@ -213,6 +213,16 @@ func newMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx in
 	}
 }
 
+func fromMessageID(msgID MessageID) *messageID {
+	return &messageID{
+		ledgerID:     msgID.LedgerID(),
+		entryID:      msgID.EntryID(),
+		batchIdx:     msgID.BatchIdx(),
+		partitionIdx: msgID.PartitionIdx(),
+		batchSize:    msgID.BatchSize(),
+	}
+}
+
 func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partitionIdx int32, batchSize int32,
 	tracker *ackTracker) *trackingMessageID {
 	return &trackingMessageID{
@@ -228,22 +238,26 @@ func newTrackingMessageID(ledgerID int64, entryID int64, batchIdx int32, partiti
 	}
 }
 
-func toTrackingMessageID(msgID MessageID) *trackingMessageID {
-	if mid, ok := msgID.(*messageID); ok {
-		return &trackingMessageID{
-			messageID:    mid,
-			receivedTime: time.Now(),
-		}
-	} else if mid, ok := msgID.(*trackingMessageID); ok {
+// checkMessageIDType checks if the MessageID is user-defined
+func checkMessageIDType(msgID MessageID) (valid bool) {
+	switch msgID.(type) {
+	case *trackingMessageID:
+		return true
+	case *chunkMessageID:
+		return true
+	case *messageID:
+		return true
+	default:
+		return false
+	}
+}
+
+func toTrackingMessageID(msgID MessageID) (trackingMsgID *trackingMessageID) {
+	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,
-		}
-	} else {
-		return nil
+	}
+	return &trackingMessageID{
+		messageID: fromMessageID(msgID),
 	}
 }
 
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..6ff7991 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,26 @@ 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) {
+					p.log.Debugf("Closed connection from pool due to inactivity. logical_addr=%+v physical_addr=%+v",
+						c.logicalAddr, c.physicalAddr)
+					delete(p.connections, k)
+					c.Close()
+				}
+			}
+			p.Unlock()
+		}
+	}
+}
diff --git a/pulsar/internal/helper.go b/pulsar/internal/helper.go
new file mode 100644
index 0000000..3bca1ee
--- /dev/null
+++ b/pulsar/internal/helper.go
@@ -0,0 +1,33 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package internal
+
+import "time"
+
+// These method should only be used by tests
+
+func StartCleanConnectionsTask(p *ConnectionPool, connectionMaxIdleTime time.Duration) {
+	go (*p).(*connectionPool).checkAndCleanIdleConnections(connectionMaxIdleTime)
+}
+
+func GetConnectionsCount(p *ConnectionPool) int {
+	pool := (*p).(*connectionPool)
+	pool.Lock()
+	defer pool.Unlock()
+	return len(pool.connections)
+}
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 68dd084..c7620ad 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -51,8 +51,8 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		return nil, newError(InvalidConfiguration, "StartMessageID is required")
 	}
 
-	startMessageID := toTrackingMessageID(options.StartMessageID)
-	if startMessageID == nil {
+	var startMessageID *trackingMessageID
+	if !checkMessageIDType(options.StartMessageID) {
 		// 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,10 +60,9 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 			return nil, err
 		}
 		// de-serialized MessageID is a messageID
-		startMessageID = &trackingMessageID{
-			messageID:    deserMsgID.(*messageID),
-			receivedTime: time.Now(),
-		}
+		startMessageID = toTrackingMessageID(deserMsgID)
+	} else {
+		startMessageID = toTrackingMessageID(options.StartMessageID)
 	}
 
 	subscriptionName := options.SubscriptionName
@@ -148,12 +147,10 @@ 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 := toTrackingMessageID(msgID); mid != nil {
-				r.pc.lastDequeuedMsg = mid
-				r.pc.AckID(mid)
-				return cm.Message, nil
-			}
-			return nil, newError(InvalidMessage, fmt.Sprintf("invalid message id type %T", msgID))
+			mid := toTrackingMessageID(msgID)
+			r.pc.lastDequeuedMsg = mid
+			r.pc.AckID(mid)
+			return cm.Message, nil
 		case <-ctx.Done():
 			return nil, ctx.Err()
 		}
@@ -202,10 +199,6 @@ func (r *reader) Close() {
 
 func (r *reader) messageID(msgID MessageID) *trackingMessageID {
 	mid := toTrackingMessageID(msgID)
-	if mid == nil {
-		r.log.Warnf("invalid message id type %T", msgID)
-		return nil
-	}
 
 	partition := int(mid.partitionIdx)
 	// did we receive a valid partition index?
@@ -221,6 +214,11 @@ func (r *reader) Seek(msgID MessageID) error {
 	r.Lock()
 	defer r.Unlock()
 
+	if !checkMessageIDType(msgID) {
+		r.log.Warnf("invalid message id type %T", msgID)
+		return fmt.Errorf("invalid message id type %T", msgID)
+	}
+
 	mid := r.messageID(msgID)
 	if mid == nil {
 		return nil


Test patch:
diff --git a/pulsar/auth/athenz_test.go b/pulsar/auth/athenz_test.go
index 8541ca9..8911808 100644
--- a/pulsar/auth/athenz_test.go
+++ b/pulsar/auth/athenz_test.go
@@ -30,7 +30,11 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-const tlsClientKeyPath = "../../integration-tests/certs/client-key.pem"
+const (
+	clientKeyPath  = "../../integration-tests/certs/client-key.pem"
+	clientCertPath = "../../integration-tests/certs/client-cert.pem"
+	caCertPath     = "../../integration-tests/certs/cacert.pem"
+)
 
 type MockTokenBuilder struct {
 	mock.Mock
@@ -69,7 +73,7 @@ func (m *MockRoleToken) RoleTokenValue() (string, error) {
 
 func MockZmsNewTokenBuilder(domain, name string, privateKeyPEM []byte, keyVersion string) (zms.TokenBuilder, error) {
 	// assertion
-	key, err := os.ReadFile(tlsClientKeyPath)
+	key, err := os.ReadFile(clientKeyPath)
 	if err != nil {
 		return nil, err
 	}
@@ -105,16 +109,35 @@ func MockZtsNewRoleToken(tok zms.Token, domain string, opts zts.RoleTokenOptions
 	return mockRoleToken
 }
 
+func MockZtsNewRoleTokenFromCert(certFile, keyFile, domain string, opts zts.RoleTokenOptions) zts.RoleToken {
+	// assertion
+	if certFile != clientCertPath ||
+		keyFile != clientKeyPath ||
+		domain != "pulsar.test.provider" ||
+		opts.BaseZTSURL != "http://localhost:9999/zts/v1" ||
+		opts.AuthHeader != "" ||
+		opts.CACert == nil {
+		return nil
+	}
+
+	mockRoleToken := new(MockRoleToken)
+	mockRoleToken.On("RoleTokenValue").Return("mockRoleTokenFromCert", nil)
+	return mockRoleToken
+}
+
 func TestAthenzAuth(t *testing.T) {
-	privateKey := "file://" + tlsClientKeyPath
+	privateKey := "file://" + clientKeyPath
 	provider := NewAuthenticationAthenz(
-		"pulsar.test.provider",
-		"pulsar.test.tenant",
-		"service",
-		privateKey,
-		"",
-		"",
-		"http://localhost:9999")
+		"pulsar.test.provider",  // providerDomain
+		"pulsar.test.tenant",    // tenantDomain
+		"service",               // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		"",                      // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
 
 	// inject mock function
 	athenz := provider.(*athenzAuthProvider)
@@ -128,3 +151,105 @@ func TestAthenzAuth(t *testing.T) {
 	assert.Equal(t, []byte("mockRoleToken"), data)
 	assert.NoError(t, err)
 }
+
+func TestCopperArgos(t *testing.T) {
+	privateKey := "file://" + clientKeyPath
+	x509CertChain := "file://" + clientCertPath
+	caCert := "file://" + caCertPath
+
+	provider := NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		x509CertChain,           // x509CertChain
+		caCert,                  // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+
+	// inject mock function
+	athenz := provider.(*athenzAuthProvider)
+	athenz.ztsNewRoleTokenFromCert = MockZtsNewRoleTokenFromCert
+
+	err := athenz.Init()
+	assert.NoError(t, err)
+
+	data, err := athenz.GetData()
+	assert.Equal(t, []byte("mockRoleTokenFromCert"), data)
+	assert.NoError(t, err)
+}
+
+func TestIllegalParams(t *testing.T) {
+	privateKey := "file://" + clientKeyPath
+	x509CertChain := "file://" + clientCertPath
+
+	provider := NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		"",                      // privateKey
+		"",                      // keyID
+		"",                      // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz := provider.(*athenzAuthProvider)
+
+	err := athenz.Init()
+	assert.Error(t, err, "Should fail due to missing privateKey parameter")
+	assert.Equal(t, "missing required parameters", err.Error())
+
+	provider = NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"pulsar.test.tenant",    // tenantDomain
+		"",                      // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		"",                      // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz = provider.(*athenzAuthProvider)
+
+	err = athenz.Init()
+	assert.Error(t, err, "Should fail due to missing tenantService parameter")
+	assert.Equal(t, "missing required parameters", err.Error())
+
+	provider = NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		privateKey,              // privateKey
+		"",                      // keyID
+		"data:foo",              // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz = provider.(*athenzAuthProvider)
+
+	err = athenz.Init()
+	assert.Error(t, err, "Should fail due to incorrect x509CertChain scheme")
+	assert.Equal(t, "x509CertChain and privateKey must be specified as file paths", err.Error())
+
+	provider = NewAuthenticationAthenz(
+		"pulsar.test.provider",  // providerDomain
+		"",                      // tenantDomain
+		"",                      // tenantService
+		"data:bar",              // privateKey
+		"",                      // keyID
+		x509CertChain,           // x509CertChain
+		"",                      // caCert
+		"",                      // principalHeader
+		"",                      // roleHeader
+		"http://localhost:9999") // ztsURL
+	athenz = provider.(*athenzAuthProvider)
+
+	err = athenz.Init()
+	assert.Error(t, err, "Should fail due to incorrect privateKey scheme")
+	assert.Equal(t, "x509CertChain and privateKey must be specified as file paths", err.Error())
+}
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()
+}