Fixed compatibility with the sarama hash partitioner (segmentio#336)

Previously, if the 32 bit hash code had its most significant bit set, kafka-go
and sarama would select different partitions.

Author: Steve van Loben Sels

balancer.go

@@ -141,7 +141,7 @@ type Hash struct {
 	lock sync.Mutex
 }
 
-func (h *Hash) Balance(msg Message, partitions ...int) (partition int) {
+func (h *Hash) Balance(msg Message, partitions ...int) int {
 	if msg.Key == nil {
 		return h.rr.Balance(msg, partitions...)
 	}
@@ -161,12 +161,14 @@ func (h *Hash) Balance(msg Message, partitions ...int) (partition int) {
 	}
 
 	// uses same algorithm that Sarama's hashPartitioner uses
-	partition = int(hasher.Sum32()) % len(partitions)
+	// note the type conversions here.  if the uint32 hash code is not cast to
+	// an int32, we do not get the same result as sarama.
+	partition := int32(hasher.Sum32()) % int32(len(partitions))
 	if partition < 0 {
 		partition = -partition
 	}
 
-	return
+	return int(partition)
 }
 
 type randomBalancer struct {

balancer_test.go

@@ -40,6 +40,13 @@ func TestHashBalancer(t *testing.T) {
 			Partitions: []int{0, 1, 2},
 			Partition:  1,
 		},
+		// in a previous version, this test would select a different partition
+		// than sarama's hash partitioner.
+		"hash code with MSB set": {
+			Key:        []byte("20"),
+			Partitions: []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15},
+			Partition:  1,
+		},
 	}
 
 	for label, test := range testCases {