WriteMessages in Async mode is blocking

[sagarkrkv]

Describe the bug

WriteMessages call should be non-blocking in async mode. But when there are connection issues with the Kafka broker, i.e when connections take too long establish or are slow, we observed that WriteMessages call that typically finishes in a microsecond, blocks for couple of minutes.

Kafka Version
Kafka 2.2

To Reproduce
It's hard to reproduce the exact scenario where this happens.

Expected behavior

WriteMessages call should drop messages to avoid degradation of system performance.

Additional context

We can either make the existing WriteMessages non-blocking, add a new config option to enable non-blocking mode, or add a new method to write messages in a non-blocking mode.

[achille-roussel]

@sagarkrkv you're correct, this could be better documented but once the internal queue is full the call blocks.

Dropping messages can be controversial, especially if it happens during a kafka outage. Backpressure can help control the amount of data loss that may occur, so it can be a preferable approach in some cases.

I'm not against changing the behavior, since it's not documented yet, I hope that no program depends on this implementation detail.

I'd love to get @stevevls's take as well on the topic as well.

In the mean time you can pass a context with a timeout to unblock the call.

[sagarkrkv]

Yes, we are currently adding a context with timeout.

But async and backpressure are a little contradictory, but I agree that this might be disruptive and unexpected change for users who have come to rely on this behavior. So if we want to minimize the disruption we can consider either

add a new config option to enable non-blocking mode or add a new method to write messages in a non-blocking mode.

[stevevls]

Coming late to this party. 😄 I think that if we add behavior to drop messages, it should definitely be guarded by a configuration flag that does not drop by default. Another configuration that we already have is the QueueCapacity setting that can be used to increase the amount of time until there is backpressure.

[sagarkrkv]

In a scenario where we have persistant connection issues with the Kafka cluster, or when the producer produces a large amount of messages due to a traffic spike, the QueueCapacity will not help a lot.

Is this something you're willing to accept a PR for? i.e add a config flag to make this non-blocking.

[illotum]

We encountered a similar issue in a different scenario. The Hash balancer combined with many hundreds partitions leads to the Writer.msgs bottlenecking due to the scheduler not keeping up draining it across all runners.

Frankly, you seem to be going for a very "proper" for Go design with nested runners and the hierarchy of channels, but it's counter productive in this particular case. AFAIU partitionWriters always belong to one and only one Writer, so running them in a go routine only adds scheduling overhead. I'd rather see Writer.run() append messages to batches directly, or call a corresponding partitionWriter func in sync. Very likely it's gonna be faster (if you dispatch the actual network write in a go routine).

Here's me raising the QueueCapacity to 60k, only to delay ineviatable:

[illotum]

Missed your v0.4 merge, sounds like this ticket can be closed. Is it resolved?

[gibsn]

Hi! Is this issue still present in the latest release?

[illotum]

@gibsn I verified, it no longer bottlenecks on the queue. Happily chugs along in my application.

[achille-roussel]

Thanks for following up everyone! Glad to hear 0.4 has been serving you well 👍

[gibsn]

@illotum I missed your reply 😮 thank you