Repository: apache/pulsar-client-go
Pull request #1283 (2024-09-11 16:30:29)

improve: use chan *message instead of chan []*message as queueCh
### Motivation

We currently use `chan []*message` as queueCh, and use the slice to stage the messages to send the message to the parent consumer, this will result in excessive use of memory.

This PR optimizes potentially reducing overall memory overhead.


### Modifications

- Use `chan *message` instead of `chan []*message`.
- Fix test.


Patch:
diff --git a/.header b/.header
deleted file mode 100644
index b248758..0000000
--- a/.header
+++ /dev/null
@@ -1,16 +0,0 @@
-// 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.
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/Makefile b/Makefile
index 1fd66be..23844c4 100644
--- a/Makefile
+++ b/Makefile
@@ -53,27 +53,25 @@ container:
 test: container test_standalone test_clustered test_extensible_load_manager
 
 test_standalone: container
-	docker run -v /var/run/docker.sock:/var/run/docker.sock -i ${IMAGE_NAME} bash -c "cd /pulsar/pulsar-client-go && ./scripts/run-ci.sh"
+	docker run --rm -v /var/run/docker.sock:/var/run/docker.sock -i ${IMAGE_NAME} bash -c "cd /pulsar/pulsar-client-go && ./scripts/run-ci.sh"
 
 test_clustered: container
 	PULSAR_VERSION=${PULSAR_VERSION} docker compose -f integration-tests/clustered/docker-compose.yml up -d
 	until curl http://localhost:8080/metrics > /dev/null 2>&1; do sleep 1; done
-	docker run --network "clustered_pulsar" -i ${IMAGE_NAME} bash -c "cd /pulsar/pulsar-client-go && ./scripts/run-ci-clustered.sh"
+	docker run --rm --network "clustered_pulsar" -i ${IMAGE_NAME} bash -c "cd /pulsar/pulsar-client-go && ./scripts/run-ci-clustered.sh"
 	PULSAR_VERSION=${PULSAR_VERSION} docker compose -f integration-tests/clustered/docker-compose.yml down
 
 test_extensible_load_manager: container
 	PULSAR_VERSION=${PULSAR_VERSION} docker compose -f integration-tests/extensible-load-manager/docker-compose.yml up -d
 	until curl http://localhost:8080/metrics > /dev/null 2>&1; do sleep 1; done
-	docker run --network "extensible-load-manager_pulsar" -i ${IMAGE_NAME} bash -c "cd /pulsar/pulsar-client-go && ./scripts/run-ci-extensible-load-manager.sh"
+	docker run --rm --network "extensible-load-manager_pulsar" -i ${IMAGE_NAME} bash -c "cd /pulsar/pulsar-client-go && ./scripts/run-ci-extensible-load-manager.sh"
 
 	PULSAR_VERSION=${PULSAR_VERSION} docker compose -f integration-tests/blue-green/docker-compose.yml up -d
 	until curl http://localhost:8081/metrics > /dev/null 2>&1 ; do sleep 1; done
 
-	# run blue-green migration test (run this test from this env to access both clusters)
-	go test -race -coverprofile=/tmp/coverage-blue_green_topic_migration -timeout=5m -tags extensible_load_manager -v -run TestBlueGreenMigrationTestSuite ./pulsar
-	go tool cover -html=/tmp/coverage-blue_green_topic_migration -o coverage-blue_green_topic_migration.html
-	PULSAR_VERSION=${PULSAR_VERSION} docker compose -f integration-tests/extensible-load-manager/docker-compose.yml down
+	docker run --rm --network="extensible-load-manager_pulsar" -i ${IMAGE_NAME} bash -c "cd /pulsar/pulsar-client-go && ./scripts/run-ci-blue-green-cluster.sh"
 	PULSAR_VERSION=${PULSAR_VERSION} docker compose -f integration-tests/blue-green/docker-compose.yml down
+	PULSAR_VERSION=${PULSAR_VERSION} docker compose -f integration-tests/extensible-load-manager/docker-compose.yml down
 
 clean:
 	docker rmi --force $(IMAGE_NAME) || true
