Frequent connection losses

[crepererum]

On the producer side, we see loads (~once per minute) connection losses in prod, like:

level=info msg="request encountered non-fatal error - backing off" e="Request error: Connection is poisened: Cannot read framed message: Cannot read data: early eof" request_name=produce backoff_secs=0 target="rskafka::backoff"

They are automatically retried, but it would be nice if we could prevent this from happening THAT often.

[tustvold]

The cancellation fixes in #113 appears to have fixed this 🎉

[crepererum]

I think it fixed the concrete usage when an aggregating producer is involved. However the underlying issue is this:

rskafka/src/messenger.rs

Line 333 in 39cd7d8

self.send_message(&buf).await?;

While the Messenger receives data using a background task, it sends data within fn request that might be canceled at any moment. This can result in half-send messages which would would mess up our message framing, so the server will likely close the connection (at least this is what I've often seen when we sent garbage to the server).

Long story short: Messenger::request needs to be cancelation-safe to fix this completely.

Note that #113 is still an important PR since it fixes cancelation-safety within the aggregated producer, which is required even when Messenger::request would be cancelation-safe.