...

As part of the client-side configuration for producers and consumers, the following property could be added: property max.response.size

It could be set as any other producer or consumer property:

Properties kafkaProducerConfigs = new Properties();
kafkaProducerConfigs.setProperty("socket.response.max.bytes", "104857600");
...
KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProducerConfigs);

...

...

 will be added.

Note that this property has a different purpose from fetch.max.bytes defined for consumers. The latter is used to control the size of batches to be received by a consumer, while the former applies to any message read from the socket. Also note fetch.max.bytes will not put a limit the size of the first batch received by a consumer.

Proposed Changes

This new element of configuration was originally suggested on the discussion thread and mirrors what already exists server-side, i.e. the property socket.request.max.bytes - which was actually added in brokers for the very same reason. 

...

• It does not modify the message decoding logic on the client. That is, the client is not trying to identify a failed SSL handshake by forward-reading a message. Indeed, doing so would affect the code path for all messages, thus would be likely introducing a better-avoided overhead;
• It introduces minimal change in the codebase. Because this functionality is already implemented server-side and the client uses the same network-level classes as the server, propagating the value of this new property is a minimal-impact, low-risk change. It is also fully transparent from a server perspective.
• It is intended to exhibit no change in behaviour for the existing clients, as long as the default value for the maximum allowed size is above that of any valid message (note: need to check this).

This proposal appears to come with multiple drawbacks, some of which may be considered as red flags.

Limited scope

The proposed change However, it cannot provide an definite guarantee against OOM in all scenarios - for instance, it will still manifest if the maximum size is set to 100 MB and the JVM is under memory pressure and have less than 100 MB of allocatable memory.

Potential duplication and overlap.

• The name max.response.size intends to mirror the existing max.request.size from the producer's configuration properties. However, max.request.size intends to check the size of producer records as provided by a client; while max.response.size is to check the size directly decoded from the network according to Kafka's binary protocol.
• On the broker, the property socket.request.max.bytes is used to validate the size of messages received by the server. The new property serves the same purpose, which introduces duplicated semantic, even if one property is characterised with the keyword "request" and the other with "response", reflecting the perspective adopted from either a client or a server.

Compatibility, Deprecation, and Migration Plan

...