choose random partition if none is set

sandglass · Dec 2, 2017 · a427d12 · a427d12
1 parent 3ff3e1e
commit a427d12
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 1 deletion.
diff --git a/broker/broker_produce.go b/broker/broker_produce.go
@@ -28,7 +28,7 @@ func (b *Broker) Produce(ctx context.Context, req *sgproto.ProduceMessageRequest
 			return nil, fmt.Errorf("unknown partition '%s'", req.Partition)
 		}
 	} else { // choose one
-		p = t.ChoosePartition(req.Messages[0]) // FIXME: choose randomly
+		p = t.ChooseRandomPartition() // FIXME: choose randomly
 	}
 
 	leader := b.getPartitionLeader(req.Topic, p.Id)

diff --git a/topic/topic.go b/topic/topic.go
@@ -1,6 +1,7 @@
 package topic
 
 import (
+	"math/rand"
 	"path/filepath"
 	"sync"
 
@@ -107,6 +108,11 @@ func (t *Topic) ChoosePartition(msg *sgproto.Message) *Partition {
 	return t.Partitions[idx]
 }
 
+func (t *Topic) ChooseRandomPartition() *Partition {
+	idx := rand.Int() % len(t.Partitions)
+	return t.Partitions[idx]
+}
+
 func (t *Topic) PutMessage(partition string, msg *sgproto.Message) error {
 	var p *Partition
 	if partition != "" {