[FAB-1361] Move partitioner functions to own file

https://jira.hyperledger.org/browse/FAB-1361

Change-Id: If951bb6d76d96f7f519471b76c73b1ec01b86063
Signed-off-by: Kostas Christidis <[email protected]>

Author: kchristidis

orderer/kafka/partitioner.go

@@ -0,0 +1,42 @@
+/*
+Copyright IBM Corp. 2016 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+                 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kafka
+
+import "github.com/Shopify/sarama"
+
+// newStaticPartitioner returns a PartitionerConstructor that
+// returns a Partitioner that always chooses the specified partition.
+func newStaticPartitioner(partition int32) sarama.PartitionerConstructor {
+	return func(topic string) sarama.Partitioner {
+		return &staticPartitioner{partition}
+	}
+}
+
+type staticPartitioner struct {
+	partitionID int32
+}
+
+// Partition takes a message and partition count and chooses a partition.
+func (p *staticPartitioner) Partition(message *sarama.ProducerMessage, numPartitions int32) (int32, error) {
+	return p.partitionID, nil
+}
+
+// RequiresConsistency indicates to the user of the partitioner
+// whether the mapping of key->partition is consistent or not.
+func (p *staticPartitioner) RequiresConsistency() bool {
+	return true
+}

orderer/kafka/partitioner_test.go

@@ -0,0 +1,42 @@
+/*
+Copyright IBM Corp. 2016 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+                 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kafka
+
+import (
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/hyperledger/fabric/orderer/common/bootstrap/static"
+)
+
+func TestStaticPartitioner(t *testing.T) {
+	var partition int32 = 3
+	var numberOfPartitions int32 = 6
+
+	partitionerConstructor := newStaticPartitioner(partition)
+	partitioner := partitionerConstructor(static.TestChainID)
+
+	for i := 0; i < 10; i++ {
+		assignedPartition, err := partitioner.Partition(new(sarama.ProducerMessage), numberOfPartitions)
+		if err != nil {
+			t.Fatal("Partitioner not functioning as expected:", err)
+		}
+		if assignedPartition != partition {
+			t.Fatalf("Partitioner not returning the expected partition - expected %d, got %v", partition, assignedPartition)
+		}
+	}
+}

orderer/kafka/util.go

@@ -29,11 +29,15 @@ const (
 
 func newBrokerConfig(conf *config.TopLevel) *sarama.Config {
 	brokerConfig := sarama.NewConfig()
+
 	brokerConfig.Version = conf.Kafka.Version
+	// A partitioner is actually not needed the way we do things now,
+	// but we're adding it now to allow for flexibility in the future.
 	brokerConfig.Producer.Partitioner = newStaticPartitioner(conf.Kafka.PartitionID)
-	// set equivalent of kafka producer config max.request.bytes to the deafult
-	// value of a kafka server's socket.request.max.bytes property (100MiB).
+	// Set equivalent of kafka producer config max.request.bytes to the deafult
+	// value of a Kafka broker's socket.request.max.bytes property (100 MiB).
 	brokerConfig.Producer.MaxMessageBytes = int(sarama.MaxRequestSize)
+
 	return brokerConfig
 }
 
@@ -54,23 +58,3 @@ func newOffsetReq(conf *config.TopLevel, seek int64) *sarama.OffsetRequest {
 	req.AddBlock(conf.Kafka.Topic, conf.Kafka.PartitionID, seek, 1)
 	return req
 }
-
-// newStaticPartitioner returns a PartitionerConstructor that returns a Partitioner
-// that always chooses the specified partition.
-func newStaticPartitioner(partition int32) sarama.PartitionerConstructor {
-	return func(topic string) sarama.Partitioner {
-		return &staticPartitioner{partition}
-	}
-}
-
-type staticPartitioner struct {
-	partitionID int32
-}
-
-func (p *staticPartitioner) Partition(message *sarama.ProducerMessage, numPartitions int32) (int32, error) {
-	return p.partitionID, nil
-}
-
-func (p *staticPartitioner) RequiresConsistency() bool {
-	return true
-}

orderer/kafka/util_test.go

@@ -22,25 +22,6 @@ import (
 	"github.com/Shopify/sarama"
 )
 
-func TestStaticPartitioner(t *testing.T) {
-
-	var partition int32 = 3
-	var numberOfPartitions int32 = 6
-
-	partitionerConstructor := newStaticPartitioner(partition)
-	partitioner := partitionerConstructor(testConf.Kafka.Topic)
-
-	for i := 0; i < 10; i++ {
-		assignedPartition, err := partitioner.Partition(new(sarama.ProducerMessage), numberOfPartitions)
-		if err != nil {
-			t.Fatal(err)
-		}
-		if assignedPartition != partition {
-			t.Fatalf("Expected: %v. Actual: %v", partition, assignedPartition)
-		}
-	}
-}
-
 func TestProducerConfigMessageMaxBytes(t *testing.T) {
 
 	topic := testConf.Kafka.Topic