Consumer async partitions

[eapache]

Feed consumer responses to the user asynchronously

Give each PartitionConsumer a goroutine whose job is to feed the user, and have
the brokerConsumer feed each partition with the response, then wait for acks.

This makes the consumer much more resilient to weird user consumption ordering,
as the old "one partition at a time" logic was very unforgiving.

Fixes #325
Replaces #398

@Shopify/kafka careful 👀 on this please

[eapache]

After several network errors and one (valid) race condition bug found, CI is finally green.

[wvanbergen]

I personally don't like using fallthrough like this, but I suppose it has always worked like this so no need to change.

[eapache]

The alternative is to duplicate these three lines twice (and the sendError call one additional time). Using fallthrough explicitly, (as opposed to the stupid implicit C version) seems much more elegant to me.

I don't want to derail the PR (as you mentioned, this code is just moved, not changed), but you've piqued my curiousity - what bothers you about this pattern?

[wvanbergen]

It's just way harder to parse in my mind, especially with the default case not be the last one. I'd prefer a bunch of if statements.

[eapache]

Thoughts on the third commit? It does make the behaviour less subtle, I think, though perhaps at the cost of another moving part...

[wvanbergen]

I like this approach better. Using a WaitGroup for waiting for a bunch of goroutines feels much more natural to me.

I am not a big fan of the extra bool field on PartitionConsumer. Maybe we should make the channel hold a wrapper type with a partitionConsumer and a bool, instead of adding a member to the partitionConsumer type?

(Rant: Go really should support multi-value channels so you can simply feed multi-value return values to a channel without having to wrap them in a shitty type.)

[eapache]

Maybe we should make the channel hold a wrapper type with a partitionConsumer and a bool, instead of adding a member to the partitionConsumer type?

Which channel?

[wvanbergen]

the acks channel

[eapache]

There is no acks channel anymore, it's a wait group o.o

[wvanbergen]

Argh, I am an idiot.

[eapache]

OK, this version removes the needsDispatch flag by switching the dying channel to carry actual useful values. Two side-effects:

• we can't close dying in the AsyncClose function since we might have an internal thread writing to it concurrently now; fortunately writing a nil value is just as good
• simplifies the switch which was bothering you, which means it's now just as easy to not use fallthrough, so do that

[wvanbergen]

I think this makes sense, but we should stress test this as well.

@shivnagarajan @thegedge, opinions on this?

[eapache]

I will merge this now to prep master for stress testing, but feel free to continue commenting @shivnagarajan and @thegedge.