Fix kafka issue (#419)

buger · Jan 25, 2017 · e726e45 · e726e45
1 parent d66ba03
commit e726e45
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/input_kafka.go b/input_kafka.go
@@ -22,7 +22,7 @@ func NewKafkaInput(address string, config *KafkaConfig) *KafkaInput {
 
 	var con sarama.Consumer
 
-	if config.consumer.(*mocks.Consumer) != nil {
+	if mock, ok := config.consumer.(*mocks.Consumer); ok && mock != nil {
 		con = config.consumer
 	} else {
 		var err error

diff --git a/output_kafka.go b/output_kafka.go
@@ -26,7 +26,7 @@ func NewKafkaOutput(address string, config *KafkaConfig) io.Writer {
 
 	var producer sarama.AsyncProducer
 
-	if config.producer.(*mocks.AsyncProducer) != nil {
+	if mock, ok := config.producer.(*mocks.AsyncProducer); ok && mock != nil {
 		producer = config.producer
 	} else {
 		c.Producer.RequiredAcks = sarama.WaitForLocal