fix: add internal error handle , fix inaccurate PublishErrorsMsgTooLarge metric.

Author: Gleiphir2769

pulsar/internal/commands.go

@@ -50,6 +50,8 @@ var ErrEOM = errors.New("EOF")
 
 var ErrConnectionClosed = errors.New("connection closed")
 
+var ErrExceedMaxMessageSize = errors.New("encryptedPayload exceeds MaxMessageSize")
+
 func NewMessageReader(headersAndPayload Buffer) *MessageReader {
 	return &MessageReader{
 		buffer: headersAndPayload,
@@ -265,8 +267,8 @@ func serializeMessage(wb Buffer,
 	}
 	// the maxMessageSize check of batching message is in here
 	if len(encryptedPayload) > int(maxMessageSize) {
-		return fmt.Errorf("encryptedPayload size exceeds MaxMessageSize, size: %d, MaxMessageSize: %d",
-			len(encryptedPayload), maxMessageSize)
+		return fmt.Errorf("%w, size: %d, MaxMessageSize: %d",
+			ErrExceedMaxMessageSize, len(encryptedPayload), maxMessageSize)
 	}
 
 	cmdSize := uint32(proto.Size(cmdSend))

pulsar/producer_partition.go

@@ -993,6 +993,10 @@ func (p *partitionProducer) Send(ctx context.Context, msg *ProducerMessage) (Mes
 
 	// wait for send request to finish
 	<-doneCh
+
+	// handle internal error
+	p.internalErrHandle(err)
+
 	return msgID, err
 }
 
@@ -1237,3 +1241,15 @@ func (p *partitionProducer) canAddToQueue(sr *sendRequest) bool {
 	p.metrics.BytesPending.Add(float64(len(sr.msg.Payload)))
 	return true
 }
+
+func (p *partitionProducer) internalErrHandle(err error) {
+	if err == nil {
+		return
+	} else if errors.Is(err, internal.ErrExceedMaxMessageSize) {
+		p.log.WithError(errMessageTooLarge).
+			Errorf("internal err: %s", err)
+		p.metrics.PublishErrorsMsgTooLarge.Inc()
+		return
+	}
+	// other internal error can be handled here
+}

pulsar/producer_test.go

@@ -19,6 +19,7 @@ package pulsar
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -958,7 +959,7 @@ func TestMaxMessageSize(t *testing.T) {
 			Payload: payload,
 		})
 		if bias <= 0 {
-			assert.Error(t, err)
+			assert.Equal(t, true, errors.Is(err, internal.ErrExceedMaxMessageSize))
 			assert.Nil(t, ID)
 		} else {
 			assert.Equal(t, errMessageTooLarge, err)