Delivery failed after 5 min timeout for certain value sizes

[ivonindza]

Description

If the size of the value supplied to producer.produce() falls in the range of [999886-1000033] bytes (inclusive), the delivery will fail with _MSG_TIMED_OUT after 5 minutes (message.timeout.ms). For values with 999885 bytes and less messages are delivered normally, and for those with 1000034 bytes and more produce() returns MSG_SIZE_TOO_LARGE.

Messages stop being sent to the brokers until this 5 minute timeout expires. Producer buffer gets filled however. This problem can also affect a number of messages produced after the one with the "wrong" size, which will also time out.

How to reproduce

import sys
from confluent_kafka import Producer, KafkaException


SIZES = list(range(999886, 1000033+1))

def delivery_callback(err, msg):
    if err is not None:
        print(f'Delivery failed with {err}')

p = Producer({
    'bootstrap.servers': ...,
    'default.topic.config': {'acks': 'all'}
})

size = SIZES[0]
base_size = sys.getsizeof(b'')
val = b'0' * (size - base_size)

try:
    p.produce('test', val, on_delivery=delivery_callback)
except KafkaException as ex:
    print(f'Produce failed with {ex.args[0]}')
finally:
    p.flush()

Checklist

Please provide the following information:

• confluent-kafka-python and librdkafka version (confluent_kafka.version() and confluent_kafka.libversion()): ('0.11.5', 722176), ('0.11.5', 722431)
• Apache Kafka broker version: 2.11-2.0.0
• Client configuration: {...}
• Operating system: linux
• Provide client logs (with 'debug': '..' as necessary)
• Provide broker log excerpts
• Critical issue

[ivonindza]

Update: It seems the range given above is not constant. I've seen the same problem with different data and slightly smaller messages as well. I don't know what it might depend on.

[countergram]

Any word on this? We had this crop up in prod and it was very difficult to figure out where the timeouts were coming from until we found this issue.

[rnpridgeon]

Please set debug=broker,topic,msg in your producer config, run reproduce the issue and provide the resulting logs.

[ivonindza]

OK, here's the log. Trying to send a message of 999886 bytes. Using version 1.1.0 of confluent_kafka.
confluent-kafka.zip

[gesongzhe]

hello, I have encountered your problem before, just need to add a parameter to solve. Add parameters to the producer: "max.request.size". This parameter is used to set the size of the sent message.
Example：
p = Producer({
'bootstrap.servers': ...,
'default.topic.config': {'acks': 'all'},
'max.request.size': 1024 *1024 * 30
})

[ivonindza]

Well, limiting max message size is not the issue. There is the predefined max size and I can do it in my logic before the producer call as well. It's that specific message sizes have unexpected (buggy) behaviour. Also, in my particular case, I don't know or control the message size.

[rnpridgeon]

Thanks @ivonindza , it would seem that overhead associated with newer message format bloats your message just large enough to exceed message.max.bytes.

https://github.com/edenhill/librdkafka/blob/v1.1.0/src/rdkafka_msgset_writer.c#L851-L853
https://github.com/edenhill/librdkafka/blob/v1.1.0/src/rdkafka_msg.h#L146-L160
https://github.com/edenhill/librdkafka/blob/v1.1.0/src/rdkafka_proto.h#L484-L500

To demonstrate that this in fact the case I force the producer to use a lesser message format with the following properties:

 'api.version.request': False,
 'broker.version.fallback': '0.9.0',

With these additional properties I can successfully produce the message without issue. It's a bit unfortunate as I would expect this to raise a MSG_SIZE_TOO_LARGE error as it does when your raw message exceeds the limit.

I went to file an issue with librdkafka for this however it appears it's already been reported.

confluentinc/librdkafka#993

In the meantime I recommend setting message.max.bytes slightly larger than the largest value you expect to see. I was able to raise the default max.message.bytes value by 1 byte to get this to succeed. I suspect you were aware of this already though as the range implies you were walking your way up to the limit :)

I'll try to sync with @edenhill in the coming to weeks to see how we wish to proceed with librdkafka on this front.

[ivonindza]

I suspect you were aware of this already though as the range implies you were walking your way up to the limit :)

Yep. This one is easy to work around, but it was very hard to figure it out. It seemed nondeterministic at first and it resembles the situation when a broker disconnects for some reason (e.g. misconfigured zookeeper, not enough memory, etc.) and messages time out.