Assign partitions based on lag

[Nevon]

In cases where the offset lag is unevenly spread across partitions, it makes sense to assign partitions across consumers based on the current offset lag, rather than completely round robin.

Essentially:

1. For each topic
   1.1. Calculate the offset lag for each partition in the topic
   1.2. Sort the partitions by offset lag in descending order
   1.3. For each partition
   1.3.1. Assign the partition to the consumer with the lowest number of assigned partitions (across all topics)
   1.3.2. If two or more consumers have the same number of assigned partitions, assign the partition to the one where the sum of the offset lag for the assigned partitions is the lowest

[Nevon]

One thing I'm not sure about is how to calculate the offset lag per partition. Maybe I'm missing something, but I don't see a way to get this using ListOffset, etc.

[tulios]

You will have to use broker#offsetFetch because you want the group offset and not the topic offset.

[Nevon]

But that just gives you the group offset, but to calculate the lag you'd also need the highwater mark and the starting offset.

[tulios]

It's a combination of listOffset and offsetFetch

[Nevon]

I have a PoC implementation now that seems to work. I'll create a PR with it as soon as I've dealt with some error cases, and get some feedback. The assignment algorithm itself seems to produce balanced assignments. It's mostly calculating the offset lag that's not the nicest at the moment.

I'm not sure it makes sense to bundle it with the main distribution of KafkaJS though, as for most people the round-robin assigner would be fine. Since the assigner is pluggable anyway, I think it might make more sense to publish it as a separate module.