From 0b1355f294ef9943fc46541abbe738a6cc5ecf2e Mon Sep 17 00:00:00 2001
From: Tayida Tapjinda <ttapjinda@gmail.com>
Date: Mon, 18 Jul 2016 20:19:19 +0700
Subject: [PATCH] Fix HashPartitioner's negative result by applying modulo
 before checking negative

---
 partitioner.go      |  7 ++++---
 partitioner_test.go | 17 +++++++++++++++++
 2 files changed, 21 insertions(+), 3 deletions(-)

diff --git a/partitioner.go b/partitioner.go
index 8c2bca413..892dce6db 100644
--- a/partitioner.go
+++ b/partitioner.go
@@ -112,10 +112,11 @@ func (p *hashPartitioner) Partition(message *ProducerMessage, numPartitions int3
 		return -1, err
 	}
 	hash := int32(p.hasher.Sum32())
-	if hash < 0 {
-		hash = -hash
+	partition := hash % numPartitions
+	if partition < 0 {
+		partition = -partition
 	}
-	return hash % numPartitions, nil
+	return partition, nil
 }
 
 func (p *hashPartitioner) RequiresConsistency() bool {
diff --git a/partitioner_test.go b/partitioner_test.go
index f44c509d6..3d391c59c 100644
--- a/partitioner_test.go
+++ b/partitioner_test.go
@@ -100,6 +100,23 @@ func TestHashPartitioner(t *testing.T) {
 	}
 }
 
+func TestHashPartitionerMinInt32(t *testing.T) {
+	partitioner := NewHashPartitioner("mytopic")
+
+	msg := ProducerMessage{}
+	// "1468509572224" generates 2147483648 (uint32) result from Sum32 function
+	// which is -2147483648 or int32's min value
+	msg.Key = StringEncoder("1468509572224")
+
+	choice, err := partitioner.Partition(&msg, 50)
+	if err != nil {
+		t.Error(partitioner, err)
+	}
+	if choice < 0 || choice >= 50 {
+		t.Error("Returned partition", choice, "outside of range for nil key.")
+	}
+}
+
 func TestManualPartitioner(t *testing.T) {
 	partitioner := NewManualPartitioner("mytopic")