feat(producer): support per-topic message size limits

async_producer.go

@@ -652,8 +652,14 @@ func (p *asyncProducer) dispatcher() {
 		}
 
 		size := msg.ByteSize(version)
-		if size > p.conf.Producer.MaxMessageBytes {
-			p.returnError(msg, ConfigurationError(fmt.Sprintf("Attempt to produce message larger than configured Producer.MaxMessageBytes: %d > %d", size, p.conf.Producer.MaxMessageBytes)))
+		maxMsgBytes := p.conf.producerMaxMessageBytes(msg.Topic)
+		if size > maxMsgBytes {
+			_, topicOverride := p.conf.Producer.TopicMaxMessageBytes[msg.Topic]
+			if topicOverride {
+				p.returnError(msg, ConfigurationError(fmt.Sprintf("Attempt to produce message larger than configured Producer.TopicMaxMessageBytes for topic %q: %d > %d", msg.Topic, size, maxMsgBytes)))
+			} else {
+				p.returnError(msg, ConfigurationError(fmt.Sprintf("Attempt to produce message larger than configured Producer.MaxMessageBytes: %d > %d", size, maxMsgBytes)))
+			}
 			continue
 		}
 

async_producer_test.go

@@ -61,6 +61,72 @@ func expectResults(t *testing.T, p AsyncProducer, successCount, errorCount int)
 	expectResultsWithTimeout(t, p, successCount, errorCount, 5*time.Minute)
 }
 
