Method assignPartitionWithLoad

pkg/consistent/consistent.go:178–220

assignPartitionWithLoad distributes a partition to a member based on the load factor. It ensures that no member exceeds the calculated average load while distributing partitions. If the distribution fails due to insufficient capacity, it panics with an error message.

(
	partitionID, startIndex int,
	partitionAssignments map[int]*Member,
	memberLoads map[string]float64,
)

// It ensures that no member exceeds the calculated average load while distributing partitions.
// If the distribution fails due to insufficient capacity, it panics with an error message.
func (c *Consistent) assignPartitionWithLoad(
	partitionID, startIndex int,
	partitionAssignments map[int]*Member,
	memberLoads map[string]float64,
) {
	// Calculate the average load to determine the maximum load a member can handle.
	averageLoad := c.calculateAverageLoad()
	var attempts int

	// Iterate to find a suitable member for the partition.
	for {
		attempts++

		// If the loop exceeds the number of members, it indicates that the partition
		// cannot be distributed with the current configuration.
		if attempts >= len(c.sortedSet) {
			panic("not enough capacity to distribute partitions: consider decreasing the partition count, increasing the member count, or increasing the load factor")
		}

		// Get the current hash value from the sorted set.
		currentHash := c.sortedSet[startIndex]

		// Retrieve the member associated with the hash value.
		currentMember := *c.ring[currentHash]

		// Check the current load of the member.
		currentLoad := memberLoads[currentMember.String()]

		// If the member's load is within the acceptable range, assign the partition.
		if currentLoad+1 <= averageLoad {
			partitionAssignments[partitionID] = &currentMember
			memberLoads[currentMember.String()]++
			return
		}

		// Move to the next member in the sorted set.
		startIndex++
		if startIndex >= len(c.sortedSet) {
			// Loop back to the beginning of the sorted set if we reach the end.
			startIndex = 0
		}
	}
}

// distributePartitions evenly distributes partitions among members while respecting the load factor.
// It ensures that partitions are assigned to members based on consistent hashing and load constraints.