Arroyo provides little insight into what is actually being stored in input/output buffers

[untitaker]

While working on #270 I noticed that with certain transformation functions, the returned bytestring did not get stored in the output buffer at all. It's not clear to me why this happens.

I noticed a pattern where returning a plain b"x" * 20 did not get stored in the output buffer, but wrapping it in a KafkaPayload like the tests do in #270 do would store the bytestring out of band.

This behavior could explain some mysterious performance regressions we had in the past (output buffer not being used). Hopefully the new output buffer metrics will provide insight into this. If that metric is much lower than input batch size in bytes, we have a problem in that area.

We also don't know how much data is being transmitted in-band via pool.submit vs what is being sent over the buffer. Right now we only emit metrics for the size of the out-of-band buffer, not whatever else we pickle.