Rebalance doesn't work reliably fix

[LMnet]

This is a fix for #532

Some words about expected behavior. fs2 is a pull-based library: stream will not push the new step until the previous step is completed. So, if we have a flatMap on a stream, the next step will be executed only when the current inner stream completes. Projecting this onto our situation, if we have a code like this:

consumer.partitionsMapStream.flatMap { assignment =>
  val innerStream = handle(assignment)
  innerStream
} 

the next assignment will be received only when the innerStream finished. If innerStream is based on partition streams from the assignment, it will automatically work: on rebalance, old streams are finished, and innerStream is finished too. And this is the expected behavior.

But sometimes not all old partition streams finished. And that's why the rebalance doesn't work. I added a test case for this, which is failed on the current stable branch.

Actually, I didn't found a root cause for this issue. Mostly because the current solution for the finishing old partition streams is pretty complex, and also distributed across multiple files (KafkaConsumer and KafkaConsumerActor). I decided to try to change and simplify old partition streams completion logic. And it worked.

In my implementation I don't try to track all partition streams with partitionStreamId. I just interrupt old partition streams on rebalance. Actually, this is a lot closer to the kafka ideology: when the partition revoked, I interrupt old streams. When the partition assigned, I create new streams for each partition. It turns out that having Ref[F, Deferred[F, Unit]] is enough for this. On each rebalance I created a new Deferred and saved it in Ref, and at the same time, I completed the old Deferred. Each partition stream in a current assignment subscribed on the Deferred from the current assignment, and when it completes, streams interrupts. All logic is in the one method partitionsMapStream.

Because of a new logic for closing old partition streams functionality with tracking partitionStreamId is redundant now. So, I deleted it. I think it removes a bit of the inner complexity of a library.

[bplommer]

Thanks for this - I'll take a look later today.

[LMnet]

It's still work in progress, but the root cause is fixed

[LMnet]

@vlovgr @bplommer it's ready for the review

[bplommer]

This looks good as far as I can tell but I'd like to get @vlovgr's thoughts too.