diff --git a/go.mod b/go.mod
index 2266621..1a7025f 100644
--- a/go.mod
+++ b/go.mod
@@ -9,7 +9,11 @@ require (
 	github.com/bits-and-blooms/bitset v1.4.0
 	github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b
 	github.com/davecgh/go-spew v1.1.1
+	github.com/golang-jwt/jwt/v5 v5.2.1
+	github.com/golang/protobuf v1.5.4
 	github.com/google/uuid v1.6.0
+	github.com/hamba/avro/v2 v2.26.0
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/klauspost/compress v1.17.9
 	github.com/onsi/ginkgo v1.16.5
 	github.com/onsi/gomega v1.19.0
@@ -21,21 +25,14 @@ require (
 	github.com/spaolacci/murmur3 v1.1.0
 	github.com/spf13/cobra v1.6.1
 	github.com/stretchr/testify v1.9.0
+	github.com/testcontainers/testcontainers-go v0.32.0
 	go.uber.org/atomic v1.7.0
-	golang.org/x/mod v0.18.0
+	golang.org/x/mod v0.20.0
 	golang.org/x/oauth2 v0.11.0
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8
 	google.golang.org/protobuf v1.33.0
 )
 
-require (
-	github.com/golang-jwt/jwt/v5 v5.2.1
-	github.com/golang/protobuf v1.5.4
-	github.com/hamba/avro/v2 v2.22.2-0.20240625062549-66aad10411d9
-	github.com/hashicorp/go-multierror v1.1.1
-	github.com/testcontainers/testcontainers-go v0.32.0
-)
-
 require (
 	dario.cat/mergo v1.0.0 // indirect
 	github.com/99designs/go-keychain v0.0.0-20191008050251-8e49817e8af4 // indirect
diff --git a/go.sum b/go.sum
index 8a5cd93..b546ffa 100644
--- a/go.sum
+++ b/go.sum
@@ -211,8 +211,8 @@ github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 h1:YBftPWNWd4WwGqtY2yeZL2ef8rH
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0/go.mod h1:YN5jB8ie0yfIUg6VvR9Kz84aCaG7AsGZnLjhHbUqwPg=
 github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c h1:6rhixN/i8ZofjG1Y75iExal34USq5p+wiN1tpie8IrU=
 github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c/go.mod h1:NMPJylDgVpX0MLRlPy15sqSwOFv/U1GZ2m21JhFfek0=
-github.com/hamba/avro/v2 v2.22.2-0.20240625062549-66aad10411d9 h1:NEoabXt33PDWK4fXryK4e+XX+fSKDmmu9vg3yb9YI2M=
-github.com/hamba/avro/v2 v2.22.2-0.20240625062549-66aad10411d9/go.mod h1:fQVdB2mFZBhPW1D5Abej41LMvrErARGrrdjOnKbm5yw=
+github.com/hamba/avro/v2 v2.26.0 h1:IaT5l6W3zh7K67sMrT2+RreJyDTllBGVJm4+Hedk9qE=
+github.com/hamba/avro/v2 v2.26.0/go.mod h1:I8glyswHnpED3Nlx2ZdUe+4LJnCOOyiCzLMno9i/Uu0=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
 github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
@@ -451,8 +451,8 @@ golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzB
 golang.org/x/mod v0.1.1-0.20191107180719-034126e5016b/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
-golang.org/x/mod v0.18.0 h1:5+9lSbEzPSdWkH32vYPBwEpX8KwDbM52Ud9xBUvNlb0=
-golang.org/x/mod v0.18.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
+golang.org/x/mod v0.20.0 h1:utOm6MM3R3dnawAiJgn0y+xvuYRsm1RKM/4giyfDgV0=
+golang.org/x/mod v0.20.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
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 f307972..831f763 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
@@ -152,7 +154,7 @@ type partitionConsumer struct {
 
 	// the size of the queue channel for buffering messages
 	maxQueueSize    int32
-	queueCh         chan []*message
+	queueCh         chan *message
 	startMessageID  atomicMessageID
 	lastDequeuedMsg *trackingMessageID
 
@@ -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,
@@ -328,7 +338,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		partitionIdx:         int32(options.partitionIdx),
 		eventsCh:             make(chan interface{}, 10),
 		maxQueueSize:         int32(options.receiverQueueSize),
-		queueCh:              make(chan []*message, options.receiverQueueSize),
+		queueCh:              make(chan *message, options.receiverQueueSize),
 		startMessageID:       atomicMessageID{msgID: options.startMessageID},
 		connectedCh:          make(chan struct{}),
 		messageCh:            messageCh,
@@ -339,6 +349,7 @@ func newPartitionConsumer(parent Consumer, client *client, options *partitionCon
 		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
 		}
@@ -1051,37 +1057,33 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 			return fmt.Errorf("discarding message on decryption error :%v", err)
 		case crypto.ConsumerCryptoFailureActionConsume:
 			pc.log.Warnf("consuming encrypted message due to error in decryption :%v", err)
-			messages := []*message{
-				{
-					publishTime:  timeFromUnixTimestampMillis(msgMeta.GetPublishTime()),
-					eventTime:    timeFromUnixTimestampMillis(msgMeta.GetEventTime()),
-					key:          msgMeta.GetPartitionKey(),
-					producerName: msgMeta.GetProducerName(),
-					properties:   internal.ConvertToStringMap(msgMeta.GetProperties()),
-					topic:        pc.topic,
-					msgID: newMessageID(
-						int64(pbMsgID.GetLedgerId()),
-						int64(pbMsgID.GetEntryId()),
-						pbMsgID.GetBatchIndex(),
-						pc.partitionIdx,
-						pbMsgID.GetBatchSize(),
-					),
-					payLoad:             headersAndPayload.ReadableSlice(),
-					schema:              pc.options.schema,
-					replicationClusters: msgMeta.GetReplicateTo(),
-					replicatedFrom:      msgMeta.GetReplicatedFrom(),
-					redeliveryCount:     response.GetRedeliveryCount(),
-					encryptionContext:   createEncryptionContext(msgMeta),
-					orderingKey:         string(msgMeta.OrderingKey),
-				},
-			}
-
 			if pc.options.autoReceiverQueueSize {
 				pc.incomingMessages.Inc()
 				pc.markScaleIfNeed()
 			}
 
-			pc.queueCh <- messages
+			pc.queueCh <- &message{
+				publishTime:  timeFromUnixTimestampMillis(msgMeta.GetPublishTime()),
+				eventTime:    timeFromUnixTimestampMillis(msgMeta.GetEventTime()),
+				key:          msgMeta.GetPartitionKey(),
+				producerName: msgMeta.GetProducerName(),
+				properties:   internal.ConvertToStringMap(msgMeta.GetProperties()),
+				topic:        pc.topic,
+				msgID: newMessageID(
+					int64(pbMsgID.GetLedgerId()),
+					int64(pbMsgID.GetEntryId()),
+					pbMsgID.GetBatchIndex(),
+					pc.partitionIdx,
+					pbMsgID.GetBatchSize(),
+				),
+				payLoad:             headersAndPayload.ReadableSlice(),
+				schema:              pc.options.schema,
+				replicationClusters: msgMeta.GetReplicateTo(),
+				replicatedFrom:      msgMeta.GetReplicatedFrom(),
+				redeliveryCount:     response.GetRedeliveryCount(),
+				encryptionContext:   createEncryptionContext(msgMeta),
+				orderingKey:         string(msgMeta.OrderingKey),
+			}
 			return nil
 		}
 	}
@@ -1099,22 +1101,24 @@ 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 {
 		numMsgs = int(msgMeta.GetNumMessagesInBatch())
 	}
 
-	messages := make([]*message, 0)
 	var ackTracker *ackTracker
 	// are there multiple messages in this batch?
 	if numMsgs > 1 {
@@ -1135,7 +1139,6 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 	pc.metrics.PrefetchedMessages.Add(float64(numMsgs))
 
 	var (
-		bytesReceived   int
 		skippedMessages int32
 	)
 	for i := 0; i < numMsgs; i++ {
@@ -1255,22 +1258,19 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 			Message:  msg,
 		})
 
-		messages = append(messages, msg)
-		bytesReceived += msg.size()
-	}
+		if pc.options.autoReceiverQueueSize {
+			pc.client.memLimit.ForceReserveMemory(int64(msg.size()))
+			pc.incomingMessages.Add(int32(1))
+			pc.markScaleIfNeed()
+		}
 
