Repository: apache/pulsar-client-go
Pull request #1092 (2023-09-11 10:07:06)

Add 0.11.1 change log
<!--
### Contribution Checklist
  
  - Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
    Skip *Issue XYZ* if there is no associated github issue for this pull request.
    Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.

  - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
  
  - Each pull request should address only one issue, not mix up code from multiple issues.
  
  - Each commit in the pull request has a meaningful commit message

  - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.

**(The sections below can be removed for hotfixes of typos)**
-->

### Motivation

Add 0.11.1 change log

**Note: This PR shouldn't be merged until the release vote is passed.**

### Documentation

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index f1d4839..d7afe2c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -23,6 +23,25 @@
 
 All notable changes to this project will be documented in this file.
 
+[0.11.1] 2023-09-11
+
+- Close consumer resources if the creation fails by @michaeljmarshall in [#1070](https://github.com/apache/pulsar-client-go/pull/1070)
+- Fix the transaction acknowledgement and send logic for chunked message by @liangyepianzhou in [#1069](https://github.com/apache/pulsar-client-go/pull/1069)
+- Correct the `SendAsync()` description by @Gleiphir2769 in [#1066](https://github.com/apache/pulsar-client-go/pull/1066)
+- Fix the panic when try to flush in `DisableBatching=true` by @Gleiphir2769 in [#1065](https://github.com/apache/pulsar-client-go/pull/1065)
+- Fix client reconnected every authenticationRefreshCheckSeconds when using tls authentication by @jffp113 in [#1062](https://github.com/apache/pulsar-client-go/pull/1062)
+- Send Close Command on Producer/Consumer create timeout by @michaeljmarshall in [#1061](https://github.com/apache/pulsar-client-go/pull/1061)
+- Fail all messages that are pending requests when closing by @graysonzeng in [#1059](https://github.com/apache/pulsar-client-go/pull/1059)
+- Fix the producer flush opertion is not guarantee to flush all messages by @Gleiphir2769 in [#1058](https://github.com/apache/pulsar-client-go/pull/1058)
+- Fix inaccurate producer mem limit in chunking and schema by @Gleiphir2769 in [#1055](https://github.com/apache/pulsar-client-go/pull/1055)
+- Fix ctx in `partitionProducer.Send()` is not performing as expected by @Gleiphir2769 in [#1053](https://github.com/apache/pulsar-client-go/pull/1053)
+- Stop block request even if Value and Payload are both set by @gunli in [#1052](https://github.com/apache/pulsar-client-go/pull/1052)
+- Simplify the flush logic by @gunli in [#1049](https://github.com/apache/pulsar-client-go/pull/1049)
+- Check if message is nil by @gunli in [#1047](https://github.com/apache/pulsar-client-go/pull/1047)
+- Return when registerSendOrAckOp() failed by @gunli in [#1045](https://github.com/apache/pulsar-client-go/pull/1045)
+
+**Full Changelog**: https://github.com/apache/pulsar-client-go/compare/v0.11.0...v0.11.1-candidate-1
+
 [0.11.0] 2023-07-04
 
 ## Features
diff --git a/VERSION b/VERSION
index 89d00e6..dbf0637 100644
--- a/VERSION
+++ b/VERSION
@@ -1,3 +1,3 @@
 // This version number refers to the currently released version number
 // Please fix the version when release.
-v0.11.0
+v0.11.1
diff --git a/pulsar/client.go b/pulsar/client.go
index d6f18c9..d9ac2a7 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -153,7 +153,7 @@ type ClientOptions struct {
 	MetricsRegisterer prometheus.Registerer
 
 	// Release the connection if it is not used for more than ConnectionMaxIdleTime.
-	// Default is 60 seconds, negative such as -1 to disable.
+	// Default is 180 seconds, minimum is 60 seconds. Negative such as -1 to disable.
 	ConnectionMaxIdleTime time.Duration
 
 	EnableTransaction bool
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 801eab3..7daf6f6 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -31,7 +31,6 @@ import (
 )
 
 const (
-	defaultConnectionTimeout           = 10 * time.Second
 	defaultOperationTimeout            = 30 * time.Second
 	defaultKeepAliveInterval           = 30 * time.Second
 	defaultMemoryLimitBytes            = 64 * 1024 * 1024
@@ -117,10 +116,10 @@ func newClient(options ClientOptions) (Client, error) {
 		return nil, err
 	}
 
+	// the default timeout respects Go's default timeout which is no timeout
+	// Missing user specified timeout renders 0 values that matches
+	// net.Dailer's default if time.Duration value is not initialized
 	connectionTimeout := options.ConnectionTimeout
-	if connectionTimeout.Nanoseconds() == 0 {
-		connectionTimeout = defaultConnectionTimeout
-	}
 
 	operationTimeout := options.OperationTimeout
 	if operationTimeout.Nanoseconds() == 0 {
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 364dae7..2ba8427 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -1644,6 +1644,13 @@ func (pc *partitionConsumer) reconnectToBroker() {
 		pc.log.Info("Reconnecting to broker in ", delayReconnectTime)
 		time.Sleep(delayReconnectTime)
 
+		// double check
+		if pc.getConsumerState() != consumerReady {
+			// Consumer is already closing
+			pc.log.Info("consumer state not ready, exit reconnect")
+			return
+		}
+
 		err := pc.grabConn()
 		if err == nil {
 			// Successfully reconnected
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index e2ae7ac..840ecc4 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -256,7 +256,11 @@ func (c *connection) connect() bool {
 
 	if c.tlsOptions == nil {
 		// Clear text connection
-		cnx, err = net.DialTimeout("tcp", c.physicalAddr.Host, c.connectionTimeout)
+		if c.connectionTimeout.Nanoseconds() > 0 {
+			cnx, err = net.DialTimeout("tcp", c.physicalAddr.Host, c.connectionTimeout)
+		} else {
+			cnx, err = net.Dial("tcp", c.physicalAddr.Host)
+		}
 	} else {
 		// TLS connection
 		tlsConfig, err = c.getTLSConfig()
@@ -265,6 +269,8 @@ func (c *connection) connect() bool {
 			return false
 		}
 
+		// time.Duration is initialized to 0 by default, net.Dialer's default timeout is no timeout
+		// therefore if c.connectionTimeout is 0, it means no timeout
 		d := &net.Dialer{Timeout: c.connectionTimeout}
 		cnx, err = tls.DialWithDialer(d, "tcp", c.physicalAddr.Host, tlsConfig)
 	}
diff --git a/pulsar/producer_interceptor.go b/pulsar/producer_interceptor.go
index cb2cc15..e18994c 100644
--- a/pulsar/producer_interceptor.go
+++ b/pulsar/producer_interceptor.go
@@ -19,6 +19,7 @@ package pulsar
 
 type ProducerInterceptor interface {
 	// BeforeSend This is called before send the message to the brokers. This method is allowed to modify the
+	// message.
 	BeforeSend(producer Producer, message *ProducerMessage)
 
 	// OnSendAcknowledgement This method is called when the message sent to the broker has been acknowledged,
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 720c7df..8608df3 100755
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -251,7 +251,7 @@ func (p *partitionProducer) grabCnx() error {
 	res, err := p.client.rpcClient.Request(lr.LogicalAddr, lr.PhysicalAddr, id, pb.BaseCommand_PRODUCER, cmdProducer)
 	if err != nil {
 		p.log.WithError(err).Error("Failed to create producer at send PRODUCER request")
-		if err == internal.ErrRequestTimeOut {
+		if errors.Is(err, internal.ErrRequestTimeOut) {
 			id := p.client.rpcClient.NewRequestID()
 			_, _ = p.client.rpcClient.Request(lr.LogicalAddr, lr.PhysicalAddr, id, pb.BaseCommand_CLOSE_PRODUCER,
 				&pb.CommandCloseProducer{
@@ -396,6 +396,11 @@ func (p *partitionProducer) reconnectToBroker() {
 		maxRetry = int(*p.options.MaxReconnectToBroker)
 	}
 
+	var (
+		delayReconnectTime time.Duration
+		defaultBackoff     = internal.DefaultBackoff{}
+	)
+
 	for maxRetry != 0 {
 		if p.getProducerState() != producerReady {
 			// Producer is already closing
@@ -403,11 +408,6 @@ func (p *partitionProducer) reconnectToBroker() {
 			return
 		}
 
-		var (
-			delayReconnectTime time.Duration
-			defaultBackoff     = internal.DefaultBackoff{}
-		)
-
 		if p.options.BackoffPolicy == nil {
 			delayReconnectTime = defaultBackoff.Next()
 		} else {
@@ -478,231 +478,85 @@ func runCallback(cb func(MessageID, *ProducerMessage, error), id MessageID, msg
 	cb(id, msg, err)
 }
 
-func (p *partitionProducer) internalSend(request *sendRequest) {
-	p.log.Debug("Received send request: ", *request.msg)
-
-	msg := request.msg
-
-	// read payload from message
-	uncompressedPayload := msg.Payload
+func (p *partitionProducer) internalSend(sr *sendRequest) {
+	p.log.Debug("Received send request: ", *sr.msg)
 
-	var schemaPayload []byte
-	var err error
+	if sr.sendAsBatch {
+		smm := p.genSingleMessageMetadataInBatch(sr.msg, int(sr.uncompressedSize))
+		multiSchemaEnabled := !p.options.DisableMultiSchema
 
-	// The block chan must be closed when returned with exception
-	defer request.stopBlock()
-	if !p.canAddToQueue(request) {
-		return
-	}
+		added := addRequestToBatch(
+			smm, p, sr.uncompressedPayload, sr, sr.msg, sr.deliverAt, sr.schemaVersion, multiSchemaEnabled)
+		if !added {
+			// The current batch is full. flush it and retry
+			p.internalFlushCurrentBatch()
 
-	if p.options.DisableMultiSchema {
-		if msg.Schema != nil && p.options.Schema != nil &&
-			msg.Schema.GetSchemaInfo().hash() != p.options.Schema.GetSchemaInfo().hash() {
-			runCallback(request.callback, nil, request.msg, fmt.Errorf("msg schema can not match with producer schema"))
-			p.log.WithError(err).Errorf("The producer %s of the topic %s is disabled the `MultiSchema`", p.producerName, p.topic)
-			return
-		}
-	}
-	var schema Schema
-	var schemaVersion []byte
-	if msg.Schema != nil {
-		schema = msg.Schema
-	} else if p.options.Schema != nil {
-		schema = p.options.Schema
-	}
-	if msg.Value != nil {
-		// payload and schema are mutually exclusive
-		// try to get payload from schema value only if payload is not set
-		if uncompressedPayload == nil && schema != nil {
-			schemaPayload, err = schema.Encode(msg.Value)
-			if err != nil {
-				runCallback(request.callback, nil, request.msg, newError(SchemaFailure, err.Error()))
-				p.log.WithError(err).Errorf("Schema encode message failed %s", msg.Value)
+			// after flushing try again to add the current payload
+			ok := addRequestToBatch(
+				smm, p, sr.uncompressedPayload, sr, sr.msg, sr.deliverAt, sr.schemaVersion, multiSchemaEnabled)
+			if !ok {
+				p.log.WithField("size", sr.uncompressedSize).
+					WithField("properties", sr.msg.Properties).
+					Error("unable to add message to batch")
+				sr.done(nil, errFailAddToBatch)
 				return
 			}
 		}
-	}
-	if uncompressedPayload == nil {
-		uncompressedPayload = schemaPayload
-	}
 
-	if schema != nil {
-		schemaVersion = p.schemaCache.Get(schema.GetSchemaInfo())
-		if schemaVersion == nil {
-			schemaVersion, err = p.getOrCreateSchema(schema.GetSchemaInfo())
-			if err != nil {
-				p.log.WithError(err).Error("get schema version fail")
-				runCallback(request.callback, nil, request.msg, fmt.Errorf("get schema version fail, err: %w", err))
-				return
-			}
-			p.schemaCache.Put(schema.GetSchemaInfo(), schemaVersion)
+		if sr.flushImmediately {
+			p.internalFlushCurrentBatch()
 		}
-	}
-
-	uncompressedSize := len(uncompressedPayload)
-
-	// try to reserve memory for uncompressedPayload
-	if !p.canReserveMem(request, int64(uncompressedSize)) {
 		return
 	}
 
-	deliverAt := msg.DeliverAt
-	if msg.DeliverAfter.Nanoseconds() > 0 {
-		deliverAt = time.Now().Add(msg.DeliverAfter)
-	}
-
-	mm := p.genMetadata(msg, uncompressedSize, deliverAt)
-
-	// set default ReplicationClusters when DisableReplication
-	if msg.DisableReplication {
-		msg.ReplicationClusters = []string{"__local__"}
-	}
-
-	sendAsBatch := !p.options.DisableBatching &&
-		msg.ReplicationClusters == nil &&
-		deliverAt.UnixNano() < 0
-
-	// Once the batching is enabled, it can close blockCh early to make block finish
-	if sendAsBatch {
-		request.stopBlock()
-	} else {
-		// update sequence id for metadata, make the size of msgMetadata more accurate
-		// batch sending will update sequence ID in the BatchBuilder
-		p.updateMetadataSeqID(mm, msg)
-	}
-
-	maxMessageSize := int(p._getConn().GetMaxMessageSize())
-
-	// compress payload if not batching
-	var compressedPayload []byte
-	var compressedSize int
-	var checkSize int
-	if !sendAsBatch {
-		compressedPayload = p.compressionProvider.Compress(nil, uncompressedPayload)
-		compressedSize = len(compressedPayload)
-		checkSize = compressedSize
-
-		// set the compress type in msgMetaData
-		compressionType := pb.CompressionType(p.options.CompressionType)
-		if compressionType != pb.CompressionType_NONE {
-			mm.Compression = &compressionType
-		}
-	} else {
-		// final check for batching message is in serializeMessage
-		// this is a double check
-		checkSize = uncompressedSize
-	}
-
-	// if msg is too large and chunking is disabled
-	if checkSize > maxMessageSize && !p.options.EnableChunking {
-		p.releaseSemaphoreAndMem(int64(uncompressedSize))
-		runCallback(request.callback, nil, request.msg, errMessageTooLarge)
-		p.log.WithError(errMessageTooLarge).
-			WithField("size", checkSize).
-			WithField("properties", msg.Properties).
-			Errorf("MaxMessageSize %d", maxMessageSize)
-		p.metrics.PublishErrorsMsgTooLarge.Inc()
+	if sr.totalChunks <= 1 {
+		p.internalSingleSend(sr.mm, sr.compressedPayload, sr, uint32(sr.maxMessageSize))
 		return
 	}
 
-	var totalChunks int
-	// max chunk payload size
-	var payloadChunkSize int
-	if sendAsBatch || !p.options.EnableChunking {
-		totalChunks = 1
-		payloadChunkSize = int(p._getConn().GetMaxMessageSize())
-	} else {
-		payloadChunkSize = int(p._getConn().GetMaxMessageSize()) - proto.Size(mm)
-		if payloadChunkSize <= 0 {
-			p.releaseSemaphoreAndMem(int64(uncompressedSize))
-			runCallback(request.callback, nil, msg, errMetaTooLarge)
-			p.log.WithError(errMetaTooLarge).
-				WithField("metadata size", proto.Size(mm)).
-				WithField("properties", msg.Properties).
-				Errorf("MaxMessageSize %d", int(p._getConn().GetMaxMessageSize()))
-			p.metrics.PublishErrorsMsgTooLarge.Inc()
-			return
-		}
-		// set ChunkMaxMessageSize
-		if p.options.ChunkMaxMessageSize != 0 {
-			payloadChunkSize = int(math.Min(float64(payloadChunkSize), float64(p.options.ChunkMaxMessageSize)))
-		}
-		totalChunks = int(math.Max(1, math.Ceil(float64(compressedSize)/float64(payloadChunkSize))))
-	}
-
-	// set total chunks to send request
-	request.totalChunks = totalChunks
-
-	if !sendAsBatch {
-		if totalChunks > 1 {
-			var lhs, rhs int
-			uuid := fmt.Sprintf("%s-%s", p.producerName, strconv.FormatUint(*mm.SequenceId, 10))
-			mm.Uuid = proto.String(uuid)
-			mm.NumChunksFromMsg = proto.Int32(int32(totalChunks))
-			mm.TotalChunkMsgSize = proto.Int32(int32(compressedSize))
-			cr := newChunkRecorder()
-			for chunkID := 0; chunkID < totalChunks; chunkID++ {
-				lhs = chunkID * payloadChunkSize
-				if rhs = lhs + payloadChunkSize; rhs > compressedSize {
-					rhs = compressedSize
-				}
-				// update chunk id
-				mm.ChunkId = proto.Int32(int32(chunkID))
-				nsr := &sendRequest{
-					ctx:              request.ctx,
-					msg:              request.msg,
-					callback:         request.callback,
-					callbackOnce:     request.callbackOnce,
-					publishTime:      request.publishTime,
-					blockCh:          request.blockCh,
-					closeBlockChOnce: request.closeBlockChOnce,
-					totalChunks:      totalChunks,
-					chunkID:          chunkID,
-					uuid:             uuid,
-					chunkRecorder:    cr,
-					transaction:      request.transaction,
-					reservedMem:      int64(rhs - lhs),
-				}
-				// the permit of first chunk has acquired
-				if chunkID != 0 && !p.canAddToQueue(nsr) {
-					p.releaseSemaphoreAndMem(int64(uncompressedSize - lhs))
-					return
-				}
-				p.internalSingleSend(mm, compressedPayload[lhs:rhs], nsr, uint32(maxMessageSize))
-			}
-			// close the blockCh when all the chunks acquired permits
-			request.stopBlock()
-		} else {
-			// close the blockCh when totalChunks is 1 (it has acquired permits)
-			request.stopBlock()
-			p.internalSingleSend(mm, compressedPayload, request, uint32(maxMessageSize))
+	var lhs, rhs int
+	uuid := fmt.Sprintf("%s-%s", p.producerName, strconv.FormatUint(*sr.mm.SequenceId, 10))
+	sr.mm.Uuid = proto.String(uuid)
+	sr.mm.NumChunksFromMsg = proto.Int32(int32(sr.totalChunks))
+	sr.mm.TotalChunkMsgSize = proto.Int32(int32(sr.compressedSize))
+	cr := newChunkRecorder()
+	for chunkID := 0; chunkID < sr.totalChunks; chunkID++ {
+		lhs = chunkID * sr.payloadChunkSize
+		if rhs = lhs + sr.payloadChunkSize; rhs > sr.compressedSize {
+			rhs = sr.compressedSize
 		}
-	} else {
-		smm := p.genSingleMessageMetadataInBatch(msg, uncompressedSize)
-		multiSchemaEnabled := !p.options.DisableMultiSchema
-		added := addRequestToBatch(smm, p, uncompressedPayload, request, msg, deliverAt, schemaVersion,
-			multiSchemaEnabled)
-		if !added {
-			// The current batch is full. flush it and retry
-
-			p.internalFlushCurrentBatch()
-
-			// after flushing try again to add the current payload
-			if ok := addRequestToBatch(smm, p, uncompressedPayload, request, msg, deliverAt, schemaVersion,
-				multiSchemaEnabled); !ok {
-				p.releaseSemaphoreAndMem(int64(uncompressedSize))
-				runCallback(request.callback, nil, request.msg, errFailAddToBatch)
-				p.log.WithField("size", uncompressedSize).
-					WithField("properties", msg.Properties).
-					Error("unable to add message to batch")
-				return
-			}
+		// update chunk id
+		sr.mm.ChunkId = proto.Int32(int32(chunkID))
+		nsr := &sendRequest{
+			ctx:                 sr.ctx,
+			msg:                 sr.msg,
+			producer:            sr.producer,
+			callback:            sr.callback,
+			callbackOnce:        sr.callbackOnce,
+			publishTime:         sr.publishTime,
+			flushImmediately:    sr.flushImmediately,
+			totalChunks:         sr.totalChunks,
+			chunkID:             chunkID,
+			uuid:                uuid,
+			chunkRecorder:       cr,
+			transaction:         sr.transaction,
+			memLimit:            sr.memLimit,
+			semaphore:           sr.semaphore,
+			reservedMem:         int64(rhs - lhs),
+			sendAsBatch:         sr.sendAsBatch,
+			schema:              sr.schema,
+			schemaVersion:       sr.schemaVersion,
+			uncompressedPayload: sr.uncompressedPayload,
+			uncompressedSize:    sr.uncompressedSize,
+			compressedPayload:   sr.compressedPayload,
+			compressedSize:      sr.compressedSize,
+			payloadChunkSize:    sr.payloadChunkSize,
+			mm:                  sr.mm,
+			deliverAt:           sr.deliverAt,
+			maxMessageSize:      sr.maxMessageSize,
 		}
-		if request.flushImmediately {
-
-			p.internalFlushCurrentBatch()
 
-		}
+		p.internalSingleSend(nsr.mm, nsr.compressedPayload[lhs:rhs], nsr, uint32(nsr.maxMessageSize))
 	}
 }
 
@@ -774,8 +628,10 @@ func (p *partitionProducer) updateSingleMessageMetadataSeqID(smm *pb.SingleMessa
 	}
 }
 
-func (p *partitionProducer) genSingleMessageMetadataInBatch(msg *ProducerMessage,
-	uncompressedSize int) (smm *pb.SingleMessageMetadata) {
+func (p *partitionProducer) genSingleMessageMetadataInBatch(
+	msg *ProducerMessage,
+	uncompressedSize int,
+) (smm *pb.SingleMessageMetadata) {
 	smm = &pb.SingleMessageMetadata{
 		PayloadSize: proto.Int32(int32(uncompressedSize)),
 	}
@@ -801,11 +657,13 @@ func (p *partitionProducer) genSingleMessageMetadataInBatch(msg *ProducerMessage
 	return
 }
 
-func (p *partitionProducer) internalSingleSend(mm *pb.MessageMetadata,
+func (p *partitionProducer) internalSingleSend(
+	mm *pb.MessageMetadata,
 	compressedPayload []byte,
-	request *sendRequest,
-	maxMessageSize uint32) {
-	msg := request.msg
+	sr *sendRequest,
+	maxMessageSize uint32,
+) {
+	msg := sr.msg
 
 	payloadBuf := internal.NewBuffer(len(compressedPayload))
 	payloadBuf.Write(compressedPayload)
@@ -820,8 +678,8 @@ func (p *partitionProducer) internalSingleSend(mm *pb.MessageMetadata,
 	var mostSigBits uint64
 	var leastSigBits uint64
 
-	if request.transaction != nil {
-		txnID := request.transaction.GetTxnID()
+	if sr.transaction != nil {
+		txnID := sr.transaction.GetTxnID()
 		useTxn = true
 		mostSigBits = txnID.MostSigBits
 		leastSigBits = txnID.LeastSigBits
@@ -841,8 +699,7 @@ func (p *partitionProducer) internalSingleSend(mm *pb.MessageMetadata,
 	)
 
 	if err != nil {
-		runCallback(request.callback, nil, request.msg, err)
-		p.releaseSemaphoreAndMem(request.reservedMem)
+		sr.done(nil, err)
 		p.log.WithError(err).Errorf("Single message serialize failed %s", msg.Value)
 		return
 	}
@@ -851,7 +708,7 @@ func (p *partitionProducer) internalSingleSend(mm *pb.MessageMetadata,
 		sentAt:       time.Now(),
 		buffer:       buffer,
 		sequenceID:   sid,
-		sendRequests: []interface{}{request},
+		sendRequests: []interface{}{sr},
 	})
 	p._getConn().WriteData(buffer)
 }
@@ -862,11 +719,18 @@ type pendingItem struct {
 	sequenceID    uint64
 	sentAt        time.Time
 	sendRequests  []interface{}
-	completed     bool
+	isDone        bool
 	flushCallback func(err error)
 }
 
 func (p *partitionProducer) internalFlushCurrentBatch() {
+	if p.batchBuilder == nil {
+		// batch is not enabled
+		// the batch flush ticker should be stopped but it might still called once
+		// depends on when stop() is called concurrently
+		// so we add check to prevent the flow continues on a nil batchBuilder
+		return
+	}
 	if p.batchBuilder.IsMultiBatches() {
 		p.internalFlushCurrentBatches()
 		return
@@ -882,15 +746,14 @@ func (p *partitionProducer) internalFlushCurrentBatch() {
 	if err != nil {
 		for _, cb := range callbacks {
 			if sr, ok := cb.(*sendRequest); ok {
-				runCallback(sr.callback, nil, sr.msg, err)
+				sr.done(nil, err)
 			}
 		}
+
 		if errors.Is(err, internal.ErrExceedMaxMessageSize) {
-			p.log.WithError(errMessageTooLarge).
-				Errorf("internal err: %s", err)
-			p.metrics.PublishErrorsMsgTooLarge.Inc()
-			return
+			p.log.WithError(errMessageTooLarge).Errorf("internal err: %s", err)
 		}
+
 		return
 	}
 
@@ -979,29 +842,11 @@ func (p *partitionProducer) failTimeoutMessages() {
 
 			for _, i := range pi.sendRequests {
 				sr := i.(*sendRequest)
-				if sr.msg != nil {
-					size := len(sr.msg.Payload)
-					p.releaseSemaphoreAndMem(sr.reservedMem)
-					p.metrics.MessagesPending.Dec()
-					p.metrics.BytesPending.Sub(float64(size))
-					p.metrics.PublishErrorsTimeout.Inc()
-					p.log.WithError(errSendTimeout).
-						WithField("size", size).
-						WithField("properties", sr.msg.Properties)
-				}
-
-				if sr.callback != nil {
-					sr.callbackOnce.Do(func() {
-						runCallback(sr.callback, nil, sr.msg, errSendTimeout)
-					})
-				}
-				if sr.transaction != nil {
-					sr.transaction.endSendOrAckOp(nil)
-				}
+				sr.done(nil, errSendTimeout)
 			}
 
 			// flag the sending has completed with error, flush make no effect
-			pi.Complete(errSendTimeout)
+			pi.done(errSendTimeout)
 			pi.Unlock()
 
 			// finally reached the last view item, current iteration ends
@@ -1025,15 +870,15 @@ func (p *partitionProducer) internalFlushCurrentBatches() {
 		if errs[i] != nil {
 			for _, cb := range callbacks[i] {
 				if sr, ok := cb.(*sendRequest); ok {
-					runCallback(sr.callback, nil, sr.msg, errs[i])
+					sr.done(nil, errs[i])
 				}
 			}
+
 			if errors.Is(errs[i], internal.ErrExceedMaxMessageSize) {
-				p.log.WithError(errMessageTooLarge).
-					Errorf("internal err: %s", errs[i])
-				p.metrics.PublishErrorsMsgTooLarge.Inc()
+				p.log.WithError(errMessageTooLarge).Errorf("internal err: %s", errs[i])
 				return
 			}
+
 			continue
 		}
 		if batchesData[i] == nil {
@@ -1076,7 +921,7 @@ func (p *partitionProducer) internalFlush(fr *flushRequest) {
 	pi.Lock()
 	defer pi.Unlock()
 
-	if pi.completed {
+	if pi.isDone {
 		// The last item in the queue has been completed while we were
 		// looking at it. It's safe at this point to assume that every
 		// message enqueued before Flush() was called are now persisted
@@ -1121,78 +966,240 @@ func (p *partitionProducer) SendAsync(ctx context.Context, msg *ProducerMessage,
 	p.internalSendAsync(ctx, msg, callback, false)
 }
 
-func (p *partitionProducer) internalSendAsync(ctx context.Context, msg *ProducerMessage,
-	callback func(MessageID, *ProducerMessage, error), flushImmediately bool) {
+func (p *partitionProducer) validateMsg(msg *ProducerMessage) error {
 	if msg == nil {
-		p.log.Error("Message is nil")
-		runCallback(callback, nil, msg, newError(InvalidMessage, "Message is nil"))
-		return
+		return newError(InvalidMessage, "Message is nil")
 	}
 
 	if msg.Value != nil && msg.Payload != nil {
-		p.log.Error("Can not set Value and Payload both")
-		runCallback(callback, nil, msg, newError(InvalidMessage, "Can not set Value and Payload both"))
-		return
+		return newError(InvalidMessage, "Can not set Value and Payload both")
 	}
 
-	// Register transaction operation to transaction and the transaction coordinator.
-	var newCallback func(MessageID, *ProducerMessage, error)
-	var txn *transaction
-	if msg.Transaction != nil {
-		transactionImpl := (msg.Transaction).(*transaction)
-		txn = transactionImpl
-		if transactionImpl.state != TxnOpen {
-			p.log.WithField("state", transactionImpl.state).Error("Failed to send message" +
-				" by a non-open transaction.")
-			runCallback(callback, nil, msg, newError(InvalidStatus, "Failed to send message by a non-open transaction."))
-			return
+	if p.options.DisableMultiSchema {
+		if msg.Schema != nil && p.options.Schema != nil &&
+			msg.Schema.GetSchemaInfo().hash() != p.options.Schema.GetSchemaInfo().hash() {
+			p.log.Errorf("The producer %s of the topic %s is disabled the `MultiSchema`", p.producerName, p.topic)
+			return fmt.Errorf("msg schema can not match with producer schema")
 		}
+	}
 
-		if err := transactionImpl.registerProducerTopic(p.topic); err != nil {
-			runCallback(callback, nil, msg, err)
-			return
+	return nil
+}
+
+func (p *partitionProducer) prepareTransaction(sr *sendRequest) error {
+	if sr.msg.Transaction == nil {
+		return nil
+	}
+
+	txn := (sr.msg.Transaction).(*transaction)
+	if txn.state != TxnOpen {
+		p.log.WithField("state", txn.state).Error("Failed to send message" +
+			" by a non-open transaction.")
+		return newError(InvalidStatus, "Failed to send message by a non-open transaction.")
+	}
+
+	if err := txn.registerProducerTopic(p.topic); err != nil {
+		return err
+	}
+
+	if err := txn.registerSendOrAckOp(); err != nil {
+		return err
+	}
+
+	sr.transaction = txn
+	return nil
+}
+
+func (p *partitionProducer) updateSchema(sr *sendRequest) error {
+	var schema Schema
+	var schemaVersion []byte
+	var err error
+
+	if sr.msg.Schema != nil {
+		schema = sr.msg.Schema
+	} else if p.options.Schema != nil {
+		schema = p.options.Schema
+	}
+
+	if schema == nil {
+		return nil
+	}
+
+	schemaVersion = p.schemaCache.Get(schema.GetSchemaInfo())
+	if schemaVersion == nil {
+		schemaVersion, err = p.getOrCreateSchema(schema.GetSchemaInfo())
+		if err != nil {
+			return fmt.Errorf("get schema version fail, err: %w", err)
 		}
-		if err := transactionImpl.registerSendOrAckOp(); err != nil {
-			runCallback(callback, nil, msg, err)
-			return
+		p.schemaCache.Put(schema.GetSchemaInfo(), schemaVersion)
+	}
+
+	sr.schema = schema
+	sr.schemaVersion = schemaVersion
+	return nil
+}
+
+func (p *partitionProducer) updateUncompressedPayload(sr *sendRequest) error {
+	// read payload from message
+	sr.uncompressedPayload = sr.msg.Payload
+
+	if sr.msg.Value != nil {
+		if sr.schema == nil {
+			p.log.Errorf("Schema encode message failed %s", sr.msg.Value)
+			return newError(SchemaFailure, "set schema value without setting schema")
 		}
-		newCallback = func(id MessageID, producerMessage *ProducerMessage, err error) {
-			runCallback(callback, id, producerMessage, err)
-			transactionImpl.endSendOrAckOp(err)
+
+		// payload and schema are mutually exclusive
+		// try to get payload from schema value only if payload is not set
+		schemaPayload, err := sr.schema.Encode(sr.msg.Value)
+		if err != nil {
+			p.log.WithError(err).Errorf("Schema encode message failed %s", sr.msg.Value)
+			return newError(SchemaFailure, err.Error())
 		}
+
+		sr.uncompressedPayload = schemaPayload
+	}
+
+	sr.uncompressedSize = int64(len(sr.uncompressedPayload))
+	return nil
+}
+
+func (p *partitionProducer) updateMetaData(sr *sendRequest) {
+	deliverAt := sr.msg.DeliverAt
+	if sr.msg.DeliverAfter.Nanoseconds() > 0 {
+		deliverAt = time.Now().Add(sr.msg.DeliverAfter)
+	}
+
+	// set default ReplicationClusters when DisableReplication
+	if sr.msg.DisableReplication {
+		sr.msg.ReplicationClusters = []string{"__local__"}
+	}
+
+	sr.mm = p.genMetadata(sr.msg, int(sr.uncompressedSize), deliverAt)
+
+	sr.sendAsBatch = !p.options.DisableBatching &&
+		sr.msg.ReplicationClusters == nil &&
+		deliverAt.UnixNano() < 0
+
+	if !sr.sendAsBatch {
+		// update sequence id for metadata, make the size of msgMetadata more accurate
+		// batch sending will update sequence ID in the BatchBuilder
+		p.updateMetadataSeqID(sr.mm, sr.msg)
+	}
+
+	sr.deliverAt = deliverAt
+}
+
+func (p *partitionProducer) updateChunkInfo(sr *sendRequest) error {
+	checkSize := sr.uncompressedSize
+	if !sr.sendAsBatch {
+		sr.compressedPayload = p.compressionProvider.Compress(nil, sr.uncompressedPayload)
+		sr.compressedSize = len(sr.compressedPayload)
+
+		// set the compress type in msgMetaData
+		compressionType := pb.CompressionType(p.options.CompressionType)
+		if compressionType != pb.CompressionType_NONE {
+			sr.mm.Compression = &compressionType
+		}
+
+		checkSize = int64(sr.compressedSize)
+	}
+
+	sr.maxMessageSize = p._getConn().GetMaxMessageSize()
+
+	// if msg is too large and chunking is disabled
+	if checkSize > int64(sr.maxMessageSize) && !p.options.EnableChunking {
+		p.log.WithError(errMessageTooLarge).
+			WithField("size", checkSize).
+			WithField("properties", sr.msg.Properties).
+			Errorf("MaxMessageSize %d", sr.maxMessageSize)
+		return errMessageTooLarge
+	}
+
+	if sr.sendAsBatch || !p.options.EnableChunking {
+		sr.totalChunks = 1
+		sr.payloadChunkSize = int(sr.maxMessageSize)
 	} else {
-		newCallback = callback
+		sr.payloadChunkSize = int(sr.maxMessageSize) - proto.Size(sr.mm)
+		if sr.payloadChunkSize <= 0 {
+			p.log.WithError(errMetaTooLarge).
+				WithField("metadata size", proto.Size(sr.mm)).
+				WithField("properties", sr.msg.Properties).
+				Errorf("MaxMessageSize %d", int(p._getConn().GetMaxMessageSize()))
+			return errMetaTooLarge
+		}
+		// set ChunkMaxMessageSize
+		if p.options.ChunkMaxMessageSize != 0 {
+			sr.payloadChunkSize = int(math.Min(float64(sr.payloadChunkSize), float64(p.options.ChunkMaxMessageSize)))
+		}
+		sr.totalChunks = int(math.Max(1, math.Ceil(float64(sr.compressedSize)/float64(sr.payloadChunkSize))))
 	}
-	if p.getProducerState() != producerReady {
-		// Producer is closing
-		runCallback(newCallback, nil, msg, errProducerClosed)
+
+	return nil
+}
+
+func (p *partitionProducer) internalSendAsync(
+	ctx context.Context,
+	msg *ProducerMessage,
+	callback func(MessageID, *ProducerMessage, error),
+	flushImmediately bool,
+) {
+	if err := p.validateMsg(msg); err != nil {
+		p.log.Error(err)
+		runCallback(callback, nil, msg, err)
 		return
 	}
 
-	// bc only works when DisableBlockIfQueueFull is false
-	bc := make(chan struct{})
-
-	// callbackOnce make sure the callback is only invoked once in chunking
-	callbackOnce := &sync.Once{}
 	sr := &sendRequest{
 		ctx:              ctx,
 		msg:              msg,
-		callback:         newCallback,
-		callbackOnce:     callbackOnce,
+		producer:         p,
+		callback:         callback,
+		callbackOnce:     &sync.Once{},
 		flushImmediately: flushImmediately,
 		publishTime:      time.Now(),
-		blockCh:          bc,
-		closeBlockChOnce: &sync.Once{},
-		transaction:      txn,
+		chunkID:          -1,
 	}
+
+	if err := p.prepareTransaction(sr); err != nil {
+		sr.done(nil, err)
+		return
+	}
+
+	if p.getProducerState() != producerReady {
+		sr.done(nil, errProducerClosed)
+		return
+	}
+
 	p.options.Interceptors.BeforeSend(p, msg)
 
-	p.dataChan <- sr
+	if err := p.updateSchema(sr); err != nil {
+		p.log.Error(err)
+		sr.done(nil, err)
+		return
+	}
+
+	if err := p.updateUncompressedPayload(sr); err != nil {
+		p.log.Error(err)
+		sr.done(nil, err)
+		return
+	}
+
+	p.updateMetaData(sr)
 
-	if !p.options.DisableBlockIfQueueFull {
-		// block if queue full
-		<-bc
+	if err := p.updateChunkInfo(sr); err != nil {
+		p.log.Error(err)
+		sr.done(nil, err)
+		return
 	}
+
+	if err := p.reserveResources(sr); err != nil {
+		p.log.Error(err)
+		sr.done(nil, err)
+		return
+	}
+
+	p.dataChan <- sr
 }
 
 func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt) {
@@ -1206,14 +1213,14 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 
 	if pi.sequenceID < response.GetSequenceId() {
 		// Force connection closing so that messages can be re-transmitted in a new connection
-		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, closing connection", response.GetMessageId(),
-			response.GetSequenceId(), pi.sequenceID)
+		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, local < remote, closing connection",
+			response.GetMessageId(), response.GetSequenceId(), pi.sequenceID)
 		p._getConn().Close()
 		return
 	} else if pi.sequenceID > response.GetSequenceId() {
 		// Ignoring the ack since it's referring to a message that has already timed out.
-		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, closing connection", response.GetMessageId(),
-			response.GetSequenceId(), pi.sequenceID)
+		p.log.Warnf("Received ack for %v on sequenceId %v - expected: %v, local > remote, ignore it",
+			response.GetMessageId(), response.GetSequenceId(), pi.sequenceID)
 		return
 	} else {
 		// The ack was indeed for the expected item in the queue, we can remove it and trigger the callback
@@ -1229,59 +1236,44 @@ func (p *partitionProducer) ReceivedSendReceipt(response *pb.CommandSendReceipt)
 		for idx, i := range pi.sendRequests {
 			sr := i.(*sendRequest)
 			atomic.StoreInt64(&p.lastSequenceID, int64(pi.sequenceID))
-			p.releaseSemaphoreAndMem(sr.reservedMem)
-			p.metrics.PublishLatency.Observe(float64(now-sr.publishTime.UnixNano()) / 1.0e9)
-			p.metrics.MessagesPublished.Inc()
-			p.metrics.MessagesPending.Dec()
-			payloadSize := float64(len(sr.msg.Payload))
-			p.metrics.BytesPublished.Add(payloadSize)
-			p.metrics.BytesPending.Sub(payloadSize)
-
-			if sr.callback != nil || len(p.options.Interceptors) > 0 {
-				msgID := newMessageID(
-					int64(response.MessageId.GetLedgerId()),
-					int64(response.MessageId.GetEntryId()),
-					int32(idx),
-					p.partitionIdx,
-					batchSize,
-				)
-
-				if sr.totalChunks > 1 {
-					if sr.chunkID == 0 {
-						sr.chunkRecorder.setFirstChunkID(
-							&messageID{
-								int64(response.MessageId.GetLedgerId()),
-								int64(response.MessageId.GetEntryId()),
-								-1,
-								p.partitionIdx,
-								0,
-							})
-					} else if sr.chunkID == sr.totalChunks-1 {
-						sr.chunkRecorder.setLastChunkID(
-							&messageID{
-								int64(response.MessageId.GetLedgerId()),
-								int64(response.MessageId.GetEntryId()),
-								-1,
-								p.partitionIdx,
-								0,
-							})
-						// use chunkMsgID to set msgID
-						msgID = &sr.chunkRecorder.chunkedMsgID
-					}
-				}
 
-				if sr.totalChunks <= 1 || sr.chunkID == sr.totalChunks-1 {
-					runCallback(sr.callback, msgID, sr.msg, nil)
-					p.options.Interceptors.OnSendAcknowledgement(p, sr.msg, msgID)
+			msgID := newMessageID(
+				int64(response.MessageId.GetLedgerId()),
+				int64(response.MessageId.GetEntryId()),
+				int32(idx),
+				p.partitionIdx,
+				batchSize,
+			)
+
+			if sr.totalChunks > 1 {
+				if sr.chunkID == 0 {
+					sr.chunkRecorder.setFirstChunkID(
+						&messageID{
+							int64(response.MessageId.GetLedgerId()),
+							int64(response.MessageId.GetEntryId()),
+							-1,
+							p.partitionIdx,
+							0,
+						})
+				} else if sr.chunkID == sr.totalChunks-1 {
+					sr.chunkRecorder.setLastChunkID(
+						&messageID{
+							int64(response.MessageId.GetLedgerId()),
+							int64(response.MessageId.GetEntryId()),
+							-1,
+							p.partitionIdx,
+							0,
+						})
+					// use chunkMsgID to set msgID
+					msgID = &sr.chunkRecorder.chunkedMsgID
 				}
 			}
-			if sr.transaction != nil {
-				sr.transaction.endSendOrAckOp(nil)
-			}
+
+			sr.done(msgID, nil)
 		}
 
 		// Mark this pending item as done
-		pi.Complete(nil)
+		pi.done(nil)
 	}
 }
 
@@ -1344,28 +1336,11 @@ func (p *partitionProducer) failPendingMessages() {
 
 		for _, i := range pi.sendRequests {
 			sr := i.(*sendRequest)
-			if sr.msg != nil {
-				size := len(sr.msg.Payload)
-				p.releaseSemaphoreAndMem(sr.reservedMem)
-				p.metrics.MessagesPending.Dec()
-				p.metrics.BytesPending.Sub(float64(size))
-				p.log.WithError(errProducerClosed).
-					WithField("size", size).
-					WithField("properties", sr.msg.Properties)
-			}
-
-			if sr.callback != nil {
-				sr.callbackOnce.Do(func() {
-					runCallback(sr.callback, nil, sr.msg, errProducerClosed)
-				})
-			}
-			if sr.transaction != nil {
-				sr.transaction.endSendOrAckOp(nil)
-			}
+			sr.done(nil, errProducerClosed)
 		}
 
 		// flag the sending has completed with error, flush make no effect
-		pi.Complete(errProducerClosed)
+		pi.done(errProducerClosed)
 		pi.Unlock()
 
 		// finally reached the last view item, current iteration ends
@@ -1422,25 +1397,151 @@ func (p *partitionProducer) Close() {
 type sendRequest struct {
 	ctx              context.Context
 	msg              *ProducerMessage
+	producer         *partitionProducer
 	callback         func(MessageID, *ProducerMessage, error)
 	callbackOnce     *sync.Once
 	publishTime      time.Time
 	flushImmediately bool
-	blockCh          chan struct{}
-	closeBlockChOnce *sync.Once
 	totalChunks      int
 	chunkID          int
 	uuid             string
 	chunkRecorder    *chunkRecorder
-	transaction      *transaction
-	reservedMem      int64
+
+	/// resource management
+
+	memLimit          internal.MemoryLimitController
+	reservedMem       int64
+	semaphore         internal.Semaphore
+	reservedSemaphore int
+
+	/// convey settable state
+
+	sendAsBatch         bool
+	transaction         *transaction
+	schema              Schema
+	schemaVersion       []byte
+	uncompressedPayload []byte
+	uncompressedSize    int64
+	compressedPayload   []byte
+	compressedSize      int
+	payloadChunkSize    int
+	mm                  *pb.MessageMetadata
+	deliverAt           time.Time
+	maxMessageSize      int32
 }
 
-// stopBlock can be invoked multiple times safety
-func (sr *sendRequest) stopBlock() {
-	sr.closeBlockChOnce.Do(func() {
-		close(sr.blockCh)
-	})
+func (sr *sendRequest) done(msgID MessageID, err error) {
+	if err == nil {
+		sr.producer.metrics.PublishLatency.Observe(float64(time.Now().UnixNano()-sr.publishTime.UnixNano()) / 1.0e9)
+		sr.producer.metrics.MessagesPublished.Inc()
+		sr.producer.metrics.BytesPublished.Add(float64(sr.reservedMem))
+
+		if sr.totalChunks <= 1 || sr.chunkID == sr.totalChunks-1 {
+			if sr.producer.options.Interceptors != nil {
+				sr.producer.options.Interceptors.OnSendAcknowledgement(sr.producer, sr.msg, msgID)
+			}
+		}
+	}
+
+	if err != nil {
+		sr.producer.log.WithError(err).
+			WithField("size", sr.reservedMem).
+			WithField("properties", sr.msg.Properties)
+	}
+
+	if errors.Is(err, errSendTimeout) {
+		sr.producer.metrics.PublishErrorsTimeout.Inc()
+	}
+
+	if errors.Is(err, errMessageTooLarge) {
+		sr.producer.metrics.PublishErrorsMsgTooLarge.Inc()
+	}
+
+	if sr.semaphore != nil {
+		sr.semaphore.Release()
+		sr.producer.metrics.MessagesPending.Dec()
+	}
+
+	if sr.memLimit != nil {
+		sr.memLimit.ReleaseMemory(sr.reservedMem)
+		sr.producer.metrics.BytesPending.Sub(float64(sr.reservedMem))
+	}
+
+	// sr.chunkID == -1 means a chunked message is not yet prepared, so that we should fail it immediately
+	if sr.totalChunks <= 1 || sr.chunkID == -1 || sr.chunkID == sr.totalChunks-1 {
+		sr.callbackOnce.Do(func() {
+			runCallback(sr.callback, msgID, sr.msg, err)
+		})
+
+		if sr.transaction != nil {
+			sr.transaction.endSendOrAckOp(err)
+		}
+	}
+}
+
+func (p *partitionProducer) blockIfQueueFull() bool {
+	//DisableBlockIfQueueFull == false means enable block
+	return !p.options.DisableBlockIfQueueFull
+}
+
+func (p *partitionProducer) reserveSemaphore(sr *sendRequest) error {
+	for i := 0; i < sr.totalChunks; i++ {
+		if p.blockIfQueueFull() {
+			if !p.publishSemaphore.Acquire(sr.ctx) {
+				return errContextExpired
+			}
+
+			// update sr.semaphore and sr.reservedSemaphore here so that we can release semaphore in the case
+			// of that only a part of the chunks acquire succeed
+			sr.semaphore = p.publishSemaphore
+			sr.reservedSemaphore++
+			p.metrics.MessagesPending.Inc()
+		} else {
+			if !p.publishSemaphore.TryAcquire() {
+				return errSendQueueIsFull
+			}
+
+			// update sr.semaphore and sr.reservedSemaphore here so that we can release semaphore in the case
+			// of that only a part of the chunks acquire succeed
+			sr.semaphore = p.publishSemaphore
+			sr.reservedSemaphore++
+			p.metrics.MessagesPending.Inc()
+		}
+	}
+
+	return nil
+}
+
+func (p *partitionProducer) reserveMem(sr *sendRequest) error {
+	requiredMem := sr.uncompressedSize
+	if !sr.sendAsBatch {
+		requiredMem = int64(sr.compressedSize)
+	}
+
+	if p.blockIfQueueFull() {
+		if !p.client.memLimit.ReserveMemory(sr.ctx, requiredMem) {
+			return errContextExpired
+		}
+	} else {
+		if !p.client.memLimit.TryReserveMemory(requiredMem) {
+			return errMemoryBufferIsFull
+		}
+	}
+
+	sr.memLimit = p.client.memLimit
+	sr.reservedMem += requiredMem
+	p.metrics.BytesPending.Add(float64(requiredMem))
+	return nil
+}
+
+func (p *partitionProducer) reserveResources(sr *sendRequest) error {
+	if err := p.reserveSemaphore(sr); err != nil {
+		return err
+	}
+	if err := p.reserveMem(sr); err != nil {
+		return err
+	}
+	return nil
 }
 
 type closeProducer struct {
@@ -1452,11 +1553,11 @@ type flushRequest struct {
 	err    error
 }
 
-func (i *pendingItem) Complete(err error) {
-	if i.completed {
+func (i *pendingItem) done(err error) {
+	if i.isDone {
 		return
 	}
-	i.completed = true
+	i.isDone = true
 	buffersPool.Put(i.buffer)
 	if i.flushCallback != nil {
 		i.flushCallback(err)
@@ -1472,53 +1573,12 @@ func (p *partitionProducer) _setConn(conn internal.Connection) {
 // _getConn returns internal connection field of this partition producer atomically.
 // Note: should only be called by this partition producer before attempting to use the connection
 func (p *partitionProducer) _getConn() internal.Connection {
-	// Invariant: The conn must be non-nil for the lifetime of the partitionProducer.
+	// Invariant: p.conn must be non-nil for the lifetime of the partitionProducer.
 	//            For this reason we leave this cast unchecked and panic() if the
 	//            invariant is broken
 	return p.conn.Load().(internal.Connection)
 }
 
-func (p *partitionProducer) releaseSemaphoreAndMem(size int64) {
-	p.publishSemaphore.Release()
-	p.client.memLimit.ReleaseMemory(size)
-}
-
-func (p *partitionProducer) canAddToQueue(sr *sendRequest) bool {
-	if p.options.DisableBlockIfQueueFull {
-		if !p.publishSemaphore.TryAcquire() {
-			runCallback(sr.callback, nil, sr.msg, errSendQueueIsFull)
-			return false
-		}
-	} else {
-		if !p.publishSemaphore.Acquire(sr.ctx) {
-			runCallback(sr.callback, nil, sr.msg, errContextExpired)
-			return false
-		}
-	}
-	p.metrics.MessagesPending.Inc()
-	return true
-}
-
-func (p *partitionProducer) canReserveMem(sr *sendRequest, size int64) bool {
-	if p.options.DisableBlockIfQueueFull {
-		if !p.client.memLimit.TryReserveMemory(size) {
-			p.publishSemaphore.Release()
-			runCallback(sr.callback, nil, sr.msg, errMemoryBufferIsFull)
-			return false
-		}
-
-	} else {
-		if !p.client.memLimit.ReserveMemory(sr.ctx, size) {
-			p.publishSemaphore.Release()
-			runCallback(sr.callback, nil, sr.msg, errContextExpired)
-			return false
-		}
-	}
-	sr.reservedMem += size
-	p.metrics.BytesPending.Add(float64(size))
-	return true
-}
-
 type chunkRecorder struct {
 	chunkedMsgID chunkMessageID
 }
diff --git a/pulsar/reader.go b/pulsar/reader.go
index d58d06f..5e1a73b 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -134,4 +134,7 @@ type Reader interface {
 	//            the message publish time where to reposition the subscription
 	//
 	SeekByTime(time time.Time) error
+
+	// GetLastMessageID get the last message id available for consume.
+	GetLastMessageID() (MessageID, error)
 }
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index 5a2128a..ffc92de 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -244,3 +244,7 @@ func (r *reader) SeekByTime(time time.Time) error {
 
 	return r.pc.SeekByTime(time)
 }
+
+func (r *reader) GetLastMessageID() (MessageID, error) {
+	return r.pc.getLastMessageID()
+}
diff --git a/pulsar/schema.go b/pulsar/schema.go
index 5c063e3..fd9d412 100644
--- a/pulsar/schema.go
+++ b/pulsar/schema.go
@@ -54,7 +54,7 @@ const (
 	_                             //
 	_                             //
 	KeyValue                      //A Schema that contains Key Schema and Value Schema.
-	BYTES       = -1              //A bytes array.
+	BYTES       = 0               //A bytes array.
 	AUTO        = -2              //
 	AutoConsume = -3              //Auto Consume Type.
 	AutoPublish = -4              // Auto Publish Type.
diff --git a/pulsaradmin/pkg/admin/auth/oauth2.go b/pulsaradmin/pkg/admin/auth/oauth2.go
index 2864eae..4bd1546 100644
--- a/pulsaradmin/pkg/admin/auth/oauth2.go
+++ b/pulsaradmin/pkg/admin/auth/oauth2.go
@@ -73,10 +73,15 @@ func NewAuthenticationOAuth2(issuer oauth2.Issuer, store store.Store) (*OAuth2Pr
 
 // NewAuthenticationOAuth2WithDefaultFlow uses memory to save the grant
 func NewAuthenticationOAuth2WithDefaultFlow(issuer oauth2.Issuer, keyFile string) (Provider, error) {
-	st := store.NewMemoryStore()
-	flow, err := oauth2.NewDefaultClientCredentialsFlow(oauth2.ClientCredentialsFlowOptions{
+	return NewAuthenticationOAuth2WithFlow(issuer, oauth2.ClientCredentialsFlowOptions{
 		KeyFile: keyFile,
 	})
+}
+
+func NewAuthenticationOAuth2WithFlow(
+	issuer oauth2.Issuer, flowOptions oauth2.ClientCredentialsFlowOptions) (Provider, error) {
+	st := store.NewMemoryStore()
+	flow, err := oauth2.NewDefaultClientCredentialsFlow(flowOptions)
 	if err != nil {
 		return nil, err
 	}
diff --git a/stable.txt b/stable.txt
index 66c4773..7e38472 100644
--- a/stable.txt
+++ b/stable.txt
@@ -1,3 +1,3 @@
 // This version number refers to the current stable version, generally is `VERSION - 1`.
 // Please fix the version when release.
-v0.11.0
+v0.11.1


Test patch:
diff --git a/pulsar/message_chunking_test.go b/pulsar/message_chunking_test.go
index ee3ab17..59fdb5e 100644
--- a/pulsar/message_chunking_test.go
+++ b/pulsar/message_chunking_test.go
@@ -24,6 +24,7 @@ import (
 	"math/rand"
 	"net/http"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -531,12 +532,13 @@ func TestChunkBlockIfQueueFull(t *testing.T) {
 	assert.NotNil(t, producer)
 	defer producer.Close()
 
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 	// Large messages will be split into 11 chunks, exceeding the length of pending queue
-	ID, err := producer.Send(context.Background(), &ProducerMessage{
+	_, err = producer.Send(ctx, &ProducerMessage{
 		Payload: createTestMessagePayload(100),
 	})
-	assert.NoError(t, err)
-	assert.NotNil(t, ID)
+	assert.Error(t, err)
 }
 
 func createTestMessagePayload(size int) []byte {
@@ -552,26 +554,38 @@ func sendSingleChunk(p Producer, uuid string, chunkID int, totalChunks int) {
 	msg := &ProducerMessage{
 		Payload: []byte(fmt.Sprintf("chunk-%s-%d|", uuid, chunkID)),
 	}
+	wholePayload := msg.Payload
 	producerImpl := p.(*producer).producers[0].(*partitionProducer)
-	mm := producerImpl.genMetadata(msg, len(msg.Payload), time.Now())
+	mm := producerImpl.genMetadata(msg, len(wholePayload), time.Now())
 	mm.Uuid = proto.String(uuid)
 	mm.NumChunksFromMsg = proto.Int32(int32(totalChunks))
-	mm.TotalChunkMsgSize = proto.Int32(int32(len(msg.Payload)))
+	mm.TotalChunkMsgSize = proto.Int32(int32(len(wholePayload)))
 	mm.ChunkId = proto.Int32(int32(chunkID))
 	producerImpl.updateMetadataSeqID(mm, msg)
-
-	doneCh := make(chan struct{})
 	producerImpl.internalSingleSend(
 		mm,
 		msg.Payload,
 		&sendRequest{
 			callback: func(id MessageID, producerMessage *ProducerMessage, err error) {
-				close(doneCh)
 			},
-			msg: msg,
+			callbackOnce:        &sync.Once{},
+			ctx:                 context.Background(),
+			msg:                 msg,
+			producer:            producerImpl,
+			flushImmediately:    true,
+			totalChunks:         totalChunks,
+			chunkID:             chunkID,
+			uuid:                uuid,
+			chunkRecorder:       newChunkRecorder(),
+			uncompressedPayload: wholePayload,
+			uncompressedSize:    int64(len(wholePayload)),
+			compressedPayload:   wholePayload,
+			compressedSize:      len(wholePayload),
+			payloadChunkSize:    internal.MaxMessageSize - proto.Size(mm),
+			mm:                  mm,
+			deliverAt:           time.Now(),
+			maxMessageSize:      internal.MaxMessageSize,
 		},
 		uint32(internal.MaxMessageSize),
 	)
-
-	<-doneCh
 }
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 38fec57..49e225f 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -30,7 +30,9 @@ import (
 	"time"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal"
+	pb "github.com/apache/pulsar-client-go/pulsar/internal/pulsar_proto"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/apache/pulsar-client-go/pulsar/crypto"
 	plog "github.com/apache/pulsar-client-go/pulsar/log"
@@ -2077,7 +2079,6 @@ func TestMemLimitRejectProducerMessagesWithSchema(t *testing.T) {
 }
 
 func TestMemLimitRejectProducerMessagesWithChunking(t *testing.T) {
-
 	c, err := NewClient(ClientOptions{
 		URL:              serviceURL,
 		MemoryLimitBytes: 5 * 1024,
@@ -2134,12 +2135,11 @@ func TestMemLimitRejectProducerMessagesWithChunking(t *testing.T) {
 		SendTimeout:             2 * time.Second,
 	})
 
-	// producer2 will reserve 2*1024 bytes and then release 1024 byte (release the second chunk)
-	// because it reaches MaxPendingMessages in chunking
+	// producer3 cannot reserve 2*1024 bytes because it reaches MaxPendingMessages in chunking
 	_, _ = producer3.Send(context.Background(), &ProducerMessage{
 		Payload: make([]byte, 2*1024),
 	})
-	assert.Equal(t, int64(1024), c.(*client).memLimit.CurrentUsage())
+	assert.Zero(t, c.(*client).memLimit.CurrentUsage())
 }
 
 func TestMemLimitContextCancel(t *testing.T) {
@@ -2298,3 +2298,93 @@ func TestFailPendingMessageWithClose(t *testing.T) {
 	testProducer.Close()
 	assert.Equal(t, 0, partitionProducerImp.pendingQueue.Size())
 }
+
+type pendingQueueWrapper struct {
+	pendingQueue   internal.BlockingQueue
+	writtenBuffers *[]internal.Buffer
+}
+
+func (pqw *pendingQueueWrapper) Put(item interface{}) {
+	pi := item.(*pendingItem)
+	writerIdx := pi.buffer.WriterIndex()
+	buf := internal.NewBuffer(int(writerIdx))
+	buf.Write(pi.buffer.Get(0, writerIdx))
+	*pqw.writtenBuffers = append(*pqw.writtenBuffers, buf)
+	pqw.pendingQueue.Put(item)
+}
+
+func (pqw *pendingQueueWrapper) Take() interface{} {
+	return pqw.pendingQueue.Take()
+}
+
+func (pqw *pendingQueueWrapper) Poll() interface{} {
+	return pqw.pendingQueue.Poll()
+}
+
+func (pqw *pendingQueueWrapper) CompareAndPoll(compare func(interface{}) bool) interface{} {
+	return pqw.pendingQueue.CompareAndPoll(compare)
+}
+
+func (pqw *pendingQueueWrapper) Peek() interface{} {
+	return pqw.pendingQueue.Peek()
+}
+
+func (pqw *pendingQueueWrapper) PeekLast() interface{} {
+	return pqw.pendingQueue.PeekLast()
+}
+
+func (pqw *pendingQueueWrapper) Size() int {
+	return pqw.pendingQueue.Size()
+}
+
+func (pqw *pendingQueueWrapper) ReadableSlice() []interface{} {
+	return pqw.pendingQueue.ReadableSlice()
+}
+
+func TestDisableReplication(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	testProducer, err := client.CreateProducer(ProducerOptions{
+		Topic:           newTopicName(),
+		DisableBatching: true,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, testProducer)
+	defer testProducer.Close()
+
+	writtenBuffers := make([]internal.Buffer, 0)
+	pqw := &pendingQueueWrapper{
+		pendingQueue:   internal.NewBlockingQueue(1000),
+		writtenBuffers: &writtenBuffers,
+	}
+
+	partitionProducerImp := testProducer.(*producer).producers[0].(*partitionProducer)
+	partitionProducerImp.pendingQueue = pqw
+
+	ID, err := testProducer.Send(context.Background(), &ProducerMessage{
+		Payload:            []byte("disable-replication"),
+		DisableReplication: true,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, ID)
+
+	assert.Equal(t, 1, len(writtenBuffers))
+	buf := writtenBuffers[0]
+
+	buf.Skip(4)                        // TOTAL_SIZE
+	cmdSize := buf.ReadUint32()        // CMD_SIZE
+	buf.Skip(cmdSize)                  // CMD
+	buf.Skip(2)                        // MAGIC_NUMBER
+	buf.Skip(4)                        // CHECKSUM
+	metadataSize := buf.ReadUint32()   // METADATA_SIZE
+	metadata := buf.Read(metadataSize) // METADATA
+
+	var msgMetadata pb.MessageMetadata
+	err = proto.Unmarshal(metadata, &msgMetadata)
+	assert.NoError(t, err)
+	assert.Equal(t, []string{"__local__"}, msgMetadata.GetReplicateTo())
+}
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index 0a1b2a1..ec10f8f 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -901,3 +901,45 @@ func TestReaderWithBackoffPolicy(t *testing.T) {
 	partitionConsumerImp.reconnectToBroker()
 	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
 }
+
+func TestReaderGetLastMessageID(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.Nil(t, err)
+	topic := newTopicName()
+	ctx := context.Background()
+	schema := NewStringSchema(nil)
+	// create producer
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           topic,
+		DisableBatching: true,
+		Schema:          schema,
+	})
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	var lastMsgID MessageID
+	// send 10 messages
+	for i := 0; i < 10; i++ {
+		msgID, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte(fmt.Sprintf("hello-%d", i)),
+		})
+		assert.NoError(t, err)
+		assert.NotNil(t, msgID)
+		lastMsgID = msgID
+	}
+
+	reader, err := client.CreateReader(ReaderOptions{
+		Topic:          topic,
+		StartMessageID: EarliestMessageID(),
+	})
+	assert.Nil(t, err)
+	getLastMessageID, err := reader.GetLastMessageID()
+	if err != nil {
+		return
+	}
+
+	assert.Equal(t, lastMsgID.LedgerID(), getLastMessageID.LedgerID())
+	assert.Equal(t, lastMsgID.EntryID(), getLastMessageID.EntryID())
+}
diff --git a/pulsar/transaction_test.go b/pulsar/transaction_test.go
index 385b197..74e8dd0 100644
--- a/pulsar/transaction_test.go
+++ b/pulsar/transaction_test.go
@@ -458,7 +458,6 @@ func TestAckChunkMessage(t *testing.T) {
 	// Create transaction and register the send operation.
 	txn, err := client.NewTransaction(time.Hour)
 	require.Nil(t, err)
-	txn.(*transaction).registerSendOrAckOp()
 
 	// Create a producer with chunking enabled to send a large message that will be split into chunks.
 	producer, err := client.CreateProducer(ProducerOptions{