lib/poseidon/consumer_group.rb
# Rebalance algorithm:
#
# * let CG be all consumers in the same group that consume topic T
# * let PT be all partitions producing topic T
# * sort CG
# * sort PT (so partitions on the same broker are clustered together)
# * let POS be our index position in CG and let N = size(PT)/size(CG)
# * assign partitions from POS*N to (POS+1)*N-1
Rebalance algorithm:
let CG be all consumers in the same group that consume topic T
let PT be all partitions producing topic T
sort CG
sort PT (so partitions on the same broker are clustered together)
let POS be our index position in CG and let N = size(PT)/size(CG)
assign partitions from POS*N to (POS+1)*N-1
Rebalance algorithm : let CG be all consumers in the same group that consume topic T let PT be all partitions producing topic T sort CG sort PT ( so partitions on the same broker are clustered together ) let POS be our index position in CG and let N = size ( PT ) / size ( CG ) assign partitions from POS * N to ( POS + 1 ) * N - 1
def rebalance!
      return if @pending
      @pending = true
      @mutex.synchronize do
        @pending = nil
        release_all!
        reload
        ids = zk.children(registries[:consumer], watch: true)
        pms = partitions
        rng = self.class.pick(pms.size, ids, id)
        pms[rng].each do |pm|
          if @pending
            release_all!
            break
          end
          consumer = claim!(pm.id)
          @consumers.push(consumer) if consumer
        end if rng
      end
    end
def rebalance! return if @pending @pending = true @mutex . synchronize do @pending = nil release_all! reload ids = zk . children ( registries [ :consumer ] , watch : true ) pms = partitions rng = self . class . pick ( pms . size , ids , id ) pms [ rng ] . each do | pm | if @pending release_all! break end consumer = claim! ( pm . id ) @consumers . push ( consumer ) if consumer end if rng end end
Rebalance algorithm:
 let CG be all consumers in the same group that consume topic T
 let PT be all partitions producing topic T
 sort CG
 sort PT (so partitions on the same broker are clustered together)
 let POS be our index position in CG and let N = size(PT)/size(CG)
 assign partitions from POS*N to (POS+1)*N-1
Rebalance algorithm : let CG be all consumers in the same group that consume topic T let PT be all partitions producing topic T sort CG sort PT ( so partitions on the same broker are clustered together ) let POS be our index position in CG and let N = size ( PT ) / size ( CG ) assign partitions from POS * N to ( POS + 1 ) * N - 1