[JamesMGreene]

TL;DR

Add optional per-topic overrides for producer message and batch size checks while preserving the global fallback behavior.

• Closes Producer: support per-topic Producer.MaxMessageBytes overrides #3697

Problem

Sarama currently uses one global Producer.MaxMessageBytes value for two jobs:

• rejecting an individual message that is too large
• deciding when a (topic, partition) batch should roll over

Kafka's max.message.bytes can differ by topic. When one producer writes to topics with different limits, the global setting either allows an oversized batch for a smaller topic or rejects a valid message for a larger one.

This PR adds the opt-in per-topic override proposed in #3697.

What changed

• Add Producer.TopicMaxMessageBytes map[string]int.
• Use the topic override for both individual-message validation and per-partition batch rollover.
• Fall back to Producer.MaxMessageBytes when no override exists, preserving existing behavior by default.
• Allow overrides to be smaller or larger than the global value.
• Reject non-positive overrides deterministically and warn for values at or above MaxRequestSize.
• Keep the existing global-limit error text, with a topic-specific error when an override applies.
• Document that sizing is a pre-compression estimate and that the configuration map must not change after client creation.

config.Producer.MaxMessageBytes = 4 * 1024 * 1024
config.Producer.TopicMaxMessageBytes = map[string]int{
    "small-cap-topic": 80 * 1024,
}

This does not discover broker configuration or guarantee an exact post-compression batch size. The caller supplies the topic limits, and Sarama applies them to the same local size checks it already performs.

Testing

• Added coverage for global fallback, topic-specific rejection, smaller batch limits, larger overrides, and deterministic validation.

Note

AI assistance was used to help generate the new tests. I reviewed the test cases and verified that the full test suite passes.

[JamesMGreene]

@dnwe Would greatly appreciate a review so I can at least judge if this might get merged, or if I need to address this separately within my own codebase instead. Thank you! 🙇🏻

[JamesMGreene]

@puellanivis Thank you for the review! I've made some updates based on your feedback. Would greatly appreciate a re-review when you get a chance. 🙇🏻

[puellanivis]

Really just the one thing I noticed.

[JamesMGreene]

@puellanivis Before I make any more changes, I'm hitting a bit of a personal design crisis that could use your guidance. 🙏🏻

This PR is currently designed to allow the Topic-level overrides to be either smaller OR larger than the Producer-level config. This means it's treating the Producer-level config as more of a default/fallback limit rather than an absolute maximum limit.

Does that seem like the most desirable approach for this library, or would it be better if the Topic-level overrides could only be smaller than (<=) the Producer-level config? 🤔

Either way works fine for my particular use case (so far, anyway), so please advise what would be best for the library. Thank you! 🙇🏻

[puellanivis]

I don’t have any feedback or bias on which way we should go: either enforce “must be less than” or “can be anything”. We just want the warning message and the enforcement to align.

I suppose one could look at the Java implementation(s)(?) to look at what the correct design choice here should be. But Personally, either works from a code quality perspective (which is where my code reviews usually focus).

[JamesMGreene]

I didn't spot a specific implementation of this in the Java client, but all of the Kafka documentation (and their AI chatbot) seem to suggest that the topic-level configuration is indeed an override of the default, rather than only being allowed to reduce the size.

From a session with their AI chatbot:

A topic’s max.message.bytes can be configured independently, but producers and brokers must still be able to handle the resulting record-batch size. Prefer setting a larger limit only for the specific topic that needs it rather than increasing the broker-wide default.

• message.max.bytes is the broker-level default maximum record-batch size.
• max.message.bytes is the per-topic maximum record-batch size and defaults to the broker’s message.max.bytes.
• Therefore, a topic can override that default with a larger value.
• Producers must set max.request.size large enough to send the batch.

Supporting documentation:

• https://docs.confluent.io/platform/current/kafka/post-deployment.html#handling-large-message-sizes
• https://developer.confluent.io/faq/apache-kafka/concepts/#apache-kafka-concepts-what-is-kafkas-max-message-size
• https://support.confluent.io/hc/en-us/articles/360024032811-How-to-handle-large-messages-with-Kafka (requires login)