use default channel size from sarama in kafka package

lovoo · Jan 4, 2018 · 0baf3ad · 0baf3ad
1 parent 00d2946
commit 0baf3ad
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 3 deletions.
diff --git a/kafka/config.go b/kafka/config.go
@@ -12,7 +12,6 @@ func CreateDefaultSaramaConfig(clientID string, partitioner sarama.PartitionerCo
 
 	config.Version = sarama.V0_10_1_0
 	config.ClientID = clientID
-	config.ChannelBufferSize = defaultChannelBufferSize
 
 	// consumer configuration
 	config.Consumer.Return.Errors = true

diff --git a/kafka/consumer.go b/kafka/consumer.go
@@ -9,7 +9,7 @@ import (
 
 const (
 	// size of sarama buffer for consumer and producer
-	defaultChannelBufferSize = 10
+	defaultChannelBufferSize = 256
 
 	// time sarama-cluster assumes the processing of an event may take
 	defaultMaxProcessingTime = 1 * time.Second
@@ -52,7 +52,11 @@ type saramaConsumer struct {
 
 // NewSaramaConsumer creates a new Consumer using sarama
 func NewSaramaConsumer(brokers []string, group string, config *cluster.Config) (Consumer, error) {
-	events := make(chan Event, defaultChannelBufferSize)
+	chsize := config.Config.ChannelBufferSize
+	if chsize == 0 {
+		chsize = defaultChannelBufferSize
+	}
+	events := make(chan Event, chsize)
 
 	g, err := newGroupConsumer(brokers, group, events, config)
 	if err != nil {