-	if pc.options.autoReceiverQueueSize {
-		pc.client.memLimit.ForceReserveMemory(int64(bytesReceived))
-		pc.incomingMessages.Add(int32(len(messages)))
-		pc.markScaleIfNeed()
+		pc.queueCh <- msg
 	}
 
 	if skippedMessages > 0 {
 		pc.availablePermits.add(skippedMessages)
 	}
 
-	// send messages to the dispatcher
-	pc.queueCh <- messages
 	return nil
 }
 
@@ -1426,20 +1426,19 @@ func (pc *partitionConsumer) dispatcher() {
 	defer func() {
 		pc.log.Debug("exiting dispatch loop")
 	}()
-	var messages []*message
+	var queueMsg *message
 	for {
-		var queueCh chan []*message
+		var queueCh chan *message
 		var messageCh chan ConsumerMessage
 		var nextMessage ConsumerMessage
 		var nextMessageSize int
 
-		// are there more messages to send?
-		if len(messages) > 0 {
+		if queueMsg != nil {
 			nextMessage = ConsumerMessage{
 				Consumer: pc.parentConsumer,
-				Message:  messages[0],
+				Message:  queueMsg,
 			}
-			nextMessageSize = messages[0].size()
+			nextMessageSize = queueMsg.size()
 
 			if pc.dlq.shouldSendToDlq(&nextMessage) {
 				// pass the message to the DLQ router
@@ -1451,7 +1450,7 @@ func (pc *partitionConsumer) dispatcher() {
 			}
 
 			pc.metrics.PrefetchedMessages.Dec()
-			pc.metrics.PrefetchedBytes.Sub(float64(len(messages[0].payLoad)))
+			pc.metrics.PrefetchedBytes.Sub(float64(len(queueMsg.payLoad)))
 		} else {
 			queueCh = pc.queueCh
 		}
@@ -1466,7 +1465,7 @@ func (pc *partitionConsumer) dispatcher() {
 			}
 			pc.log.Debug("dispatcher received connection event")
 
-			messages = nil
+			queueMsg = nil
 
 			// reset available permits
 			pc.availablePermits.reset()
@@ -1484,19 +1483,16 @@ func (pc *partitionConsumer) dispatcher() {
 				pc.log.WithError(err).Error("unable to send initial permits to broker")
 			}
 
-		case msgs, ok := <-queueCh:
+		case msg, ok := <-queueCh:
 			if !ok {
 				return
 			}
-			// we only read messages here after the consumer has processed all messages
-			// in the previous batch
-			messages = msgs
+
+			queueMsg = msg
 
 		// if the messageCh is nil or the messageCh is full this will not be selected
 		case messageCh <- nextMessage:
-			// allow this message to be garbage collected
-			messages[0] = nil
-			messages = messages[1:]
+			queueMsg = nil
 
 			pc.availablePermits.inc()
 
@@ -1519,14 +1515,14 @@ func (pc *partitionConsumer) dispatcher() {
 				if m == nil {
 					break
 				} else if nextMessageInQueue == nil {
-					nextMessageInQueue = toTrackingMessageID(m[0].msgID)
+					nextMessageInQueue = toTrackingMessageID(m.msgID)
 				}
 				if pc.options.autoReceiverQueueSize {
-					pc.incomingMessages.Sub(int32(len(m)))
+					pc.incomingMessages.Sub(int32(1))
 				}
 			}
 
-			messages = nil
+			queueMsg = nil
 
 			clearQueueCb(nextMessageInQueue)
 		}
@@ -1684,18 +1680,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 {
@@ -1710,11 +1705,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,
@@ -1733,6 +1727,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")
@@ -1747,7 +1742,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 462cfe8..c246308 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -51,6 +51,7 @@ type TLSOptions struct {
 	CipherSuites            []uint16
 	MinVersion              uint16
 	MaxVersion              uint16
+	TLSConfig               *tls.Config
 }
 
 var (
@@ -1083,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,
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 97ab8b9..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
@@ -175,15 +187,16 @@ func newPartitionProducer(client *client, topic string, options *ProducerOptions
 		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()
@@ -457,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 {
@@ -488,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{
@@ -512,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")
@@ -545,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
+}
diff --git a/scripts/run-ci-blue-green-cluster.sh b/scripts/run-ci-blue-green-cluster.sh
new file mode 100755
index 0000000..a8528cb
--- /dev/null
+++ b/scripts/run-ci-blue-green-cluster.sh
@@ -0,0 +1,22 @@
+#!/bin/bash
+# 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.
+
+set -e -x
+
+go test -race -coverprofile=/tmp/coverage-blue_green_topic_migration -timeout=5m -tags extensible_load_manager -v -run TestBlueGreenMigrationTestSuite ./pulsar
+go tool cover -html=/tmp/coverage-blue_green_topic_migration -o coverage-blue_green_topic_migration.html
\ No newline at end of file


Test patch:
diff --git a/integration-tests/blue-green/docker-compose.yml b/integration-tests/blue-green/docker-compose.yml
index eea5852..b026ce8 100644
--- a/integration-tests/blue-green/docker-compose.yml
+++ b/integration-tests/blue-green/docker-compose.yml
@@ -17,18 +17,20 @@
 
 version: '3'
 networks:
-  green-pulsar:
-    driver: bridge
+  # We use external extensible-load-manager_pulsar docker network
+  # to make sure blue cluster and green cluster are in the same network
+  extensible-load-manager_pulsar:
+    external: true
 services:
   # Start ZooKeeper
-  zookeeper:
+  green-zookeeper:
     image: apachepulsar/pulsar:${PULSAR_VERSION}
     container_name: green-zookeeper
     restart: on-failure
     networks:
-      - green-pulsar
+      - extensible-load-manager_pulsar
     environment:
-      - metadataStoreUrl=zk:zookeeper:2181
+      - metadataStoreUrl=zk:green-zookeeper:2181
       - PULSAR_MEM=-Xms128m -Xmx128m -XX:MaxDirectMemorySize=56m
     command: >
       bash -c "bin/apply-config-from-env.py conf/zookeeper.conf && \
@@ -41,79 +43,79 @@ services:
       retries: 30
 
   # Initialize cluster metadata
-  pulsar-init:
+  green-pulsar-init:
     container_name: green-pulsar-init
-    hostname: pulsar-init
+    hostname: green-pulsar-init
     image: apachepulsar/pulsar:${PULSAR_VERSION}
     networks:
-      - green-pulsar
+      - extensible-load-manager_pulsar
     environment:
       - PULSAR_MEM=-Xms128m -Xmx128m -XX:MaxDirectMemorySize=56m
     command: >
       bin/pulsar initialize-cluster-metadata \
-               --cluster cluster-a \
-               --zookeeper zookeeper:2181 \
-               --configuration-store zookeeper:2181 \
-               --web-service-url http://broker-1:8080 \
-               --broker-service-url pulsar://broker-1:6650
+               --cluster cluster-green \
+               --zookeeper green-zookeeper:2181 \
+               --configuration-store green-zookeeper:2181 \
+               --web-service-url http://green-broker-1:8080 \
+               --broker-service-url pulsar://green-broker-1:6650
     depends_on:
-      zookeeper:
+      green-zookeeper:
         condition: service_healthy
 
   # Start bookie
-  bookie:
+  green-bookie:
     image: apachepulsar/pulsar:${PULSAR_VERSION}
     container_name: green-bookie
     restart: on-failure
     networks:
-      - green-pulsar
+      - extensible-load-manager_pulsar
     environment:
-      - clusterName=cluster-a
-      - zkServers=zookeeper:2181
-      - metadataServiceUri=metadata-store:zk:zookeeper:2181
-      - advertisedAddress=bookie
+      - clusterName=cluster-green
+      - zkServers=green-zookeeper:2181
+      - metadataServiceUri=metadata-store:zk:green-zookeeper:2181
+      - advertisedAddress=green-bookie
       - BOOKIE_MEM=-Xms128m -Xmx128m -XX:MaxDirectMemorySize=56m
     depends_on:
-      zookeeper:
+      green-zookeeper:
         condition: service_healthy
-      pulsar-init:
+      green-pulsar-init:
         condition: service_completed_successfully
     command: bash -c "bin/apply-config-from-env.py conf/bookkeeper.conf && exec bin/pulsar bookie"
 
-  proxy:
+  green-proxy:
     image: apachepulsar/pulsar:${PULSAR_VERSION}
     container_name: green-proxy
-    hostname: proxy
+    hostname: green-proxy
     restart: on-failure
     networks:
-      - green-pulsar
+      - extensible-load-manager_pulsar
     environment:
-      - metadataStoreUrl=zk:zookeeper:2181
-      - zookeeperServers=zookeeper:2181
-      - clusterName=cluster-a
+      - metadataStoreUrl=zk:green-zookeeper:2181
+      - zookeeperServers=green-zookeeper:2181
+      - clusterName=cluster-green
       - PULSAR_MEM=-Xms128m -Xmx128m -XX:MaxDirectMemorySize=56m
     ports:
       - "8081:8080"
       - "6651:6650"
     depends_on:
-      broker-1:
+      green-broker-1:
         condition: service_started
-      broker-2:
+      green-broker-2:
         condition: service_started
     command: bash -c "bin/apply-config-from-env.py conf/proxy.conf && exec bin/pulsar proxy"
 
   # Start broker 1
-  broker-1:
+  green-broker-1:
     image: apachepulsar/pulsar:${PULSAR_VERSION}
     container_name: green-broker-1
-    hostname: broker-1
+    hostname: green-broker-1
     restart: on-failure
     networks:
-      - green-pulsar
+      - extensible-load-manager_pulsar
     environment:
-      - metadataStoreUrl=zk:zookeeper:2181
-      - zookeeperServers=zookeeper:2181
-      - clusterName=cluster-a
+      - metadataStoreUrl=zk:green-zookeeper:2181
+      - zookeeperServers=green-zookeeper:2181
+      - clusterName=cluster-green
       - managedLedgerDefaultEnsembleSize=1
       - managedLedgerDefaultWriteQuorum=1
       - managedLedgerDefaultAckQuorum=1
@@ -129,24 +131,24 @@ services:
       - brokerServiceCompactionThresholdInBytes=1000000
       - PULSAR_PREFIX_defaultNumberOfNamespaceBundles=1
     depends_on:
-      zookeeper:
+      green-zookeeper:
         condition: service_healthy
-      bookie:
+      green-bookie:
         condition: service_started
     command: bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker"
 
   # Start broker 2
-  broker-2:
+  green-broker-2:
     image: apachepulsar/pulsar:${PULSAR_VERSION}
     container_name: green-broker-2
-    hostname: broker-2
+    hostname: green-broker-2
     restart: on-failure
     networks:
-      - green-pulsar
+      - extensible-load-manager_pulsar
     environment:
-      - metadataStoreUrl=zk:zookeeper:2181
-      - zookeeperServers=zookeeper:2181
-      - clusterName=cluster-a
+      - metadataStoreUrl=zk:green-zookeeper:2181
+      - zookeeperServers=green-zookeeper:2181
+      - clusterName=cluster-green
       - managedLedgerDefaultEnsembleSize=1
       - managedLedgerDefaultWriteQuorum=1
       - managedLedgerDefaultAckQuorum=1
@@ -162,8 +164,8 @@ services:
       - brokerServiceCompactionThresholdInBytes=1000000
       - PULSAR_PREFIX_defaultNumberOfNamespaceBundles=1
     depends_on:
-      zookeeper:
+      green-zookeeper:
         condition: service_healthy
-      bookie:
+      green-bookie:
         condition: service_started
     command: bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker"
\ No newline at end of file
diff --git a/pulsar/ack_grouping_tracker_test.go b/pulsar/ack_grouping_tracker_test.go
index 202d19d..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) {
@@ -243,7 +243,7 @@ func TestCloseFlushWithoutTimer(t *testing.T) {
 func TestCloseFlushWithTimer(t *testing.T) {
 	var acker mockAcker
 	tracker := newAckGroupingTracker(
-		&AckGroupingOptions{MaxSize: 1000, MaxTime: 10 * 1000},
+		&AckGroupingOptions{MaxSize: 1000, MaxTime: 10 * time.Second},
 		nil,
 		func(id MessageID) { acker.ackCumulative(id) },
 		func(ids []*pb.MessageIdData) { acker.ack(ids) },
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..4cd2e47 100644
--- a/pulsar/blue_green_migration_test.go
+++ b/pulsar/blue_green_migration_test.go
@@ -54,16 +54,16 @@ func (suite *BlueGreenMigrationTestSuite) TestTopicMigration() {
 	for _, scenario := range []topicUnloadTestCase{
 
 		{
-			testCaseName: "proxyConnection",
-			blueAdminURL: "http://localhost:8080",
-			blueClientUrl: "pulsar://localhost:6650",
-			greenAdminURL: "http://localhost:8081",
+			testCaseName:  "proxyConnection",
+			blueAdminURL:  "http://proxy:8080",
+			blueClientUrl: "pulsar://proxy:6650",
+			greenAdminURL: "http://green-proxy:8080",
 			migrationBody: `
 						{
-							"serviceUrl": "http://localhost:8081",
-							"serviceUrlTls":"https://localhost:8085",
-							"brokerServiceUrl": "pulsar://localhost:6651",
-							"brokerServiceUrlTls": "pulsar+ssl://localhost:6655"
+							"serviceUrl": "http://green-proxy:8080",
+							"serviceUrlTls":"https://green-proxy:8081",
+							"brokerServiceUrl": "pulsar://green-proxy:6650",
+							"brokerServiceUrlTls": "pulsar+ssl://green-proxy:6651"
 						}
 					`,
 		},
@@ -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_partition_test.go b/pulsar/consumer_partition_test.go
index 21280fb..746e17a 100644
--- a/pulsar/consumer_partition_test.go
+++ b/pulsar/consumer_partition_test.go
@@ -30,7 +30,7 @@ import (
 func TestSingleMessageIDNoAckTracker(t *testing.T) {
 	eventsCh := make(chan interface{}, 1)
 	pc := partitionConsumer{
-		queueCh:              make(chan []*message, 1),
+		queueCh:              make(chan *message, 1),
 		eventsCh:             eventsCh,
 		compressionProviders: sync.Map{},
 		options:              &partitionConsumerOpts{},
@@ -47,13 +47,12 @@ func TestSingleMessageIDNoAckTracker(t *testing.T) {
 	}
 
 	// ensure the tracker was set on the message id
-	messages := <-pc.queueCh
-	for _, m := range messages {
-		assert.Nil(t, m.ID().(*trackingMessageID).tracker)
-	}
+	message := <-pc.queueCh
+	id := message.ID().(*trackingMessageID)
+	assert.Nil(t, id.tracker)
 
 	// ack the message id
-	pc.AckID(messages[0].msgID.(*trackingMessageID))
+	pc.AckID(id)
 
 	select {
 	case <-eventsCh:
@@ -69,7 +68,7 @@ func newTestMetrics() *internal.LeveledMetrics {
 func TestBatchMessageIDNoAckTracker(t *testing.T) {
 	eventsCh := make(chan interface{}, 1)
 	pc := partitionConsumer{
-		queueCh:              make(chan []*message, 1),
+		queueCh:              make(chan *message, 1),
 		eventsCh:             eventsCh,
 		compressionProviders: sync.Map{},
 		options:              &partitionConsumerOpts{},
@@ -86,13 +85,12 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 	}
 
 	// ensure the tracker was set on the message id
-	messages := <-pc.queueCh
-	for _, m := range messages {
-		assert.Nil(t, m.ID().(*trackingMessageID).tracker)
-	}
+	message := <-pc.queueCh
+	id := message.ID().(*trackingMessageID)
+	assert.Nil(t, id.tracker)
 
 	// ack the message id
-	err := pc.AckID(messages[0].msgID.(*trackingMessageID))
+	err := pc.AckID(id)
 	assert.Nil(t, err)
 
 	select {
@@ -105,7 +103,7 @@ func TestBatchMessageIDNoAckTracker(t *testing.T) {
 func TestBatchMessageIDWithAckTracker(t *testing.T) {
 	eventsCh := make(chan interface{}, 1)
 	pc := partitionConsumer{
-		queueCh:              make(chan []*message, 1),
+		queueCh:              make(chan *message, 10),
 		eventsCh:             eventsCh,
 		compressionProviders: sync.Map{},
 		options:              &partitionConsumerOpts{},
@@ -122,14 +120,21 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 	}
 
 	// ensure the tracker was set on the message id
-	messages := <-pc.queueCh
-	for _, m := range messages {
-		assert.NotNil(t, m.ID().(*trackingMessageID).tracker)
+	var messageIDs []*trackingMessageID
+	for i := 0; i < 10; i++ {
+		select {
+		case m := <-pc.queueCh:
+			id := m.ID().(*trackingMessageID)
+			assert.NotNil(t, id.tracker)
+			messageIDs = append(messageIDs, id)
+		default:
+			break
+		}
 	}
 
 	// ack all message ids except the last one
 	for i := 0; i < 9; i++ {
-		err := pc.AckID(messages[i].msgID.(*trackingMessageID))
+		err := pc.AckID(messageIDs[i])
 		assert.Nil(t, err)
 	}
 
@@ -140,7 +145,7 @@ func TestBatchMessageIDWithAckTracker(t *testing.T) {
 	}
 
 	// ack last message
-	err := pc.AckID(messages[9].msgID.(*trackingMessageID))
+	err := pc.AckID(messageIDs[9])
 	assert.Nil(t, err)
 
 	select {
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)