[Issue 240] Add check for max message size (apache#263)

* Add check for max message size

1. When creating a connection, try to get maxMessageSize from handshake
response command. If it's not set, then use the default maxMessageSize
value defined in the client side.
2. When sending a message, check whether the size of payload exceeds
maxMessageSize. If so, return error immediately without adding this
meesage into sending queue.
3. To implement these, I made some tiny modifications in Connection
interface and added a field in its implementation struct.

* Add testing for max message size

* Fix error log

integration-tests/standalone.conf

@@ -83,6 +83,9 @@ statusFilePath=/usr/local/apache/htdocs
 # Using a value of 0, is disabling unackeMessage limit check and consumer can receive messages without any restriction
 maxUnackedMessagesPerConsumer=50000
 
+# Set maxMessageSize to 1MB rather than the default value 5MB for testing
+maxMessageSize=1048576
+
 ### --- Authentication --- ###
 
 # Enable TLS

pulsar/internal/batch_builder.go

@@ -29,8 +29,6 @@ import (
 )
 
 const (
-	// MaxMessageSize limit message size for transfer
-	MaxMessageSize = 5 * 1024 * 1024
 	// MaxBatchSize will be the largest size for a batch sent from this particular producer.
 	// This is used as a baseline to allocate a new buffer that can hold the entire batch
 	// without needing costly re-allocations.

pulsar/internal/commands.go

@@ -30,8 +30,12 @@ import (
 
 const (
 	// MaxFrameSize limit the maximum size that pulsar allows for messages to be sent.
-	MaxFrameSize        = 5 * 1024 * 1024
-	magicCrc32c  uint16 = 0x0e01
+	MaxFrameSize = 5 * 1024 * 1024
+	// MessageFramePadding is for metadata and other frame headers
+	MessageFramePadding = 10 * 1024
+	// MaxMessageSize limit message size for transfer
+	MaxMessageSize        = MaxFrameSize - MessageFramePadding
+	magicCrc32c    uint16 = 0x0e01
 )
 
 // ErrCorruptedMessage is the error returned by ReadMessageData when it has detected corrupted data.

pulsar/internal/connection.go

@@ -71,6 +71,7 @@ type Connection interface {
 	AddConsumeHandler(id uint64, handler ConsumerHandler)
 	DeleteConsumeHandler(id uint64)
 	ID() string
+	GetMaxMessageSize() int32
 	Close()
 }
 
@@ -157,6 +158,8 @@ type connection struct {
 
 	tlsOptions *TLSOptions
 	auth       auth.Provider
+
+	maxMessageSize int32
 }
 
 func newConnection(logicalAddr *url.URL, physicalAddr *url.URL, tlsOptions *TLSOptions,
@@ -282,7 +285,13 @@ func (c *connection) doHandshake() bool {
 			cmd.Type)
 		return false
 	}
-
+	if cmd.Connected.MaxMessageSize != nil {
+		c.log.Debug("Got MaxMessageSize from handshake response:", *cmd.Connected.MaxMessageSize)
+		c.maxMessageSize = *cmd.Connected.MaxMessageSize
+	} else {
+		c.log.Debug("No MaxMessageSize from handshake response, use default: ", MaxMessageSize)
+		c.maxMessageSize = MaxMessageSize
+	}
 	c.log.Info("Connection is ready")
 	c.changeState(connectionReady)
 	return true
@@ -749,3 +758,7 @@ func (c *connection) consumerHandler(id uint64) (ConsumerHandler, bool) {
 func (c *connection) ID() string {
 	return fmt.Sprintf("%s -> %s", c.cnx.LocalAddr(), c.cnx.RemoteAddr())
 }
+
+func (c *connection) GetMaxMessageSize() int32 {
+	return c.maxMessageSize
+}

pulsar/producer_partition.go

@@ -40,7 +40,10 @@ const (
 	producerClosed
 )
 
-var errFailAddBatch = errors.New("message send failed")
+var (
+	errFailAddBatch    = errors.New("message send failed")
+	errMessageTooLarge = errors.New("message size exceeds MaxMessageSize")
+)
 
 type partitionProducer struct {
 	state  int32
@@ -236,6 +239,16 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 
 	msg := request.msg
 
+	// if msg is too large
+	if len(msg.Payload) > int(p.cnx.GetMaxMessageSize()) {
+		p.publishSemaphore.Release()
+		request.callback(nil, request.msg, errMessageTooLarge)
+		p.log.WithField("size", len(msg.Payload)).
+			WithField("properties", msg.Properties).
+			WithError(errMessageTooLarge).Error()
+		return
+	}
+
 	deliverAt := msg.DeliverAt
 	if msg.DeliverAfter.Nanoseconds() > 0 {
 		deliverAt = time.Now().Add(msg.DeliverAfter)

pulsar/producer_test.go

@@ -790,3 +790,30 @@ func TestDelayAbsolute(t *testing.T) {
 	assert.NotNil(t, msg)
 	canc()
 }
+
+func TestMaxMessageSize(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newTopicName(),
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+	defer producer.Close()
+	serverMaxMessageSize := 1024 * 1024
+	for bias := -1; bias <= 1; bias++ {
+		payload := make([]byte, serverMaxMessageSize+bias)
+		ID, err := producer.Send(context.Background(), &ProducerMessage{
+			Payload: payload,
+		})
+		if bias <= 0 {
+			assert.NoError(t, err)
+			assert.NotNil(t, ID)
+		} else {
+			assert.Equal(t, errMessageTooLarge, err)
+		}
+	}
+}