refactored a bit

Author: timmyb32r

client.go

@@ -46,7 +46,7 @@ type Client struct {
 	// Limit for message size, applied after compression stage.
 	//
 	// If zero, no limit is applied.
-	MaxMessageSize int64
+	MaxMessageBytes int64
 }
 
 // A ConsumerGroup and Topic as these are both strings we define a type for

produce.go

@@ -159,13 +159,15 @@ func (c *Client) Produce(ctx context.Context, req *ProduceRequest) (*ProduceResp
 				},
 			}},
 		}},
-		MaxMessageSize_: c.MaxMessageSize,
+		MaxMessageBytes: c.MaxMessageBytes,
 	})
 
 	switch {
 	case err == nil:
 	case errors.Is(err, protocol.ErrNoRecord):
 		return new(ProduceResponse), nil
+	case protocol.IsMaxMessageBytesExceeded(err):
+		return nil, MessageTooLargeError{}
 	default:
 		return nil, fmt.Errorf("kafka.(*Client).Produce: %w", err)
 	}

protocol/error.go

@@ -1,6 +1,7 @@
 package protocol
 
 import (
+	"errors"
 	"fmt"
 )
 
@@ -89,3 +90,28 @@ func (e *TopicPartitionError) Error() string {
 func (e *TopicPartitionError) Unwrap() error {
 	return e.Err
 }
+
+type MaxMessageBytesExceededError interface {
+	error
+	IsMaxMessageBytesExceeded()
+}
+
+func IsMaxMessageBytesExceeded(err error) bool {
+	var target MaxMessageBytesExceededError
+	return errors.As(err, &target)
+}
+
+type baseMaxMessageBytesExceededError struct{ error }
+
+func (b baseMaxMessageBytesExceededError) IsFatal() {}
+
+func (b baseMaxMessageBytesExceededError) Unwrap() error {
+	return b.error
+}
+
+func NewBaseMaxMessageBytesExceededError(err error) error {
+	if err == nil {
+		return nil
+	}
+	return &baseMaxMessageBytesExceededError{error: err}
+}

protocol/produce/produce.go

@@ -16,13 +16,13 @@ type Request struct {
 	Timeout         int32          `kafka:"min=v0,max=v8"`
 	Topics          []RequestTopic `kafka:"min=v0,max=v8"`
 
-	// Use this to store max.message.size
-	MaxMessageSize_ int64 `kafka:"-"`
+	// Use this to store max.message.bytes
+	MaxMessageBytes int64 `kafka:"-"`
 }
 
 func (r *Request) ApiKey() protocol.ApiKey { return protocol.Produce }
 
-func (r *Request) MaxMessageSize() int64 { return r.MaxMessageSize_ }
+func (r *Request) MaxMessageBytesSize() int64 { return r.MaxMessageBytes }
 
 func (r *Request) Broker(cluster protocol.Cluster) (protocol.Broker, error) {
 	broker := protocol.Broker{ID: -1}

protocol/protocol.go

@@ -527,11 +527,11 @@ type Merger interface {
 	Merge(messages []Message, results []interface{}) (Message, error)
 }
 
-// MaxMessageSizeKeeper is an extension of the Message interface, which aimed
-// to store max.message.size parameter
-type MaxMessageSizeKeeper interface {
-	// Returns locally stored max.message.size value
-	MaxMessageSize() int64
+// MaxMessageBytesKeeper is an extension of the Message interface, which aimed
+// to store max.message.bytes parameter
+type MaxMessageBytesKeeper interface {
+	// Returns locally stored max.message.bytes value
+	MaxMessageBytesSize() int64
 }
 
 // Result converts r to a Message or an error, or panics if r could not be

protocol/request.go

@@ -127,9 +127,9 @@ func WriteRequest(w io.Writer, apiVersion int16, correlationID int32, clientID s
 	if err == nil {
 		messageSize := uint32(b.Size()) - 4
 
-		if p, _ := msg.(MaxMessageSizeKeeper); p != nil && p.MaxMessageSize() != 0 {
-			if messageSize > uint32(p.MaxMessageSize()) {
-				return fmt.Errorf("message size: %d exceeded max.message.size: %d", messageSize, p.MaxMessageSize())
+		if p, ok := msg.(MaxMessageBytesKeeper); ok && p.MaxMessageBytesSize() != 0 {
+			if messageSize > uint32(p.MaxMessageBytesSize()) {
+				return NewBaseMaxMessageBytesExceededError(fmt.Errorf("message size: %d exceeded max.message.bytes: %d", messageSize, p.MaxMessageBytesSize()))
 			}
 		}
 

writer.go

@@ -127,8 +127,8 @@ type Writer struct {
 	BatchBytes int64
 
 	// Setting this flag to true causes the WriteMessages starts to derive 'BatchBytes'
-	// from topic 'max.message.size' setting. If writer is used to write to multiple
-	// topics each topic 'max.message.size' will be handled appropriately.
+	// from topic 'max.message.bytes' setting. If writer is used to write to multiple
+	// topics each topic 'max.message.bytes' will be handled appropriately.
 	// This option simplifies maintaining of architecture - creates the one source of
 	// truth - topic settings on broker side
 	//
@@ -138,7 +138,7 @@ type Writer struct {
 	// Setting this flag to true causes the WriteMessages starts to apply 'BatchBytes'
 	// as limiting factor after compression stage.
 	// When this flag is false - it's possible to get case, when Value can exceed
-	// 'max.message.size' setting, but after compression it's less.
+	// 'max.message.bytes' setting, but after compression it's less.
 	// And WriteMessages returns an error, when indeed there are no error.
 	//
 	// Nevertheless, 'BatchBytes' also has second function - to form batches, and
@@ -647,15 +647,14 @@ func (w *Writer) WriteMessages(ctx context.Context, msgs ...Message) error {
 		return nil
 	}
 
+	balancer := w.balancer()
 	if w.AutoDeriveBatchBytes {
 		err := w.deriveBatchBytes(msgs)
 		if err != nil {
 			return err
 		}
 	}
 
-	balancer := w.balancer()
-
 	if !w.ApplyBatchBytesAfterCompression {
 		for i := range msgs {
 			n := int64(msgs[i].totalSize())
@@ -766,7 +765,9 @@ func (w *Writer) produce(key topicPartition, batch *writeBatch) (*ProduceRespons
 	defer cancel()
 
 	client := w.client(timeout)
-	client.MaxMessageSize = w.batchBytes(key.topic)
+	if w.ApplyBatchBytesAfterCompression {
+		client.MaxMessageBytes = w.batchBytes(key.topic)
+	}
 
 	return client.Produce(ctx, &ProduceRequest{
 		Partition:    int(key.partition),
@@ -860,7 +861,10 @@ func (w *Writer) deriveBatchBytes(msgs []Message) error {
 			return err
 		}
 
-		if _, ok := w.maxMessageBytesPerTopic.Load(topic); ok {
+		if res, ok := w.maxMessageBytesPerTopic.Load(topic); ok {
+			// DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG
+			fmt.Println("Writer::deriveBatchBytes::LOAD", "topic", topic, "res", res)
+			// DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG
 			continue
 		}
 
@@ -887,6 +891,9 @@ func (w *Writer) deriveBatchBytes(msgs []Message) error {
 		if err != nil {
 			return err
 		}
+		// DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG
+		fmt.Println("Writer::deriveBatchBytes::STORE", "topic", topic, "res", maxMessageBytesStr)
+		// DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG DEBUG
 		w.maxMessageBytesPerTopic.Store(topic, int64(maxMessageBytes))
 	}
 	return nil
@@ -897,16 +904,15 @@ func (w *Writer) batchBytes(topic string) int64 {
 		if result, ok := w.maxMessageBytesPerTopic.Load(topic); ok {
 			return result.(int64)
 		}
-
 		// batchBytes expects it's called after 'deriveBatchBytes(msgs)'
 		// It means, there are no unknown topics
 		panic(fmt.Sprintf("unknown topic: %s", topic))
-	} else {
-		if w.BatchBytes > 0 {
-			return w.BatchBytes
-		}
-		return 1048576
 	}
+
+	if w.BatchBytes > 0 {
+		return w.BatchBytes
+	}
+	return 1048576
 }
 
 func (w *Writer) batchTimeout() time.Duration {