Repository: apache/pulsar-client-go
Pull request #1280 (2024-09-04 03:23:55)

[fix] Only decompress the payload if it's not empty
### Motivation

The message payload is optional and in some cases only message properties are sent. In this case, the message decompression would fail.

### Modifications

Only do the decompression if the payload is not empty.

### Verifying this change

- [ ] Make sure that the change passes the CI checks.

This change is a trivial rework / code cleanup without any test coverage.

### 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): no
  - The public API: no
  - The schema: no
  - The default values of configurations: no
  - The wire protocol: no

### Documentation

  - Does this pull request introduce a new feature? no
  - If yes, how is the feature documented? not applicable
  - If a feature is not applicable for documentation, explain why? it's just a bugfix
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a07fa0c..b09abd3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -23,6 +23,45 @@
 
 All notable changes to this project will be documented in this file.
 
+[0.14.0] 2024-09-23
+
+## Important notice
+
+- The client has dropped support for Go 1.20. The minimum supported version is now Go
+  1.21. ([Related PR](https://github.com/apache/pulsar-client-go/pull/1249))
+- The `BackoffPolicy` interface and related methods has been moved to the `backoff` package from the `internal` package.
+  The `BackoffPolicy` interface is renamed to
+  `Policy`. ([Related PR](https://github.com/apache/pulsar-client-go/pull/1197))
+
+## What's Changed
+* [improve] Install golang by image in the Dockerfile by @nodece in https://github.com/apache/pulsar-client-go/pull/1246
+* [ci] Validate multiple version builds and use golangci-lint-action by @nodece in https://github.com/apache/pulsar-client-go/pull/1250
+* [ci] Add merge limitation by @nodece in https://github.com/apache/pulsar-client-go/pull/1248
+* [fix] Fix the key-based batch can't guarantee the ordering when flushing by @RobertIndie in https://github.com/apache/pulsar-client-go/pull/1252
+* [fix] FailTimeoutMessages cannot delete outdated messages by @nodece in https://github.com/apache/pulsar-client-go/pull/1247
+* [fix] Fix pulsar admin revoke subscription permission endpoint by @NoFacePeace in https://github.com/apache/pulsar-client-go/pull/1251
+* [fix] Producer close was blocked by @nodece in https://github.com/apache/pulsar-client-go/pull/1249
+* [fix] PulsarCtl 1266 Oauth2 Client credentials flow use scopes from the keyfile as well by @Nikolajls in https://github.com/apache/pulsar-client-go/pull/1244
+* [feat] Add support for subscription expiration time namespace settings by @klevy-toast in https://github.com/apache/pulsar-client-go/pull/1254
+* [fix] Prevent panic when calling Flush on closed producer by @Gilthoniel in https://github.com/apache/pulsar-client-go/pull/1260
+* [fix] Avoid a data race when flushing with load by @Gilthoniel in https://github.com/apache/pulsar-client-go/pull/1261
+* [improve] Add InitialSubscriptionName for DLQPolicy by @crossoverJie in https://github.com/apache/pulsar-client-go/pull/1264
+* [fix] Peek message will return -1 for partitionIndex by @shibd in https://github.com/apache/pulsar-client-go/pull/1267
+* [chore] Bump github.com/docker/docker from 27.0.3+incompatible to 27.1.1+incompatible by @dependabot in https://github.com/apache/pulsar-client-go/pull/1269
+* [feat] Support the namespace offloadThresholdInSeconds API in pulsaradmin pkg by @ericsyh in https://github.com/apache/pulsar-client-go/pull/1271
+* [fix] Stop timer when close timedAckGroupingTracker by @geniusjoe in https://github.com/apache/pulsar-client-go/pull/1279
+* [improve] Refactor connection concurrency model by @Gilthoniel in https://github.com/apache/pulsar-client-go/pull/1275
+* [fix] Attempt to avoid deadlock during reconnection by @Gilthoniel in https://github.com/apache/pulsar-client-go/pull/1273
+* [fix] Fixed panic caused by memory not aligned in arm32 arch by @dream-kzx in https://github.com/apache/pulsar-client-go/pull/1286
+* [fix] Reconnection logic and Backoff policy doesn't work correctly by @crossoverJie in https://github.com/apache/pulsar-client-go/pull/1197
+
+## New Contributors
+* @NoFacePeace made their first contribution in https://github.com/apache/pulsar-client-go/pull/1251
+* @Nikolajls made their first contribution in https://github.com/apache/pulsar-client-go/pull/1244
+* @klevy-toast made their first contribution in https://github.com/apache/pulsar-client-go/pull/1254
+* @ericsyh made their first contribution in https://github.com/apache/pulsar-client-go/pull/1271
+* @dream-kzx made their first contribution in https://github.com/apache/pulsar-client-go/pull/1286
+
 [0.13.1] 2024-08-02
 
 - [fix] Avoid a data race when flushing with load by Gaylor Bosson in [#1261](https://github.com/apache/pulsar-client-go/pull/1261)
diff --git a/pulsar/ack_grouping_tracker.go b/pulsar/ack_grouping_tracker.go
index dd05186..97d9e05 100644
--- a/pulsar/ack_grouping_tracker.go
+++ b/pulsar/ack_grouping_tracker.go
@@ -120,7 +120,7 @@ type timedAckGroupingTracker struct {
 
 	// Key is the pair of the ledger id and the entry id,
 	// Value is the bit set that represents which messages are acknowledged if the entry stores a batch.
-	// The bit 1 represents the message has been acknowledged, i.e. the bits "111" represents all messages
+	// The bit 1 represents the message has not been acknowledged, i.e. the bits "111" represents all messages
 	// in the batch whose batch size is 3 are not acknowledged.
 	// After the 1st message (i.e. batch index is 0) is acknowledged, the bits will become "011".
 	// Value is nil if the entry represents a single message.
@@ -241,6 +241,9 @@ func (t *timedAckGroupingTracker) clearPendingAcks() map[[2]uint64]*bitset.BitSe
 }
 
 func (t *timedAckGroupingTracker) close() {
+	if t.ticker != nil {
+		t.ticker.Stop()
+	}
 	t.flushAndClean()
 	if t.exitCh != nil {
 		close(t.exitCh)
diff --git a/pulsar/internal/backoff.go b/pulsar/backoff/backoff.go
similarity index 77%
rename from pulsar/internal/backoff.go
rename to pulsar/backoff/backoff.go
index 3284fb7..453da57 100644
--- a/pulsar/internal/backoff.go
+++ b/pulsar/backoff/backoff.go
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package internal
+package backoff
 
 import (
 	"math/rand"
@@ -26,10 +26,17 @@ func init() {
 	rand.Seed(time.Now().UnixNano())
 }
 
-// BackoffPolicy parameterize the following options in the reconnection logic to
+// Policy parameterize the following options in the reconnection logic to
 // allow users to customize the reconnection logic (minBackoff, maxBackoff and jitterPercentage)
-type BackoffPolicy interface {
+type Policy interface {
+	// Next returns the delay to wait before next retry
 	Next() time.Duration
+
+	// IsMaxBackoffReached evaluates if the max number of retries is reached
+	IsMaxBackoffReached() bool
+
+	// Reset the backoff to the initial state
+	Reset()
 }
 
 // DefaultBackoff computes the delay before retrying an action.
@@ -38,6 +45,13 @@ type DefaultBackoff struct {
 	backoff time.Duration
 }
 
+func NewDefaultBackoff() Policy {
+	return &DefaultBackoff{}
+}
+func NewDefaultBackoffWithInitialBackOff(backoff time.Duration) Policy {
+	return &DefaultBackoff{backoff: backoff / 2}
+}
+
 const maxBackoff = 60 * time.Second
 
 // Next returns the delay to wait before next retry
@@ -61,3 +75,7 @@ func (b *DefaultBackoff) Next() time.Duration {
 func (b *DefaultBackoff) IsMaxBackoffReached() bool {
 	return b.backoff >= maxBackoff
 }
+
+func (b *DefaultBackoff) Reset() {
+	b.backoff = 0
+}
diff --git a/pulsar/client.go b/pulsar/client.go
index d9ac2a7..173cca2 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -129,6 +129,8 @@ type ClientOptions struct {
 	// TLSMaxVersion contains the maximum TLS version that is acceptable. See tls.Config MaxVersion for more information.
 	TLSMaxVersion uint16
 
+	TLSConfig *tls.Config
+
 	// Configure the net model for vpc user to connect the pulsar broker
 	ListenerName string
 
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index e391c80..4996418 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -97,6 +97,7 @@ func newClient(options ClientOptions) (Client, error) {
 			CipherSuites:            options.TLSCipherSuites,
 			MinVersion:              options.TLSMinVersion,
 			MaxVersion:              options.TLSMaxVersion,
+			TLSConfig:               options.TLSConfig,
 		}
 	default:
 		return nil, newError(InvalidConfiguration, fmt.Sprintf("Invalid URL scheme '%s'", url.Scheme))
diff --git a/pulsar/consumer.go b/pulsar/consumer.go
index bf2eafb..880cad5 100644
--- a/pulsar/consumer.go
+++ b/pulsar/consumer.go
@@ -21,7 +21,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
 )
 
 // ConsumerMessage represents a pair of a Consumer and Message.
@@ -207,9 +207,9 @@ type ConsumerOptions struct {
 	// MaxReconnectToBroker sets the maximum retry number of reconnectToBroker. (default: ultimate)
 	MaxReconnectToBroker *uint
 
-	// BackoffPolicy parameterize the following options in the reconnection logic to
+	// BackOffPolicyFunc parameterize the following options in the reconnection logic to
 	// allow users to customize the reconnection logic (minBackoff, maxBackoff and jitterPercentage)
-	BackoffPolicy internal.BackoffPolicy
+	BackOffPolicyFunc func() backoff.Policy
 
 	// Decryption represents the encryption related fields required by the consumer to decrypt a message.
 	Decryption *MessageDecryptionInfo
diff --git a/pulsar/consumer_impl.go b/pulsar/consumer_impl.go
index 679054a..a3d3e3f 100644
--- a/pulsar/consumer_impl.go
+++ b/pulsar/consumer_impl.go
@@ -174,11 +174,12 @@ func newConsumer(client *client, options ConsumerOptions) (Consumer, error) {
 		}
 	}
 
-	dlq, err := newDlqRouter(client, options.DLQ, options.Topic, options.SubscriptionName, options.Name, client.log)
+	dlq, err := newDlqRouter(client, options.DLQ, options.Topic, options.SubscriptionName, options.Name,
+		options.BackOffPolicyFunc, client.log)
 	if err != nil {
 		return nil, err
 	}
-	rlq, err := newRetryRouter(client, options.DLQ, options.RetryEnable, client.log)
+	rlq, err := newRetryRouter(client, options.DLQ, options.RetryEnable, options.BackOffPolicyFunc, client.log)
 	if err != nil {
 		return nil, err
 	}
@@ -453,7 +454,7 @@ func newPartitionConsumerOpts(topic, consumerName string, idx int, options Consu
 		readCompacted:               options.ReadCompacted,
 		interceptors:                options.Interceptors,
 		maxReconnectToBroker:        options.MaxReconnectToBroker,
-		backoffPolicy:               options.BackoffPolicy,
+		backOffPolicyFunc:           options.BackOffPolicyFunc,
 		keySharedPolicy:             options.KeySharedPolicy,
 		schema:                      options.Schema,
 		decryption:                  options.Decryption,
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index d8001dc..823e0e8 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -26,6 +26,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
@@ -110,7 +112,7 @@ type partitionConsumerOpts struct {
 	disableForceTopicCreation   bool
 	interceptors                ConsumerInterceptors
 	maxReconnectToBroker        *uint
-	backoffPolicy               internal.BackoffPolicy
+	backOffPolicyFunc           func() backoff.Policy
 	keySharedPolicy             *KeySharedPolicy
 	schema                      Schema
 	decryption                  *MessageDecryptionInfo
@@ -182,6 +184,7 @@ type partitionConsumer struct {
 	lastMessageInBroker *trackingMessageID
 
 	redirectedClusterURI string
+	backoffPolicyFunc    func() backoff.Policy
 }
 
 func (pc *partitionConsumer) ActiveConsumerChanged(isActive bool) {
@@ -318,6 +321,13 @@ func (s *schemaInfoCache) add(schemaVersionHash string, schema Schema) {
 func newPartitionConsumer(parent Consumer, client *client, options *partitionConsumerOpts,
 	messageCh chan ConsumerMessage, dlq *dlqRouter,
 	metrics *internal.LeveledMetrics) (*partitionConsumer, error) {
+	var boFunc func() backoff.Policy
+	if options.backOffPolicyFunc != nil {
+		boFunc = options.backOffPolicyFunc
+	} else {
+		boFunc = backoff.NewDefaultBackoff
+	}
+
 	pc := &partitionConsumer{
 		parentConsumer:       parent,
 		client:               client,
@@ -332,13 +342,14 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		startMessageID:       atomicMessageID{msgID: options.startMessageID},
 		connectedCh:          make(chan struct{}),
 		messageCh:            messageCh,
-		connectClosedCh:      make(chan *connectionClosed, 10),
+		connectClosedCh:      make(chan *connectionClosed, 1),
 		closeCh:              make(chan struct{}),
 		clearQueueCh:         make(chan func(id *trackingMessageID)),
 		compressionProviders: sync.Map{},
 		dlq:                  dlq,
 		metrics:              metrics,
 		schemaInfoCache:      newSchemaInfoCache(client, options.topic),
+		backoffPolicyFunc:    boFunc,
 	}
 	if pc.options.autoReceiverQueueSize {
 		pc.currentQueueSize.Store(initialReceiverQueueSize)
@@ -581,12 +592,7 @@ func (pc *partitionConsumer) getLastMessageID() (*trackingMessageID, error) {
 		return nil, errors.New("failed to getLastMessageID for the closing or closed consumer")
 	}
 	remainTime := pc.client.operationTimeout
-	var backoff internal.BackoffPolicy
-	if pc.options.backoffPolicy != nil {
-		backoff = pc.options.backoffPolicy
-	} else {
-		backoff = &internal.DefaultBackoff{}
-	}
+	bo := pc.backoffPolicyFunc()
 	request := func() (*trackingMessageID, error) {
 		req := &getLastMsgIDRequest{doneCh: make(chan struct{})}
 		pc.eventsCh <- req
@@ -604,7 +610,7 @@ func (pc *partitionConsumer) getLastMessageID() (*trackingMessageID, error) {
 			pc.log.WithError(err).Error("Failed to getLastMessageID")
 			return nil, fmt.Errorf("failed to getLastMessageID due to %w", err)
 		}
-		nextDelay := backoff.Next()
+		nextDelay := bo.Next()
 		if nextDelay > remainTime {
 			nextDelay = remainTime
 		}
@@ -1099,15 +1105,18 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 		}
 	}
 
-	// decryption is success, decompress the payload
-	uncompressedHeadersAndPayload, err := pc.Decompress(msgMeta, processedPayloadBuffer)
-	if err != nil {
-		pc.discardCorruptedMessage(pbMsgID, pb.CommandAck_DecompressionError)
-		return err
-	}
+	var uncompressedHeadersAndPayload internal.Buffer
+	// decryption is success, decompress the payload, but only if payload is not empty
+	if n := msgMeta.UncompressedSize; n != nil && *n > 0 {
+		uncompressedHeadersAndPayload, err = pc.Decompress(msgMeta, processedPayloadBuffer)
+		if err != nil {
+			pc.discardCorruptedMessage(pbMsgID, pb.CommandAck_DecompressionError)
+			return err
+		}
 
-	// Reset the reader on the uncompressed buffer
-	reader.ResetBuffer(uncompressedHeadersAndPayload)
+		// Reset the reader on the uncompressed buffer
+		reader.ResetBuffer(uncompressedHeadersAndPayload)
+	}
 
 	numMsgs := 1
 	if msgMeta.NumMessagesInBatch != nil {
@@ -1381,8 +1390,12 @@ func (pc *partitionConsumer) ConnectionClosed(closeConsumer *pb.CommandCloseCons
 		assignedBrokerURL = pc.client.selectServiceURL(
 			closeConsumer.GetAssignedBrokerServiceUrl(), closeConsumer.GetAssignedBrokerServiceUrlTls())
 	}
-	pc.connectClosedCh <- &connectionClosed{
-		assignedBrokerURL: assignedBrokerURL,
+
+	select {
+	case pc.connectClosedCh <- &connectionClosed{assignedBrokerURL: assignedBrokerURL}:
+	default:
+		// Reconnect has already been requested so we do not block the
+		// connection callback.
 	}
 }
 
@@ -1680,18 +1693,17 @@ func (pc *partitionConsumer) internalClose(req *closeRequest) {
 }
 
 func (pc *partitionConsumer) reconnectToBroker(connectionClosed *connectionClosed) {
-	var maxRetry int
+	var (
+		maxRetry                                    int
+		delayReconnectTime, totalDelayReconnectTime time.Duration
+	)
 
 	if pc.options.maxReconnectToBroker == nil {
 		maxRetry = -1
 	} else {
 		maxRetry = int(*pc.options.maxReconnectToBroker)
 	}
-
-	var (
-		delayReconnectTime time.Duration
-		defaultBackoff     = internal.DefaultBackoff{}
-	)
+	bo := pc.backoffPolicyFunc()
 
 	for maxRetry != 0 {
 		if pc.getConsumerState() != consumerReady {
@@ -1706,11 +1718,10 @@ func (pc *partitionConsumer) reconnectToBroker(connectionClosed *connectionClose
 			delayReconnectTime = 0
 			assignedBrokerURL = connectionClosed.assignedBrokerURL
 			connectionClosed = nil // Attempt connecting to the assigned broker just once
-		} else if pc.options.backoffPolicy == nil {
-			delayReconnectTime = defaultBackoff.Next()
 		} else {
-			delayReconnectTime = pc.options.backoffPolicy.Next()
+			delayReconnectTime = bo.Next()
 		}
+		totalDelayReconnectTime += delayReconnectTime
 
 		pc.log.WithFields(log.Fields{
 			"assignedBrokerURL":  assignedBrokerURL,
@@ -1729,6 +1740,7 @@ func (pc *partitionConsumer) reconnectToBroker(connectionClosed *connectionClose
 		if err == nil {
 			// Successfully reconnected
 			pc.log.Info("Reconnected consumer to broker")
+			bo.Reset()
 			return
 		}
 		pc.log.WithError(err).Error("Failed to create consumer at reconnect")
@@ -1743,7 +1755,7 @@ func (pc *partitionConsumer) reconnectToBroker(connectionClosed *connectionClose
 			maxRetry--
 		}
 		pc.metrics.ConsumersReconnectFailure.Inc()
-		if maxRetry == 0 || defaultBackoff.IsMaxBackoffReached() {
+		if maxRetry == 0 || bo.IsMaxBackoffReached() {
 			pc.metrics.ConsumersReconnectMaxRetry.Inc()
 		}
 	}
diff --git a/pulsar/default_router.go b/pulsar/default_router.go
index 6945ff1..67ef306 100644
--- a/pulsar/default_router.go
+++ b/pulsar/default_router.go
@@ -24,11 +24,10 @@ import (
 )
 
 type defaultRouter struct {
+	lastBatchTimestamp     int64
 	currentPartitionCursor uint32
-
-	lastBatchTimestamp  int64
-	msgCounter          uint32
-	cumulativeBatchSize uint32
+	msgCounter             uint32
+	cumulativeBatchSize    uint32
 }
 
 // NewDefaultRouter set the message routing mode for the partitioned producer.
diff --git a/pulsar/dlq_router.go b/pulsar/dlq_router.go
index 647c022..6b13b32 100644
--- a/pulsar/dlq_router.go
+++ b/pulsar/dlq_router.go
@@ -22,31 +22,40 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsar/log"
 )
 
 type dlqRouter struct {
-	client           Client
-	producer         Producer
-	policy           *DLQPolicy
-	messageCh        chan ConsumerMessage
-	closeCh          chan interface{}
-	topicName        string
-	subscriptionName string
-	consumerName     string
-	log              log.Logger
+	client            Client
+	producer          Producer
+	policy            *DLQPolicy
+	messageCh         chan ConsumerMessage
+	closeCh           chan interface{}
+	topicName         string
+	subscriptionName  string
+	consumerName      string
+	backOffPolicyFunc func() backoff.Policy
+	log               log.Logger
 }
 
 func newDlqRouter(client Client, policy *DLQPolicy, topicName, subscriptionName, consumerName string,
-	logger log.Logger) (*dlqRouter, error) {
+	backOffPolicyFunc func() backoff.Policy, logger log.Logger) (*dlqRouter, error) {
+	var boFunc func() backoff.Policy
+	if backOffPolicyFunc != nil {
+		boFunc = backOffPolicyFunc
+	} else {
+		boFunc = backoff.NewDefaultBackoff
+	}
 	r := &dlqRouter{
-		client:           client,
-		policy:           policy,
-		topicName:        topicName,
-		subscriptionName: subscriptionName,
-		consumerName:     consumerName,
-		log:              logger,
+		client:            client,
+		policy:            policy,
+		topicName:         topicName,
+		subscriptionName:  subscriptionName,
+		consumerName:      consumerName,
+		backOffPolicyFunc: boFunc,
+		log:               logger,
 	}
 
 	if policy != nil {
@@ -155,7 +164,7 @@ func (r *dlqRouter) getProducer(schema Schema) Producer {
 	}
 
 	// Retry to create producer indefinitely
-	backoff := &internal.DefaultBackoff{}
+	bo := r.backOffPolicyFunc()
 	for {
 		opt := r.policy.ProducerOptions
 		opt.Topic = r.policy.DeadLetterTopic
@@ -174,7 +183,7 @@ func (r *dlqRouter) getProducer(schema Schema) Producer {
 
 		if err != nil {
 			r.log.WithError(err).Error("Failed to create DLQ producer")
-			time.Sleep(backoff.Next())
+			time.Sleep(bo.Next())
 			continue
 		} else {
 			r.producer = producer
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index c1f38f8..c246308 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -35,8 +35,6 @@ import (
 
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/apache/pulsar-client-go/pulsar/log"
-
-	ua "go.uber.org/atomic"
 )
 
 const (
@@ -53,6 +51,7 @@ type TLSOptions struct {
 	CipherSuites            []uint16
 	MinVersion              uint16
 	MaxVersion              uint16
+	TLSConfig               *tls.Config
 }
 
 var (
@@ -87,7 +86,7 @@ type Connection interface {
 	ID() string
 	GetMaxMessageSize() int32
 	Close()
-	WaitForClose() <-chan interface{}
+	WaitForClose() <-chan struct{}
 	IsProxied() bool
 }
 
@@ -108,7 +107,6 @@ type connectionState int32
 const (
 	connectionInit = iota
 	connectionReady
-	connectionClosing
 	connectionClosed
 )
 
@@ -118,8 +116,6 @@ func (s connectionState) String() string {
 		return "Initializing"
 	case connectionReady:
 		return "Ready"
-	case connectionClosing:
-		return "Closing"
 	case connectionClosed:
 		return "Closed"
 	default:
@@ -139,16 +135,19 @@ type incomingCmd struct {
 }
 
 type connection struct {
-	sync.Mutex
-	cond              *sync.Cond
 	started           int32
-	state             ua.Int32
 	connectionTimeout time.Duration
 	closeOnce         sync.Once
 
+	// mu protects the fields below against concurrency accesses.
+	mu               sync.RWMutex
+	state            atomic.Int32
+	cnx              net.Conn
+	listeners        map[uint64]ConnectionListener
+	consumerHandlers map[uint64]ConsumerHandler
+
 	logicalAddr  *url.URL
 	physicalAddr *url.URL
-	cnx          net.Conn
 
 	writeBufferLock sync.Mutex
 	writeBuffer     Buffer
@@ -162,18 +161,13 @@ type connection struct {
 	incomingRequestsWG sync.WaitGroup
 	incomingRequestsCh chan *request
 	incomingCmdCh      chan *incomingCmd
-	closeCh            chan interface{}
+	closeCh            chan struct{}
+	readyCh            chan struct{}
 	writeRequestsCh    chan Buffer
 
 	pendingLock sync.Mutex
 	pendingReqs map[uint64]*request
 
-	listenersLock sync.RWMutex
-	listeners     map[uint64]ConnectionListener
-
-	consumerHandlersLock sync.RWMutex
-	consumerHandlers     map[uint64]ConsumerHandler
-
 	tlsOptions *TLSOptions
 	auth       auth.Provider
 
@@ -210,7 +204,8 @@ func newConnection(opts connectionOptions) *connection {
 		tlsOptions:           opts.tls,
 		auth:                 opts.auth,
 
-		closeCh:            make(chan interface{}),
+		closeCh:            make(chan struct{}),
+		readyCh:            make(chan struct{}),
 		incomingRequestsCh: make(chan *request, 10),
 		incomingCmdCh:      make(chan *incomingCmd, 10),
 
@@ -224,9 +219,8 @@ func newConnection(opts connectionOptions) *connection {
 		consumerHandlers: make(map[uint64]ConsumerHandler),
 		metrics:          opts.metrics,
 	}
-	cnx.setState(connectionInit)
+	cnx.state.Store(int32(connectionInit))
 	cnx.reader = newConnectionReader(cnx)
-	cnx.cond = sync.NewCond(cnx)
 	return cnx
 }
 
@@ -289,11 +283,11 @@ func (c *connection) connect() bool {
 		return false
 	}
 
-	c.Lock()
+	c.mu.Lock()
 	c.cnx = cnx
 	c.log = c.log.SubLogger(log.Fields{"local_addr": c.cnx.LocalAddr()})
 	c.log.Info("TCP connection established")
-	c.Unlock()
+	c.mu.Unlock()
 
 	return true
 }
@@ -347,7 +341,8 @@ func (c *connection) doHandshake() bool {
 	}
 	c.log.Info("Connection is ready")
 	c.setLastDataReceived(time.Now())
-	c.changeState(connectionReady)
+	c.setStateReady()
+	close(c.readyCh) // broadcast the readiness of the connection.
 	return true
 }
 
@@ -356,22 +351,13 @@ func (c *connection) IsProxied() bool {
 }
 
 func (c *connection) waitUntilReady() error {
-	// If we are going to call cond.Wait() at all, then we must call it _before_ we call cond.Broadcast().
-	// The lock is held here to prevent changeState() from calling cond.Broadcast() in the time between
-	// the state check and call to cond.Wait().
-	c.Lock()
-	defer c.Unlock()
-
-	for c.getState() != connectionReady {
-		c.log.Debugf("Wait until connection is ready state=%s", c.getState().String())
-		if c.getState() == connectionClosed {
-			return errors.New("connection error")
-		}
-		// wait for a new connection state change
-		c.cond.Wait()
+	select {
+	case <-c.readyCh:
+		return nil
+	case <-c.closeCh:
+		// Connection has been closed while waiting for the readiness.
+		return errors.New("connection error")
 	}
-
-	return nil
 }
 
 func (c *connection) failLeftRequestsWhenClose() {
@@ -492,8 +478,8 @@ func (c *connection) WriteData(data Buffer) {
 			// 1. blocked, in which case we need to wait until we have space
 			// 2. the connection is already closed, then we need to bail out
 			c.log.Debug("Couldn't write on connection channel immediately")
-			state := c.getState()
-			if state != connectionReady {
+
+			if c.getState() != connectionReady {
 				c.log.Debug("Connection was already closed")
 				return
 			}
@@ -648,8 +634,7 @@ func (c *connection) SendRequest(requestID uint64, req *pb.BaseCommand,
 	c.incomingRequestsWG.Add(1)
 	defer c.incomingRequestsWG.Done()
 
-	state := c.getState()
-	if state == connectionClosed || state == connectionClosing {
+	if c.getState() == connectionClosed {
 		callback(req, ErrConnectionClosed)
 
 	} else {
@@ -670,8 +655,7 @@ func (c *connection) SendRequestNoWait(req *pb.BaseCommand) error {
 	c.incomingRequestsWG.Add(1)
 	defer c.incomingRequestsWG.Done()
 
-	state := c.getState()
-	if state == connectionClosed || state == connectionClosing {
+	if c.getState() == connectionClosed {
 		return ErrConnectionClosed
 	}
 
@@ -751,9 +735,9 @@ func (c *connection) handleAckResponse(ackResponse *pb.CommandAckResponse) {
 func (c *connection) handleSendReceipt(response *pb.CommandSendReceipt) {
 	producerID := response.GetProducerId()
 
-	c.listenersLock.RLock()
+	c.mu.RLock()
 	producer, ok := c.listeners[producerID]
-	c.listenersLock.RUnlock()
+	c.mu.RUnlock()
 
 	if ok {
 		producer.ReceivedSendReceipt(response)
@@ -893,12 +877,13 @@ func (c *connection) handleSendError(sendError *pb.CommandSendError) {
 }
 
 func (c *connection) deletePendingProducers(producerID uint64) (ConnectionListener, bool) {
-	c.listenersLock.Lock()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	producer, ok := c.listeners[producerID]
 	if ok {
 		delete(c.listeners, producerID)
 	}
-	c.listenersLock.Unlock()
 
 	return producer, ok
 }
@@ -960,9 +945,9 @@ func (c *connection) handleTopicMigrated(commandTopicMigrated *pb.CommandTopicMi
 		return
 	}
 	if commandTopicMigrated.GetResourceType() == pb.CommandTopicMigrated_Producer {
-		c.listenersLock.RLock()
+		c.mu.RLock()
 		producer, ok := c.listeners[resourceID]
-		c.listenersLock.RUnlock()
+		c.mu.RUnlock()
 		if ok {
 			producer.SetRedirectedClusterURI(migratedBrokerServiceURL)
 			c.log.Infof("producerID:{%d} migrated to RedirectedClusterURI:{%s}",
@@ -984,73 +969,55 @@ func (c *connection) handleTopicMigrated(commandTopicMigrated *pb.CommandTopicMi
 }
 
 func (c *connection) RegisterListener(id uint64, listener ConnectionListener) error {
-	// do not add if connection is closed
-	if c.closed() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.getState() == connectionClosed {
 		c.log.Warnf("Connection closed unable register listener id=%+v", id)
 		return errUnableRegisterListener
 	}
 
-	c.listenersLock.Lock()
-	defer c.listenersLock.Unlock()
-
 	c.listeners[id] = listener
 	return nil
 }
 
 func (c *connection) UnregisterListener(id uint64) {
-	c.listenersLock.Lock()
-	defer c.listenersLock.Unlock()
+	c.mu.Lock()
+	defer c.mu.Unlock()
 
 	delete(c.listeners, id)
 }
 
 func (c *connection) ResetLastActive() {
-	c.Lock()
-	defer c.Unlock()
+	c.mu.Lock()
+	defer c.mu.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
+	return len(c.listeners) == 0 &&
+		len(c.consumerHandlers) == 0 &&
+		len(c.incomingRequestsCh) == 0 &&
+		len(c.writeRequestsCh) == 0
 }
 
 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.pendingLock.Lock()
+	sizePendingReqs := len(c.pendingReqs)
+	c.pendingLock.Unlock()
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if sizePendingReqs != 0 || !c.isIdle() {
 		c.lastActive = time.Now()
 	}
+
 	return time.Since(c.lastActive) > maxIdleTime
 }
 
-func (c *connection) WaitForClose() <-chan interface{} {
+func (c *connection) WaitForClose() <-chan struct{} {
 	return c.closeCh
 }
 
@@ -1059,10 +1026,7 @@ func (c *connection) WaitForClose() <-chan interface{} {
 // This also triggers callbacks to the ConnectionClosed listeners.
 func (c *connection) Close() {
 	c.closeOnce.Do(func() {
-		c.Lock()
-		cnx := c.cnx
-		c.Unlock()
-		c.changeState(connectionClosed)
+		listeners, consumerHandlers, cnx := c.closeAndEmptyObservers()
 
 		if cnx != nil {
 			_ = cnx.Close()
@@ -1070,22 +1034,6 @@ func (c *connection) Close() {
 
 		close(c.closeCh)
 
-		listeners := make(map[uint64]ConnectionListener)
-		c.listenersLock.Lock()
-		for id, listener := range c.listeners {
-			listeners[id] = listener
-			delete(c.listeners, id)
-		}
-		c.listenersLock.Unlock()
-
-		consumerHandlers := make(map[uint64]ConsumerHandler)
-		c.consumerHandlersLock.Lock()
-		for id, handler := range c.consumerHandlers {
-			consumerHandlers[id] = handler
-			delete(c.consumerHandlers, id)
-		}
-		c.consumerHandlersLock.Unlock()
-
 		// notify producers connection closed
 		for _, listener := range listeners {
 			listener.ConnectionClosed(nil)
@@ -1100,22 +1048,35 @@ func (c *connection) Close() {
 	})
 }
 
-func (c *connection) changeState(state connectionState) {
-	// The lock is held here because we need setState() and cond.Broadcast() to be
-	// an atomic operation from the point of view of waitUntilReady().
-	c.Lock()
-	defer c.Unlock()
+func (c *connection) closeAndEmptyObservers() ([]ConnectionListener, []ConsumerHandler, net.Conn) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.setStateClosed()
 
-	c.setState(state)
-	c.cond.Broadcast()
+	listeners := make([]ConnectionListener, 0, len(c.listeners))
+	for _, listener := range c.listeners {
+		listeners = append(listeners, listener)
+	}
+
+	handlers := make([]ConsumerHandler, 0, len(c.consumerHandlers))
+	for _, handler := range c.consumerHandlers {
+		handlers = append(handlers, handler)
+	}
+
+	return listeners, handlers, c.cnx
 }
 
 func (c *connection) getState() connectionState {
 	return connectionState(c.state.Load())
 }
 
-func (c *connection) setState(state connectionState) {
-	c.state.Store(int32(state))
+func (c *connection) setStateReady() {
+	c.state.CompareAndSwap(int32(connectionInit), int32(connectionReady))
+}
+
+func (c *connection) setStateClosed() {
+	c.state.Store(int32(connectionClosed))
 }
 
 func (c *connection) closed() bool {
@@ -1123,6 +1084,10 @@ func (c *connection) closed() bool {
 }
 
 func (c *connection) getTLSConfig() (*tls.Config, error) {
+	if c.tlsOptions.TLSConfig != nil {
+		return c.tlsOptions.TLSConfig, nil
+	}
+
 	tlsConfig := &tls.Config{
 		InsecureSkipVerify: c.tlsOptions.AllowInsecureConnection,
 		CipherSuites:       c.tlsOptions.CipherSuites,
@@ -1173,32 +1138,37 @@ func (c *connection) getTLSConfig() (*tls.Config, error) {
 }
 
 func (c *connection) AddConsumeHandler(id uint64, handler ConsumerHandler) error {
-	// do not add if connection is closed
-	if c.closed() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.getState() == connectionClosed {
 		c.log.Warnf("Closed connection unable add consumer with id=%+v", id)
 		return errUnableAddConsumeHandler
 	}
 
-	c.consumerHandlersLock.Lock()
-	defer c.consumerHandlersLock.Unlock()
 	c.consumerHandlers[id] = handler
 	return nil
 }
 
 func (c *connection) DeleteConsumeHandler(id uint64) {
-	c.consumerHandlersLock.Lock()
-	defer c.consumerHandlersLock.Unlock()
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
 	delete(c.consumerHandlers, id)
 }
 
 func (c *connection) consumerHandler(id uint64) (ConsumerHandler, bool) {
-	c.consumerHandlersLock.RLock()
-	defer c.consumerHandlersLock.RUnlock()
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	h, ok := c.consumerHandlers[id]
 	return h, ok
 }
 
 func (c *connection) ID() string {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	return fmt.Sprintf("%s -> %s", c.cnx.LocalAddr(), c.cnx.RemoteAddr())
 }
 
diff --git a/pulsar/internal/http_client.go b/pulsar/internal/http_client.go
index e68bd17..f1420ae 100644
--- a/pulsar/internal/http_client.go
+++ b/pulsar/internal/http_client.go
@@ -29,6 +29,8 @@ import (
 	"path"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsar/auth"
 
 	"github.com/apache/pulsar-client-go/pulsar/log"
@@ -148,12 +150,12 @@ func (c *httpClient) Get(endpoint string, obj interface{}, params map[string]str
 	if _, ok := err.(*url.Error); ok {
 		// We can retry this kind of requests over a connection error because they're
 		// not specific to a particular broker.
-		backoff := DefaultBackoff{100 * time.Millisecond}
+		bo := backoff.NewDefaultBackoffWithInitialBackOff(100 * time.Millisecond)
 		startTime := time.Now()
 		var retryTime time.Duration
 
 		for time.Since(startTime) < c.requestTimeout {
-			retryTime = backoff.Next()
+			retryTime = bo.Next()
 			c.log.Debugf("Retrying httpRequest in {%v} with timeout in {%v}", retryTime, c.requestTimeout)
 			time.Sleep(retryTime)
 			_, err = c.GetWithQueryParams(endpoint, obj, params, true)
@@ -337,29 +339,33 @@ func responseError(resp *http.Response) error {
 func getDefaultTransport(tlsConfig *TLSOptions) (http.RoundTripper, error) {
 	transport := http.DefaultTransport.(*http.Transport)
 	if tlsConfig != nil {
-		cfg := &tls.Config{
-			InsecureSkipVerify: tlsConfig.AllowInsecureConnection,
-			CipherSuites:       tlsConfig.CipherSuites,
-			MinVersion:         tlsConfig.MinVersion,
-			MaxVersion:         tlsConfig.MaxVersion,
-		}
-		if len(tlsConfig.TrustCertsFilePath) > 0 {
-			rootCA, err := os.ReadFile(tlsConfig.TrustCertsFilePath)
-			if err != nil {
-				return nil, err
+		if tlsConfig.TLSConfig != nil {
+			transport.TLSClientConfig = tlsConfig.TLSConfig
+		} else {
+			cfg := &tls.Config{
+				InsecureSkipVerify: tlsConfig.AllowInsecureConnection,
+				CipherSuites:       tlsConfig.CipherSuites,
+				MinVersion:         tlsConfig.MinVersion,
+				MaxVersion:         tlsConfig.MaxVersion,
+			}
+			if len(tlsConfig.TrustCertsFilePath) > 0 {
+				rootCA, err := os.ReadFile(tlsConfig.TrustCertsFilePath)
+				if err != nil {
+					return nil, err
+				}
+				cfg.RootCAs = x509.NewCertPool()
+				cfg.RootCAs.AppendCertsFromPEM(rootCA)
 			}
-			cfg.RootCAs = x509.NewCertPool()
-			cfg.RootCAs.AppendCertsFromPEM(rootCA)
-		}
 
-		if tlsConfig.CertFile != "" && tlsConfig.KeyFile != "" {
-			cert, err := tls.LoadX509KeyPair(tlsConfig.CertFile, tlsConfig.KeyFile)
-			if err != nil {
-				return nil, errors.New(err.Error())
+			if tlsConfig.CertFile != "" && tlsConfig.KeyFile != "" {
+				cert, err := tls.LoadX509KeyPair(tlsConfig.CertFile, tlsConfig.KeyFile)
+				if err != nil {
+					return nil, errors.New(err.Error())
+				}
+				cfg.Certificates = []tls.Certificate{cert}
 			}
-			cfg.Certificates = []tls.Certificate{cert}
+			transport.TLSClientConfig = cfg
 		}
-		transport.TLSClientConfig = cfg
 	}
 	transport.MaxIdleConnsPerHost = 10
 	return transport, nil
diff --git a/pulsar/internal/memory_limit_controller.go b/pulsar/internal/memory_limit_controller.go
index 5ead9f5..345ccde 100644
--- a/pulsar/internal/memory_limit_controller.go
+++ b/pulsar/internal/memory_limit_controller.go
@@ -35,9 +35,9 @@ type MemoryLimitController interface {
 }
 
 type memoryLimitController struct {
+	currentUsage int64
 	limit        int64
 	chCond       *chCond
-	currentUsage int64
 
 	triggers []*thresholdTrigger
 	// valid range is (0, 1.0)
diff --git a/pulsar/internal/rpc_client.go b/pulsar/internal/rpc_client.go
index d2e3895..0f99311 100644
--- a/pulsar/internal/rpc_client.go
+++ b/pulsar/internal/rpc_client.go
@@ -25,6 +25,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsar/auth"
 	"github.com/apache/pulsar-client-go/pulsar/log"
 
@@ -115,7 +117,7 @@ func (c *rpcClient) requestToHost(serviceNameResolver *ServiceNameResolver,
 	var host *url.URL
 	var rpcResult *RPCResult
 	startTime := time.Now()
-	backoff := DefaultBackoff{100 * time.Millisecond}
+	bo := backoff.NewDefaultBackoffWithInitialBackOff(100 * time.Millisecond)
 	// we can retry these requests because this kind of request is
 	// not specific to any particular broker
 	for time.Since(startTime) < c.requestTimeout {
@@ -130,7 +132,7 @@ func (c *rpcClient) requestToHost(serviceNameResolver *ServiceNameResolver,
 			break
 		}
 
-		retryTime := backoff.Next()
+		retryTime := bo.Next()
 		c.log.Debugf("Retrying request in {%v} with timeout in {%v}", retryTime, c.requestTimeout)
 		time.Sleep(retryTime)
 	}
diff --git a/pulsar/producer.go b/pulsar/producer.go
index 0ae51bd..997f9c0 100644
--- a/pulsar/producer.go
+++ b/pulsar/producer.go
@@ -21,7 +21,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
 )
 
 type HashingScheme int
@@ -171,9 +171,9 @@ type ProducerOptions struct {
 	// MaxReconnectToBroker specifies the maximum retry number of reconnectToBroker. (default: ultimate)
 	MaxReconnectToBroker *uint
 
-	// BackoffPolicy parameterize the following options in the reconnection logic to
+	// BackOffPolicyFunc parameterize the following options in the reconnection logic to
 	// allow users to customize the reconnection logic (minBackoff, maxBackoff and jitterPercentage)
-	BackoffPolicy internal.BackoffPolicy
+	BackOffPolicyFunc func() backoff.Policy
 
 	// BatcherBuilderType sets the batch builder type (default DefaultBatchBuilder)
 	// This will be used to create batch container when batching is enabled.
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 1677c57..f578ee4 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -28,6 +28,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsar/internal/compression"
 	internalcrypto "github.com/apache/pulsar-client-go/pulsar/internal/crypto"
 
@@ -87,6 +89,8 @@ func init() {
 }
 
 type partitionProducer struct {
+	lastSequenceID int64
+
 	state  uAtomic.Int32
 	client *client
 	topic  string
@@ -110,7 +114,6 @@ type partitionProducer struct {
 	connectClosedCh      chan *connectionClosed
 	publishSemaphore     internal.Semaphore
 	pendingQueue         internal.BlockingQueue
-	lastSequenceID       int64
 	schemaInfo           *SchemaInfo
 	partitionIdx         int32
 	metrics              *internal.LeveledMetrics
@@ -120,6 +123,7 @@ type partitionProducer struct {
 	redirectedClusterURI string
 	ctx                  context.Context
 	cancelFunc           context.CancelFunc
+	backOffPolicyFunc    func() backoff.Policy
 }
 
 type schemaCache struct {
@@ -146,6 +150,14 @@ func (s *schemaCache) Get(schema *SchemaInfo) (schemaVersion []byte) {
 func newPartitionProducer(client *client, topic string, options *ProducerOptions, partitionIdx int,
 	metrics *internal.LeveledMetrics) (
 	*partitionProducer, error) {
+
+	var boFunc func() backoff.Policy
+	if options.BackOffPolicyFunc != nil {
+		boFunc = options.BackOffPolicyFunc
+	} else {
+		boFunc = backoff.NewDefaultBackoff
+	}
+
 	var batchingMaxPublishDelay time.Duration
 	if options.BatchingMaxPublishDelay != 0 {
 		batchingMaxPublishDelay = options.BatchingMaxPublishDelay
@@ -171,19 +183,20 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 		producerID:       client.rpcClient.NewProducerID(),
 		dataChan:         make(chan *sendRequest, maxPendingMessages),
 		cmdChan:          make(chan interface{}, 10),
-		connectClosedCh:  make(chan *connectionClosed, 10),
+		connectClosedCh:  make(chan *connectionClosed, 1),
 		batchFlushTicker: time.NewTicker(batchingMaxPublishDelay),
 		compressionProvider: internal.GetCompressionProvider(pb.CompressionType(options.CompressionType),
 			compression.Level(options.CompressionLevel)),
-		publishSemaphore: internal.NewSemaphore(int32(maxPendingMessages)),
-		pendingQueue:     internal.NewBlockingQueue(maxPendingMessages),
-		lastSequenceID:   -1,
-		partitionIdx:     int32(partitionIdx),
-		metrics:          metrics,
-		epoch:            0,
-		schemaCache:      newSchemaCache(),
-		ctx:              ctx,
-		cancelFunc:       cancelFunc,
+		publishSemaphore:  internal.NewSemaphore(int32(maxPendingMessages)),
+		pendingQueue:      internal.NewBlockingQueue(maxPendingMessages),
+		lastSequenceID:    -1,
+		partitionIdx:      int32(partitionIdx),
+		metrics:           metrics,
+		epoch:             0,
+		schemaCache:       newSchemaCache(),
+		ctx:               ctx,
+		cancelFunc:        cancelFunc,
+		backOffPolicyFunc: boFunc,
 	}
 	if p.options.DisableBatching {
 		p.batchFlushTicker.Stop()
@@ -413,8 +426,12 @@ func (p *partitionProducer) ConnectionClosed(closeProducer *pb.CommandCloseProdu
 		assignedBrokerURL = p.client.selectServiceURL(
 			closeProducer.GetAssignedBrokerServiceUrl(), closeProducer.GetAssignedBrokerServiceUrlTls())
 	}
-	p.connectClosedCh <- &connectionClosed{
-		assignedBrokerURL: assignedBrokerURL,
+
+	select {
+	case p.connectClosedCh <- &connectionClosed{assignedBrokerURL: assignedBrokerURL}:
+	default:
+		// Reconnect has already been requested so we do not block the
+		// connection callback.
 	}
 }
 
@@ -453,17 +470,17 @@ func (p *partitionProducer) getOrCreateSchema(schemaInfo *SchemaInfo) (schemaVer
 }
 
 func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed) {
-	var maxRetry int
+	var (
+		maxRetry           int
+		delayReconnectTime time.Duration
+	)
 	if p.options.MaxReconnectToBroker == nil {
 		maxRetry = -1
 	} else {
 		maxRetry = int(*p.options.MaxReconnectToBroker)
 	}
 
-	var (
-		delayReconnectTime time.Duration
-		defaultBackoff     = internal.DefaultBackoff{}
-	)
+	bo := p.backOffPolicyFunc()
 
 	for maxRetry != 0 {
 		select {
@@ -484,10 +501,8 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
 			delayReconnectTime = 0
 			assignedBrokerURL = connectionClosed.assignedBrokerURL
 			connectionClosed = nil // Only attempt once
-		} else if p.options.BackoffPolicy == nil {
-			delayReconnectTime = defaultBackoff.Next()
 		} else {
-			delayReconnectTime = p.options.BackoffPolicy.Next()
+			delayReconnectTime = bo.Next()
 		}
 
 		p.log.WithFields(log.Fields{
@@ -508,6 +523,7 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
 		if err == nil {
 			// Successfully reconnected
 			p.log.WithField("cnx", p._getConn().ID()).Info("Reconnected producer to broker")
+			bo.Reset()
 			return
 		}
 		p.log.WithError(err).Error("Failed to create producer at reconnect")
@@ -541,7 +557,7 @@ func (p *partitionProducer) reconnectToBroker(connectionClosed *connectionClosed
 			maxRetry--
 		}
 		p.metrics.ProducersReconnectFailure.Inc()
-		if maxRetry == 0 || defaultBackoff.IsMaxBackoffReached() {
+		if maxRetry == 0 || bo.IsMaxBackoffReached() {
 			p.metrics.ProducersReconnectMaxRetry.Inc()
 		}
 	}
diff --git a/pulsar/reader.go b/pulsar/reader.go
index 4daa889..98bde4e 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -21,7 +21,7 @@ import (
 	"context"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
 )
 
 // ReaderMessage packages Reader and Message as a struct to use.
@@ -89,9 +89,9 @@ type ReaderOptions struct {
 	// Schema represents the schema implementation.
 	Schema Schema
 
-	// BackoffPolicy parameterize the following options in the reconnection logic to
+	// BackoffPolicyFunc parameterize the following options in the reconnection logic to
 	// allow users to customize the reconnection logic (minBackoff, maxBackoff and jitterPercentage)
-	BackoffPolicy internal.BackoffPolicy
+	BackoffPolicyFunc func() backoff.Policy
 
 	// MaxPendingChunkedMessage sets the maximum pending chunked messages. (default: 100)
 	MaxPendingChunkedMessage int
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 7cbae05..f76255e 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -112,7 +112,7 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		ReplicateSubscriptionState:  false,
 		Decryption:                  options.Decryption,
 		Schema:                      options.Schema,
-		BackoffPolicy:               options.BackoffPolicy,
+		BackOffPolicyFunc:           options.BackoffPolicyFunc,
 		MaxPendingChunkedMessage:    options.MaxPendingChunkedMessage,
 		ExpireTimeOfIncompleteChunk: options.ExpireTimeOfIncompleteChunk,
 		AutoAckIncompleteChunk:      options.AutoAckIncompleteChunk,
@@ -128,12 +128,13 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 	}
 
 	// Provide dummy dlq router with not dlq policy
-	dlq, err := newDlqRouter(client, nil, options.Topic, options.SubscriptionName, options.Name, client.log)
+	dlq, err := newDlqRouter(client, nil, options.Topic, options.SubscriptionName, options.Name,
+		options.BackoffPolicyFunc, client.log)
 	if err != nil {
 		return nil, err
 	}
 	// Provide dummy rlq router with not dlq policy
-	rlq, err := newRetryRouter(client, nil, false, client.log)
+	rlq, err := newRetryRouter(client, nil, false, options.BackoffPolicyFunc, client.log)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pulsar/retry_router.go b/pulsar/retry_router.go
index 75792ad..c8aa0b9 100644
--- a/pulsar/retry_router.go
+++ b/pulsar/retry_router.go
@@ -21,7 +21,8 @@ import (
 	"context"
 	"time"
 
-	"github.com/apache/pulsar-client-go/pulsar/internal"
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsar/log"
 )
 
@@ -44,19 +45,28 @@ type RetryMessage struct {
 }
 
 type retryRouter struct {
-	client    Client
-	producer  Producer
-	policy    *DLQPolicy
-	messageCh chan RetryMessage
-	closeCh   chan interface{}
-	log       log.Logger
+	client            Client
+	producer          Producer
+	policy            *DLQPolicy
+	messageCh         chan RetryMessage
+	closeCh           chan interface{}
+	backOffPolicyFunc func() backoff.Policy
+	log               log.Logger
 }
 
-func newRetryRouter(client Client, policy *DLQPolicy, retryEnabled bool, logger log.Logger) (*retryRouter, error) {
+func newRetryRouter(client Client, policy *DLQPolicy, retryEnabled bool, backOffPolicyFunc func() backoff.Policy,
+	logger log.Logger) (*retryRouter, error) {
+	var boFunc func() backoff.Policy
+	if backOffPolicyFunc != nil {
+		boFunc = backOffPolicyFunc
+	} else {
+		boFunc = backoff.NewDefaultBackoff
+	}
 	r := &retryRouter{
-		client: client,
-		policy: policy,
-		log:    logger,
+		client:            client,
+		policy:            policy,
+		backOffPolicyFunc: boFunc,
+		log:               logger,
 	}
 
 	if policy != nil && retryEnabled {
@@ -124,7 +134,7 @@ func (r *retryRouter) getProducer() Producer {
 	}
 
 	// Retry to create producer indefinitely
-	backoff := &internal.DefaultBackoff{}
+	bo := r.backOffPolicyFunc()
 	for {
 		opt := r.policy.ProducerOptions
 		opt.Topic = r.policy.RetryLetterTopic
@@ -138,7 +148,7 @@ func (r *retryRouter) getProducer() Producer {
 
 		if err != nil {
 			r.log.WithError(err).Error("Failed to create RLQ producer")
-			time.Sleep(backoff.Next())
+			time.Sleep(bo.Next())
 			continue
 		} else {
 			r.producer = producer
diff --git a/pulsar/transaction_coordinator_client.go b/pulsar/transaction_coordinator_client.go
index 1449d69..afde542 100644
--- a/pulsar/transaction_coordinator_client.go
+++ b/pulsar/transaction_coordinator_client.go
@@ -24,6 +24,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/apache/pulsar-client-go/pulsar/log"
@@ -143,7 +145,7 @@ func (t *transactionHandler) runEventsLoop() {
 
 func (t *transactionHandler) reconnectToBroker() {
 	var delayReconnectTime time.Duration
-	var defaultBackoff = internal.DefaultBackoff{}
+	var defaultBackoff = backoff.DefaultBackoff{}
 
 	for {
 		if t.getState() == txnHandlerClosed {
diff --git a/pulsaradmin/pkg/admin/schema.go b/pulsaradmin/pkg/admin/schema.go
index 7190bd9..7ed344b 100644
--- a/pulsaradmin/pkg/admin/schema.go
+++ b/pulsaradmin/pkg/admin/schema.go
@@ -35,6 +35,9 @@ type Schema interface {
 	// GetSchemaInfoByVersion retrieves the schema of a topic at a given <tt>version</tt>
 	GetSchemaInfoByVersion(topic string, version int64) (*utils.SchemaInfo, error)
 
+	// GetAllSchemas retrieves all schemas of a topic
+	GetAllSchemas(topic string) ([]*utils.SchemaInfoWithVersion, error)
+
 	// DeleteSchema deletes the schema associated with a given <tt>topic</tt>
 	DeleteSchema(topic string) error
 
@@ -130,6 +133,24 @@ func (s *schemas) GetSchemaInfoByVersion(topic string, version int64) (*utils.Sc
 	return info, nil
 }
 
+func (s *schemas) GetAllSchemas(topic string) ([]*utils.SchemaInfoWithVersion, error) {
+	topicName, err := utils.GetTopicName(topic)
+	if err != nil {
+		return nil, err
+	}
+	var response utils.GetAllSchemasResponse
+	endpoint := s.pulsar.endpoint(s.basePath, topicName.GetTenant(), topicName.GetNamespace(),
+		topicName.GetLocalName(), "schemas")
+
+	err = s.pulsar.Client.Get(endpoint, &response)
+	if err != nil {
+		return nil, err
+	}
+
+	infos := utils.ConvertGetAllSchemasResponseToSchemaInfosWithVersion(topicName, response)
+	return infos, nil
+}
+
 func (s *schemas) DeleteSchema(topic string) error {
 	return s.delete(topic, false)
 }
diff --git a/pulsaradmin/pkg/utils/schema_util.go b/pulsaradmin/pkg/utils/schema_util.go
index 3b83669..671627f 100644
--- a/pulsaradmin/pkg/utils/schema_util.go
+++ b/pulsaradmin/pkg/utils/schema_util.go
@@ -44,6 +44,10 @@ type GetSchemaResponse struct {
 	Properties map[string]string `json:"properties"`
 }
 
+type GetAllSchemasResponse struct {
+	Schemas []GetSchemaResponse `json:"getSchemaResponses"`
+}
+
 type IsCompatibility struct {
 	IsCompatibility             bool                        `json:"compatibility"`
 	SchemaCompatibilityStrategy SchemaCompatibilityStrategy `json:"schemaCompatibilityStrategy"`
@@ -90,3 +94,16 @@ func ConvertGetSchemaResponseToSchemaInfoWithVersion(tn *TopicName, response Get
 	info.Version = response.Version
 	return info
 }
+
+func ConvertGetAllSchemasResponseToSchemaInfosWithVersion(
+	tn *TopicName,
+	response GetAllSchemasResponse,
+) []*SchemaInfoWithVersion {
+	infos := make([]*SchemaInfoWithVersion, len(response.Schemas))
+
+	for i, schema := range response.Schemas {
+		infos[i] = ConvertGetSchemaResponseToSchemaInfoWithVersion(tn, schema)
+	}
+
+	return infos
+}


Test patch:
diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go
index 0a794f6..1fe5a56 100644
--- a/pulsar/ack_grouping_tracker_test.go
+++ b/pulsar/ack_grouping_tracker_test.go
@@ -59,9 +59,9 @@ func TestNoCacheTracker(t *testing.T) {
 }
 
 type mockAcker struct {
-	sync.Mutex
-	ledgerIDs          []int64
 	cumulativeLedgerID int64
+	sync.Mutex
+	ledgerIDs []int64
 }
 
 func (a *mockAcker) ack(ids []*pb.MessageIdData) {
@@ -217,6 +217,52 @@ func TestDuplicateAfterClose(t *testing.T) {
 	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 1}))
 }
 
+func TestCloseFlushWithoutTimer(t *testing.T) {
+	var acker mockAcker
+	tracker := newAckGroupingTracker(
+		&AckGroupingOptions{MaxSize: 3, MaxTime: 0},
+		nil,
+		func(id MessageID) { acker.ackCumulative(id) },
+		func(ids []*pb.MessageIdData) { acker.ack(ids) },
+	)
+
+	// case 1: message will not be acked because the cache is not full
+	tracker.add(&messageID{ledgerID: 1})
+	tracker.add(&messageID{ledgerID: 2})
+	assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(1)}))
+	assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(2)}))
+	assert.Equal(t, 0, len(acker.getLedgerIDs()))
+
+	// case 2: tracker close so that all messages are flushed and acked
+	tracker.close()
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 1}))
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 2}))
+	assert.Equal(t, []int64{1, 2}, acker.getLedgerIDs())
+}
+
+func TestCloseFlushWithTimer(t *testing.T) {
+	var acker mockAcker
+	tracker := newAckGroupingTracker(
+		&AckGroupingOptions{MaxSize: 1000, MaxTime: 10 * time.Second},
+		nil,
+		func(id MessageID) { acker.ackCumulative(id) },
+		func(ids []*pb.MessageIdData) { acker.ack(ids) },
+	)
+
+	// case 1: messages are not acked because the cache is not full
+	tracker.add(&messageID{ledgerID: 1})
+	tracker.add(&messageID{ledgerID: 2})
+	assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(1)}))
+	assert.True(t, tracker.isDuplicate(&messageID{ledgerID: int64(2)}))
+	assert.Equal(t, 0, len(acker.getLedgerIDs()))
+
+	// case 2: tracker close so that all messages are flushed and acked
+	tracker.close()
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 1}))
+	assert.False(t, tracker.isDuplicate(&messageID{ledgerID: 2}))
+	assert.Equal(t, []int64{1, 2}, acker.getLedgerIDs())
+}
+
 func TestTrackerPendingAcks(t *testing.T) {
 	m := make(map[uint64][]int64)
 	tracker := newAckGroupingTracker(&AckGroupingOptions{MaxSize: 3, MaxTime: 0}, nil, nil,
diff --git a/pulsar/internal/backoff_test.go b/pulsar/backoff/backoff_test.go
similarity index 96%
rename from pulsar/internal/backoff_test.go
rename to pulsar/backoff/backoff_test.go
index e05ea29..fc0a492 100644
--- a/pulsar/internal/backoff_test.go
+++ b/pulsar/backoff/backoff_test.go
@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-package internal
+package backoff
 
 import (
 	"testing"
@@ -58,4 +58,6 @@ func TestBackoff_NextMaxValue(t *testing.T) {
 	assert.Equal(t, true, backoff.IsMaxBackoffReached())
 	// max value is 60 seconds + 20% jitter = 72 seconds
 	assert.LessOrEqual(t, int64(cappedDelay), int64(72*time.Second))
+	backoff.Reset()
+	assert.Equal(t, false, backoff.IsMaxBackoffReached())
 }
diff --git a/pulsar/blue_green_migration_test.go b/pulsar/blue_green_migration_test.go
index 91667e8..672ef34 100644
--- a/pulsar/blue_green_migration_test.go
+++ b/pulsar/blue_green_migration_test.go
@@ -54,8 +54,8 @@ func (suite *BlueGreenMigrationTestSuite) TestTopicMigration() {
 	for _, scenario := range []topicUnloadTestCase{
 
 		{
-			testCaseName: "proxyConnection",
-			blueAdminURL: "http://localhost:8080",
+			testCaseName:  "proxyConnection",
+			blueAdminURL:  "http://localhost:8080",
 			blueClientUrl: "pulsar://localhost:6650",
 			greenAdminURL: "http://localhost:8081",
 			migrationBody: `
@@ -83,17 +83,17 @@ func testTopicMigrate(
 	migrationBody string) {
 	runtime.GOMAXPROCS(1)
 	const (
-		cluster = "cluster-a"
+		cluster   = "cluster-a"
 		tenant    = utils.PUBLICTENANT
 		namespace = utils.DEFAULTNAMESPACE
 
-		blueBroker1URL = "pulsar://broker-1:6650"
-		blueBroker2URL = "pulsar://broker-2:6650"
+		blueBroker1URL  = "pulsar://broker-1:6650"
+		blueBroker2URL  = "pulsar://broker-2:6650"
 		greenBroker1URL = "pulsar://green-broker-1:6650"
 		greenBroker2URL = "pulsar://green-broker-2:6650"
 
-		blueBroker1LookupURL = "broker-1:8080"
-		blueBroker2LookupURL = "broker-2:8080"
+		blueBroker1LookupURL  = "broker-1:8080"
+		blueBroker2LookupURL  = "broker-2:8080"
 		greenBroker1LookupURL = "green-broker-1:8080"
 		greenBroker2LookupURL = "green-broker-2:8080"
 	)
@@ -234,7 +234,6 @@ func testTopicMigrate(
 	req.NoError(err)
 	req.NotEmpty(bundleRange)
 
-
 	unloadURL := fmt.Sprintf(
 		"/admin/v2/namespaces/%s/%s/%s/unload?destinationBroker=%s",
 		tenant, namespace, bundleRange, dstTopicBrokerLookupURL)
diff --git a/pulsar/consumer_regex_test.go b/pulsar/consumer_regex_test.go
index 28ba4f7..e1e2ca2 100644
--- a/pulsar/consumer_regex_test.go
+++ b/pulsar/consumer_regex_test.go
@@ -159,8 +159,9 @@ func runRegexConsumerDiscoverPatternAll(t *testing.T, c Client, namespace string
 		Name:                "regex-consumer",
 	}
 
-	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", "regex-consumer", log.DefaultNopLogger())
-	rlq, _ := newRetryRouter(c.(*client), nil, false, log.DefaultNopLogger())
+	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", "regex-consumer",
+		nil, log.DefaultNopLogger())
+	rlq, _ := newRetryRouter(c.(*client), nil, false, nil, log.DefaultNopLogger())
 	consumer, err := newRegexConsumer(c.(*client), opts, tn, pattern, make(chan ConsumerMessage, 1), dlq, rlq)
 	if err != nil {
 		t.Fatal(err)
@@ -198,8 +199,9 @@ func runRegexConsumerDiscoverPatternFoo(t *testing.T, c Client, namespace string
 		Name:                "regex-consumer",
 	}
 
-	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", "regex-consumer", log.DefaultNopLogger())
-	rlq, _ := newRetryRouter(c.(*client), nil, false, log.DefaultNopLogger())
+	dlq, _ := newDlqRouter(c.(*client), nil, tn.Topic, "regex-sub", "regex-consumer",
+		nil, log.DefaultNopLogger())
+	rlq, _ := newRetryRouter(c.(*client), nil, false, nil, log.DefaultNopLogger())
 	consumer, err := newRegexConsumer(c.(*client), opts, tn, pattern, make(chan ConsumerMessage, 1), dlq, rlq)
 	if err != nil {
 		t.Fatal(err)
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index 04439cb..a4a8d99 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -30,6 +30,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsaradmin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/utils"
@@ -3874,12 +3876,14 @@ func TestConsumerWithBackoffPolicy(t *testing.T) {
 
 	topicName := newTopicName()
 
-	backoff := newTestBackoffPolicy(1*time.Second, 4*time.Second)
+	bo := newTestBackoffPolicy(1*time.Second, 4*time.Second)
 	_consumer, err := client.Subscribe(ConsumerOptions{
 		Topic:            topicName,
 		SubscriptionName: "sub-1",
 		Type:             Shared,
-		BackoffPolicy:    backoff,
+		BackOffPolicyFunc: func() backoff.Policy {
+			return bo
+		},
 	})
 	assert.Nil(t, err)
 	defer _consumer.Close()
@@ -3888,22 +3892,22 @@ func TestConsumerWithBackoffPolicy(t *testing.T) {
 	// 1 s
 	startTime := time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 2 s
 	startTime = time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 4 s
 	startTime = time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 4 s
 	startTime = time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 }
 
 func TestAckWithMessageID(t *testing.T) {
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 5b23182..afd1f09 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -34,6 +34,8 @@ import (
 	"github.com/testcontainers/testcontainers-go"
 	"github.com/testcontainers/testcontainers-go/wait"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/protobuf/proto"
 
@@ -1293,11 +1295,13 @@ func TestProducerWithBackoffPolicy(t *testing.T) {
 
 	topicName := newTopicName()
 
-	backoff := newTestBackoffPolicy(1*time.Second, 4*time.Second)
+	bo := newTestBackoffPolicy(1*time.Second, 4*time.Second)
 	_producer, err := client.CreateProducer(ProducerOptions{
-		Topic:         topicName,
-		SendTimeout:   2 * time.Second,
-		BackoffPolicy: backoff,
+		Topic:       topicName,
+		SendTimeout: 2 * time.Second,
+		BackOffPolicyFunc: func() backoff.Policy {
+			return bo
+		},
 	})
 	assert.Nil(t, err)
 	defer _producer.Close()
@@ -1306,22 +1310,22 @@ func TestProducerWithBackoffPolicy(t *testing.T) {
 	// 1 s
 	startTime := time.Now()
 	partitionProducerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 2 s
 	startTime = time.Now()
 	partitionProducerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 4 s
 	startTime = time.Now()
 	partitionProducerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 4 s
 	startTime = time.Now()
 	partitionProducerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 }
 
 func TestSendContextExpired(t *testing.T) {
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index d00346f..3c928c1 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -23,6 +23,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/apache/pulsar-client-go/pulsar/backoff"
+
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
 	"github.com/apache/pulsar-client-go/pulsaradmin"
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/admin/config"
@@ -847,6 +849,13 @@ func (b *testBackoffPolicy) Next() time.Duration {
 
 	return b.curBackoff
 }
+func (b *testBackoffPolicy) IsMaxBackoffReached() bool {
+	return false
+}
+
+func (b *testBackoffPolicy) Reset() {
+
+}
 
 func (b *testBackoffPolicy) IsExpectedIntervalFrom(startTime time.Time) bool {
 	// Approximately equal to expected interval
@@ -866,11 +875,13 @@ func TestReaderWithBackoffPolicy(t *testing.T) {
 	assert.Nil(t, err)
 	defer client.Close()
 
-	backoff := newTestBackoffPolicy(1*time.Second, 4*time.Second)
+	bo := newTestBackoffPolicy(1*time.Second, 4*time.Second)
 	_reader, err := client.CreateReader(ReaderOptions{
 		Topic:          "my-topic",
 		StartMessageID: LatestMessageID(),
-		BackoffPolicy:  backoff,
+		BackoffPolicyFunc: func() backoff.Policy {
+			return bo
+		},
 	})
 	assert.NotNil(t, _reader)
 	assert.Nil(t, err)
@@ -879,22 +890,22 @@ func TestReaderWithBackoffPolicy(t *testing.T) {
 	// 1 s
 	startTime := time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 2 s
 	startTime = time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 4 s
 	startTime = time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 
 	// 4 s
 	startTime = time.Now()
 	partitionConsumerImp.reconnectToBroker(nil)
-	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+	assert.True(t, bo.IsExpectedIntervalFrom(startTime))
 }
 
 func TestReaderGetLastMessageID(t *testing.T) {
diff --git a/pulsaradmin/pkg/admin/schema_test.go b/pulsaradmin/pkg/admin/schema_test.go
index 3560559..1b56fdc 100644
--- a/pulsaradmin/pkg/admin/schema_test.go
+++ b/pulsaradmin/pkg/admin/schema_test.go
@@ -27,6 +27,31 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestSchemas_GetAllSchemas(t *testing.T) {
+	cfg := &config.Config{}
+	admin, err := New(cfg)
+	assert.NoError(t, err)
+	assert.NotNil(t, admin)
+
+	topic := fmt.Sprintf("my-topic-%v", time.Now().Nanosecond())
+	schemaPayload := utils.PostSchemaPayload{
+		SchemaType: "STRING",
+		Schema:     "",
+	}
+	err = admin.Schemas().CreateSchemaByPayload(topic, schemaPayload)
+	assert.NoError(t, err)
+
+	infos, err := admin.Schemas().GetAllSchemas(topic)
+	assert.NoError(t, err)
+	assert.Len(t, infos, 1)
+
+	err = admin.Schemas().ForceDeleteSchema(topic)
+	assert.NoError(t, err)
+
+	_, err = admin.Schemas().GetSchemaInfo(topic)
+	assert.Errorf(t, err, "Schema not found")
+}
+
 func TestSchemas_DeleteSchema(t *testing.T) {
 	cfg := &config.Config{}
 	admin, err := New(cfg)