Take internalConsumer as a parameter

Signed-off-by: Davit Yeghshatyan <davo@uber.com>

Author: Davit Yeghshatyan

cmd/ingester/app/consumer/consumer.go

@@ -28,14 +28,15 @@ import (
 
 // Configuration stores the configurable options of a Consumer
 type Configuration struct {
-	kafkaConsumer.Configuration
+	Topic       string
 	Parallelism int
 }
 
 // Params are the parameters of a Consumer
 type Params struct {
-	BaseProcessor processor.SpanProcessor
-	Config        Configuration
+	BaseProcessor    processor.SpanProcessor
+	InternalConsumer kafkaConsumer.Consumer
+	Config           Configuration
 
 	MetricsFactory metrics.Factory
 	Logger         *zap.Logger
@@ -55,36 +56,21 @@ type Consumer struct {
 
 // New is a constructor for a Consumer
 func New(params Params) (*Consumer, error) {
-	consumerConfig := kafkaConsumer.Configuration{
-		Brokers: params.Config.Brokers,
-		Topic:   params.Config.Topic,
-		GroupID: params.Config.GroupID,
-	}
-	saramaConsumer, err := consumerConfig.NewConsumer()
-	if err != nil {
-		return nil, err
-	}
-
 	factoryParams := ProcessorFactoryParams{
-		Topic:          params.Config.Topic,
-		Parallelism:    params.Config.Parallelism,
-		SaramaConsumer: saramaConsumer,
-		BaseProcessor:  params.BaseProcessor,
-		Logger:         params.Logger,
-		Factory:        params.MetricsFactory,
+		Topic:            params.Config.Topic,
+		Parallelism:      params.Config.Parallelism,
+		InternalConsumer: params.InternalConsumer,
+		BaseProcessor:    params.BaseProcessor,
+		Logger:           params.Logger,
+		Factory:          params.MetricsFactory,
 	}
-	processorFactory, err := NewProcessorFactory(factoryParams)
-	if err != nil {
-		return nil, err
-	}
-
 	return &Consumer{
+		internalConsumer: params.InternalConsumer,
+		processorFactory: NewProcessorFactory(factoryParams),
 		metricsFactory:   params.MetricsFactory,
 		logger:           params.Logger,
 		close:            make(chan struct{}, 1),
 		isClosed:         sync.WaitGroup{},
-		internalConsumer: saramaConsumer,
-		processorFactory: processorFactory,
 	}, nil
 }
 

cmd/ingester/app/consumer/processor_factory.go

@@ -28,12 +28,12 @@ import (
 
 // ProcessorFactoryParams are the parameters of a ProcessorFactory
 type ProcessorFactoryParams struct {
-	Parallelism    int
-	Topic          string
-	BaseProcessor  processor.SpanProcessor
-	SaramaConsumer consumer.Consumer
-	Factory        metrics.Factory
-	Logger         *zap.Logger
+	Parallelism      int
+	Topic            string
+	BaseProcessor    processor.SpanProcessor
+	InternalConsumer consumer.Consumer
+	Factory          metrics.Factory
+	Logger           *zap.Logger
 }
 
 // ProcessorFactory is a factory for creating startedProcessors
@@ -47,15 +47,15 @@ type ProcessorFactory struct {
 }
 
 // NewProcessorFactory constructs a new ProcessorFactory
-func NewProcessorFactory(params ProcessorFactoryParams) (*ProcessorFactory, error) {
+func NewProcessorFactory(params ProcessorFactoryParams) *ProcessorFactory {
 	return &ProcessorFactory{
 		topic:          params.Topic,
-		consumer:       params.SaramaConsumer,
+		consumer:       params.InternalConsumer,
 		metricsFactory: params.Factory,
 		logger:         params.Logger,
 		baseProcessor:  params.BaseProcessor,
 		parallelism:    params.Parallelism,
-	}, nil
+	}
 }
 
 func (c *ProcessorFactory) new(partition int32, minOffset int64) processor.SpanProcessor {

cmd/ingester/app/options.go

@@ -22,7 +22,7 @@ import (
 
 	"github.com/spf13/viper"
 
-	"github.com/jaegertracing/jaeger/cmd/ingester/app/consumer"
+	kafkaConsumer "github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	"github.com/jaegertracing/jaeger/plugin/storage/kafka"
 )
 
@@ -46,8 +46,9 @@ const (
 
 // Options stores the configuration options for a Kafka consumer
 type Options struct {
-	consumer.Configuration
+	kafkaConsumer.Configuration
 	Encoding string
+	Parallelism int
 }
 
 // AddFlags adds flags for Options

cmd/ingester/main.go

@@ -29,6 +29,7 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/env"
 	"github.com/jaegertracing/jaeger/cmd/flags"
 	"github.com/jaegertracing/jaeger/cmd/ingester/app"
+	kafkaConsumer "github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	"github.com/jaegertracing/jaeger/cmd/ingester/app/consumer"
 	"github.com/jaegertracing/jaeger/cmd/ingester/app/processor"
 	"github.com/jaegertracing/jaeger/pkg/config"
@@ -92,11 +93,25 @@ func main() {
 			}
 			spanProcessor := processor.NewSpanProcessor(spParams)
 
+			consumerConfig := kafkaConsumer.Configuration{
+				Brokers: options.Brokers,
+				Topic:   options.Topic,
+				GroupID: options.GroupID,
+			}
+			saramaConsumer, err := consumerConfig.NewConsumer()
+			if err != nil {
+				logger.Fatal("Failed to create sarama consumer", zap.Error(err))
+			}
+
 			consumerParams := consumer.Params{
 				MetricsFactory: metricsFactory,
 				Logger:         logger,
 				BaseProcessor:  spanProcessor,
-				Config:         options.Configuration,
+				InternalConsumer: saramaConsumer,
+				Config:         consumer.Configuration{
+					 Topic: options.Topic,
+					 Parallelism: options.Parallelism,
+				},
 			}
 			spanConsumer, err := consumer.New(consumerParams)
 			if err != nil {