Function NewBalanceStrategyRange

balance_strategy.go:126–146

-------------------------------------------------------------------- NewBalanceStrategyRange returns a range balance strategy, which is the default and assigns partitions as ranges to consumer group members. This follows the same logic as https://kafka.apache.org/31/javadoc/org/apache/kafka/clients/

()

//	M1: {T1: [0, 1, 2], T2: [0, 1, 2]}
//	M2: {T1: [3, 4, 5], T2: [3, 4, 5]}
func NewBalanceStrategyRange() BalanceStrategy {
	return &balanceStrategy{
		name: RangeBalanceStrategyName,
		coreFn: func(plan BalanceStrategyPlan, memberIDs []string, topic string, partitions []int32) {
			partitionsPerConsumer := len(partitions) / len(memberIDs)
			consumersWithExtraPartition := len(partitions) % len(memberIDs)

			sort.Strings(memberIDs)

			for i, memberID := range memberIDs {
				min := i*partitionsPerConsumer + int(math.Min(float64(consumersWithExtraPartition), float64(i)))
				extra := 0
				if i < consumersWithExtraPartition {
					extra = 1
				}
				max := min + partitionsPerConsumer + extra
				plan.Add(memberID, topic, partitions[min:max]...)
			}
		},
	}
}

// Deprecated: use NewBalanceStrategyRange to avoid data race issue
var BalanceStrategyRange = NewBalanceStrategyRange()