Producer refactor 3

[eapache]

@wvanbergen depends on #550, I'll rebase once that's merged.

This is the final step, it moves the request/response handling and all state into the "aggregator" (which is now named the brokerProducer). This makes what used to be the flusher really trivial, so it gets inlined into the constructor.

Most of the logic stays the same, the most confusing change is probably around forcing a blocking flush, because we now have to handle the case where we receive a response while we are waiting, and that changes our state (see the new waitForSpace method).

[eapache]

Rebased on master now that part 2 is merged. The diff is still pretty yucky, but I'm not sure how to split it any further. It may be easier to just read the brokerProducer code in the new version from scratch, without worrying about the old version.

[eapache]

Landing this will close #433 and should let me approach #294 without tearing my hair out (finally).

[wvanbergen]

👍

With regards to #433 - this doesn't really construct a ProduceRequest incrementally, because that is still done from scratch in produceSet.buildRequest. Can this be done incrementally as we add messages to the produceset?

[eapache]

because that is still done from scratch in produceSet.buildRequest

Not from scratch - the MessageSets are built incrementally. The request itself could be build incrementally for uncompressed requests, but for compressed requests we have to wait until we're ready in order to compress the body all at once, so that would be weird. If snappy lib provided an incremental stream compressor we might be able to do both incrementally, but it's not worth the effort at this point.

[eapache]

Still tearing my hair out over #294 - I am learning to hate this state machine