Producer: support per-topic `Producer.MaxMessageBytes` overrides

[gwwar]

Description

Sarama's producer sizes every (topic, partition) record batch against a single global Producer.MaxMessageBytes. Kafka's max.message.bytes is a per-topic configuration enforced by the broker, so a producer writing to topics with different caps can't align its batch sizing to each topic.

It's a batch cap, not a per-message cap. max.message.bytes is widely misread as a per-message limit. It caps the size of a whole record batch (the grouped, post-compression message set). From the Kafka topic config docs:

The largest record batch size allowed by Kafka (after compression if compression is enabled). [...] records are always grouped into batches for efficiency.

So a caller can size every individual record correctly and still get MESSAGE_TOO_LARGE, because the broker rejects the batch.

Where the global cap is used:

• produce_set.go wouldOverflow decides when to flush a (topic, partition) buffer using Producer.MaxMessageBytes.
• async_producer.go per-message size gate rejects any single message where size > Producer.MaxMessageBytes (returns a ConfigurationError).

Messages submitted before the next flush accumulate into the same (topic, partition) batch; call boundaries don't create batch boundaries. So even a caller that pre-chunks each submission to fit a topic's cap can have several correctly-sized chunks merged into one over-cap batch, gated only by the global value. Per-call chunk sizing alone cannot guarantee compliance, because sarama may merge chunks below the caller.

As an anecdotal data point: in a GitHub production service publishing to a topic with a broker max.message.bytes of ~80 KB (while the producer's global cap had to remain several MB), an in-app chunker sized every publish call to the 80 KB budget yet MESSAGE_TOO_LARGE never reached zero. The residual rejections clustered at 2-5 messages per rejected batch, which is consistent with individually valid chunks being merged into an over-cap batch rather than any single record exceeding the cap.

Why the existing knobs don't cover it:

• Lower Producer.MaxMessageBytes globally to the smallest topic's cap: the per-message gate then client-side rejects otherwise-valid larger records destined for higher-cap topics.
• One producer/client per cap: works, but requires routing topics to clients by configured cap, and forces N sarama.Clients when caps vary across many topics.
• Flush.Bytes / Flush.MaxMessages: Flush.Bytes is a flush trigger over the broker producer's total buffer, not a per-(topic, partition) byte limit. Flush.MaxMessages = 1 prevents multi-record batches but globally disables useful batching.

What I'd like to happen: an opt-in way to tell the producer a per-topic cap, so its batch-size checks bound each topic's batches to that topic's ceiling instead of only the global one. One possible shape:

config.Producer.MaxMessageBytes = 1_000_000
config.Producer.TopicMaxMessageBytes = map[string]int{
    "small-cap-topic": 81920, // matches this topic's broker max.message.bytes
}

• Topics absent from the map fall back to MaxMessageBytes. When unset, existing runtime behavior is unchanged; the override is opt-in and additive.
• Both wouldOverflow and the async per-message gate consult the resolved value, so sarama's existing size checks apply the per-topic value instead of the global one. This aligns sarama's estimates with each topic's cap rather than guaranteeing an exact post-compression size.
• Validate() rejects non-positive overrides and warns on any override >= MaxRequestSize, mirroring the existing MaxMessageBytes checks.

Additional context

• Related: Possible Message Batching Bug? #2797 reports the same batch-level failure pattern and notes that Flush.MaxMessages = 1 avoids it.
• Other clients don't solve this producer-side either: the Java producer (max.request.size) and librdkafka (message.max.bytes) both expose only a single global producer limit and recommend separate producer instances per cap (see librdkafka#3246). I'm not aware of a KIP governing per-topic producer-side sizing; broker enforcement of per-topic max.message.bytes is the only per-topic surface today.
• Code references above are against Sarama 3e29cc9573c6da854915f8c0952017eca622eec1 (main at the time of filing).
• I have a working branch with tests and can contribute a PR if this direction makes sense, but I'll gladly defer to the maintainers on the API shape or on whether this belongs in sarama.

---

AI assistance was used to help draft this issue; I've reviewed and verified the code references and behavior described.