Recover from Kafka::MessageSizeTooLarge exception in async producer.

[RyanAmos]

You can get into a bad unrecoverable state when a Kafka::MessageSizeTooLarge exception is raised.

When we receive this exception, if we're using the sync producer, then we simply re-raise this exception. If we're using the async producer, then we attempt to remove the largest message from the buffer and simply retry. Once retried, if the buffer is still too large, we'll continue removing the largest message until we can successfully flush the buffer.

This will re-raise the exception since it is done synchronously:

DeliveryBoy.deliver('a' * 1_500_000, topic: 'test')

This will remove the largest message and continue to retry:

DeliveryBoy.deliver_async('a' * 1_500_000, topic: 'test')

Would love to get some feedback on this approach. Specifically @dasch since you had some opinions on the approach here: https://github.com/zendesk/ruby-kafka/pull/358

[dorner]

Excited for this - it's been bothering me for a while!

[dasch]

Also, ❤️

[github-actions[bot]]

Pull request has been marked as stale due to a lack of activity.

[RyanAmos]

@dasch I dropped the ball here. Are you still interested in getting this in? If so, I'll rebase and add the functional test as asked in the next couple weeks.

[dasch]

Sure!