max_buffer_bytesize for async_producer

[alan-sapaad]

I am able to set max_buffer_bytesize for a synchronous producer and as expected I am receiving the error Kafka::BufferOverflow: Cannot produce to topic, max buffer bytesize (960000 bytes) reached

But I need to use an async_producer with delivery_interval set to 50ms (0.05). I've set max_buffer_bytesize when initializing the producer. But I am able to push as many messages as possible to the Queue and all I can see is the queue size getting bigger every second (even if I don't produce any message). I am guessing it must be the [:deliver_message, nil] packets that are being pushed into the queue every 50ms.

I am not able to receive any errors from the underlying sync producer. How do I make sure that the package size is under 1MB? I am using Heroku Kafka and they will not let any packages over 1MB. I want the async_producer.produce to error out if the buffer size will be more then 950kb with the current message added.

Thanks!

[dasch]

You need to configure the queue in addition to the buffer – but there should already be an upper bound on the queue size. The async producer runs in a different thread than the calling code, so you cannot rely on knowing the current buffer size, since the producer will be trying to flush that buffer concurrently with your calling code.

[dasch]

I don't think writing to Redis is inherently more reliable than writing to Kafka. So you're now stuck with two pieces of infrastructure that may fail. Personally, I would not consider using Sidekiq for this, and I think it's terrible advice.

[alan-sapaad]

yeah, makes sense.

I am able to get the queue length by doing producer[:instance].instance_variable_get(:@queue).size. But like I said, if the queue is clogged, the size grows rapidly, but not with actual messages but with the flags [:deliver_message, nil], because we have set the delivery_interval to 120ms.

1. Is there a way to avoid adding this flag to the queue, if the previous message is already a flag?
2. Or is there a way to get the queue size excluding these flags?

[dasch]

There's no thread safe way to do either.

I would also say: a delivery_interval of 120ms is really low. Have you tried adjusting that to at least 1s?

[dasch]

There might be a different design for the async producer that uses another mechanism for signaling that the delivery interval has been reached, but I'm not going to have the time to look into it any time soon.

Have you looked at using ruby-rdkafka for this? I have been meaning to migrate DeliveryBoy from ruby-kafka to ruby-rdkafka in order to have a high level API, but, yeah, I'm off doing non-Ruby, non-Kafka things, and don't have enough time to look into this stuff anymore.

[alan-sapaad]

@dasch we do have time-sensitive transactional data flowing through the system as well. We thought 120ms is like a sweet spot to have the message delivery latency as low as possible and have a hand full of messages in the batch. The faster we get the messages on the other end, the better.

No, I have not looked into ruby-rdkafka yet. I think I'm going to try it out.

Thanks again!