+func TestAsyncProducerMessageSizeLimitForTopic(t *testing.T) {
+	tests := []struct {
+		name          string
+		topic         string
+		configure     func(*Config)
+		expectedError string
+	}{
+		{
+			name:  "global limit fallback",
+			topic: "default-topic",
+			configure: func(config *Config) {
+				config.Producer.MaxMessageBytes = 1
+			},
+			expectedError: "Attempt to produce message larger than configured Producer.MaxMessageBytes:",
+		},
+		{
+			name:  "topic limit override",
+			topic: "small-topic",
+			configure: func(config *Config) {
+				config.Producer.MaxMessageBytes = 10_000
+				config.Producer.TopicMaxMessageBytes = map[string]int{
+					"small-topic": 1,
+				}
+			},
+			expectedError: `Attempt to produce message larger than configured Producer.TopicMaxMessageBytes for topic "small-topic":`,
+		},
+		{
+			name:  "topic limit override equal to global limit",
+			topic: "equal-topic",
+			configure: func(config *Config) {
+				config.Producer.MaxMessageBytes = 1
+				config.Producer.TopicMaxMessageBytes = map[string]int{
+					"equal-topic": 1,
+				}
+			},
+			expectedError: `Attempt to produce message larger than configured Producer.TopicMaxMessageBytes for topic "equal-topic":`,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			config := NewTestConfig()
+			config.Producer.Return.Errors = true
+			test.configure(config)
+
+			producer := &asyncProducer{
+				conf:   config,
+				input:  make(chan *ProducerMessage),
+				errors: make(chan *ProducerError, 1),
+				txnmgr: &transactionManager{},
+				muter:  newPartitionMuter(),
+			}
+
+			go producer.dispatcher()
+			producer.input <- &ProducerMessage{
+				Topic: test.topic,
+				Value: StringEncoder(TestMessage),
+			}
+			close(producer.input)
+
+			producerError := <-producer.errors
+			require.ErrorContains(t, producerError.Err, test.expectedError)
+		})
+	}
+}
+
 func TestBrokerProducerSequencing(t *testing.T) {
 	type batchID struct {
 		epoch    int16

config.go

@@ -4,8 +4,10 @@ import (
 	"crypto/tls"
 	"fmt"
 	"io"
+	"maps"
 	"net"
 	"regexp"
+	"slices"
 	"time"
 
 	"github.com/klauspost/compress/gzip"
@@ -193,6 +195,12 @@ type Config struct {
 		// will still be rejected.
 		// This value must remain smaller than sarama.MaxRequestSize.
 		MaxMessageBytes int
+		// TopicMaxMessageBytes overrides MaxMessageBytes for individual topics.
+		// Each value is the maximum permitted size of a message and the
+		// pre-compression size estimate of a batch for a single partition of the
+		// corresponding topic. Topics absent from the map use MaxMessageBytes.
+		// The map must not be modified after the client is created.
+		TopicMaxMessageBytes map[string]int
 		// The level of acknowledgement reliability needed from the broker (defaults
 		// to WaitForLocal). Equivalent to the `request.required.acks` setting of the
 		// JVM producer.
@@ -651,6 +659,15 @@ func (c *Config) Validate() error {
 	if c.Producer.MaxMessageBytes >= int(MaxRequestSize) {
 		Logger.Println("Producer.MaxMessageBytes must be smaller than MaxRequestSize; it will be ignored.")
 	}
+
+	topics := slices.Sorted(maps.Keys(c.Producer.TopicMaxMessageBytes))
+	for _, topic := range topics {
+		maxMsgBytes := c.Producer.TopicMaxMessageBytes[topic]
+		if maxMsgBytes >= int(MaxRequestSize) {
+			Logger.Printf("Producer.TopicMaxMessageBytes[%q] must be smaller than MaxRequestSize; it will be ignored.\n", topic)
+		}
+	}
+
 	if c.Producer.Flush.Bytes >= int(MaxRequestSize) {
 		Logger.Println("Producer.Flush.Bytes must be smaller than MaxRequestSize; it will be ignored.")
 	}
@@ -804,6 +821,12 @@ func (c *Config) Validate() error {
 		return ConfigurationError("Producer.Retry.Backoff must be >= 0")
 	}
 
+	for _, topic := range topics {
+		if c.Producer.TopicMaxMessageBytes[topic] <= 0 {
+			return ConfigurationError(fmt.Sprintf("Producer.TopicMaxMessageBytes[%q] must be > 0", topic))
+		}
+	}
+
 	if c.Producer.Compression == CompressionLZ4 && !c.Version.IsAtLeast(V0_10_0_0) {
 		return ConfigurationError("lz4 compression requires Version >= V0_10_0_0")
 	}
@@ -938,6 +961,13 @@ func (c *Config) Validate() error {
 	return nil
 }
 
+func (c *Config) producerMaxMessageBytes(topic string) int {
+	if maxMsgBytes, ok := c.Producer.TopicMaxMessageBytes[topic]; ok {
+		return maxMsgBytes
+	}
+	return c.Producer.MaxMessageBytes
+}
+
 func (c *Config) getDialer() proxy.Dialer {
 	if c.Net.Proxy.Enable {
 		Logger.Println("using proxy")

config_test.go

@@ -349,6 +349,16 @@ func TestProducerConfigValidates(t *testing.T) {
 			},
 			"Producer.MaxMessageBytes must be > 0",
 		},
+		{
+			"TopicMaxMessageBytes",
+			func(cfg *Config) {
+				cfg.Producer.TopicMaxMessageBytes = map[string]int{
+					"z-topic": 0,
+					"a-topic": -1,
+				}
+			},
+			`Producer.TopicMaxMessageBytes["a-topic"] must be > 0`,
+		},
 		{
 			"RequiredAcks",
 			func(cfg *Config) {
@@ -467,6 +477,24 @@ func TestProducerConfigValidates(t *testing.T) {
 	}
 }
 
+func TestProducerMaxMessageBytesForTopic(t *testing.T) {
+	config := NewTestConfig()
+	config.Producer.MaxMessageBytes = 1000
+
+	assert.Equal(t, 1000, config.producerMaxMessageBytes("default-topic"))
+
+	config.Producer.TopicMaxMessageBytes = map[string]int{
+		"small-topic": 100,
+		"large-topic": 2000,
+	}
+
+	assert.Equal(t, 100, config.producerMaxMessageBytes("small-topic"))
+	assert.Equal(t, 2000, config.producerMaxMessageBytes("large-topic"))
+	assert.Equal(t, 1000, config.producerMaxMessageBytes("default-topic"))
+
+	assert.NoError(t, config.Validate())
+}
+
 func TestConsumerConfigValidates(t *testing.T) {
 	tests := []struct {
 		name string

produce_set.go

@@ -330,7 +330,7 @@ func (ps *produceSet) wouldOverflow(msg *ProducerMessage) bool {
 		return true
 	// Would we overflow the size-limit of a message-batch for this partition?
 	case ps.msgs[msg.Topic] != nil && ps.msgs[msg.Topic][msg.Partition] != nil &&
-		ps.msgs[msg.Topic][msg.Partition].bufferBytes+msg.ByteSize(version) >= ps.parent.conf.Producer.MaxMessageBytes:
+		ps.msgs[msg.Topic][msg.Partition].bufferBytes+msg.ByteSize(version) >= ps.parent.conf.producerMaxMessageBytes(msg.Topic):
 		return true
 	// Would we overflow simply in number of messages?
 	case ps.parent.conf.Producer.Flush.MaxMessages > 0 && ps.bufferCount >= ps.parent.conf.Producer.Flush.MaxMessages:

produce_set_test.go

@@ -88,6 +88,50 @@ func TestProduceSetAddingMessagesOverflowBytesLimit(t *testing.T) {
 	}
 }
 
+func TestProduceSetAddingMessagesOverflowTopicBytesLimit(t *testing.T) {
+	parent, ps := makeProduceSet()
+	parent.conf.Producer.MaxMessageBytes = 10_000
+	parent.conf.Producer.TopicMaxMessageBytes = map[string]int{
+		"small-topic": 1000,
+	}
+
+	msg := &ProducerMessage{
+		Topic: "small-topic",
+		Key:   StringEncoder(TestMessage),
+		Value: StringEncoder(TestMessage),
+	}
+
+	for ps.bufferBytes+msg.ByteSize(2) < parent.conf.Producer.TopicMaxMessageBytes[msg.Topic] {
+		if ps.wouldOverflow(msg) {
+			t.Error("set shouldn't fill up before the topic-specific limit")
+		}
+		safeAddMessage(t, ps, msg)
+	}
+
+	if !ps.wouldOverflow(msg) {
+		t.Error("set should be full at the topic-specific limit")
+	}
+}
+
+func TestProduceSetTopicBytesLimitOverridesGlobalLimit(t *testing.T) {
+	parent, ps := makeProduceSet()
+	msg := &ProducerMessage{
+		Topic: "large-topic",
+		Key:   StringEncoder(TestMessage),
+		Value: StringEncoder(TestMessage),
+	}
+	msgSize := msg.ByteSize(2)
+	parent.conf.Producer.MaxMessageBytes = 2 * msgSize
+	parent.conf.Producer.TopicMaxMessageBytes = map[string]int{
+		msg.Topic: 4 * msgSize,
+	}
+
+	safeAddMessage(t, ps, msg)
+	if ps.wouldOverflow(msg) {
+		t.Error("topic-specific limit should replace a smaller global limit")
+	}
+}
+
 func TestProduceSetPartitionTracking(t *testing.T) {
 	_, ps := makeProduceSet()