Review fixes

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

Author: Davit Yeghshatyan

cmd/ingester/app/options.go

@@ -38,7 +38,7 @@ const (
 	encodingProto = "protobuf"
 
 	defaultBroker      = "127.0.0.1:9092"
-	defaultTopic       = "jaeger-ingester-spans"
+	defaultTopic       = "jaeger-spans"
 	defaultGroupID     = "jaeger-ingester"
 	defaultParallelism = 1000
 	defaultEncoding    = encodingProto

cmd/ingester/main.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"fmt"
+	"io"
 	"log"
 	"os"
 	"os/signal"
@@ -28,10 +29,10 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/env"
 	"github.com/jaegertracing/jaeger/cmd/flags"
 	"github.com/jaegertracing/jaeger/cmd/ingester/app"
-	spanconsumer "github.com/jaegertracing/jaeger/cmd/ingester/app/consumer"
+	"github.com/jaegertracing/jaeger/cmd/ingester/app/consumer"
 	"github.com/jaegertracing/jaeger/cmd/ingester/app/processor"
 	"github.com/jaegertracing/jaeger/pkg/config"
-	"github.com/jaegertracing/jaeger/pkg/kafka/consumer"
+	kafkaConsumer "github.com/jaegertracing/jaeger/pkg/kafka/consumer"
 	pMetrics "github.com/jaegertracing/jaeger/pkg/metrics"
 	"github.com/jaegertracing/jaeger/pkg/version"
 	"github.com/jaegertracing/jaeger/plugin/storage"
@@ -92,7 +93,7 @@ func main() {
 			}
 			spanProcessor := processor.NewSpanProcessor(spParams)
 
-			consumerConfig := consumer.Configuration{
+			consumerConfig := kafkaConsumer.Configuration{
 				Brokers: options.Brokers,
 				Topic:   options.Topic,
 				GroupID: options.GroupID,
@@ -102,39 +103,46 @@ func main() {
 				logger.Fatal("Failed to create sarama consumer", zap.Error(err))
 			}
 
-			factoryParams := spanconsumer.ProcessorFactoryParams{
+			factoryParams := consumer.ProcessorFactoryParams{
 				Topic:          options.Topic,
 				Parallelism:    options.Parallelism,
 				SaramaConsumer: saramaConsumer,
 				BaseProcessor:  spanProcessor,
 				Logger:         logger,
 				Factory:        metricsFactory,
 			}
-			processorFactory, err := spanconsumer.NewProcessorFactory(factoryParams)
+			processorFactory, err := consumer.NewProcessorFactory(factoryParams)
 			if err != nil {
 				logger.Fatal("Failed to create processor factory", zap.Error(err))
 			}
 
-			consumerParams := spanconsumer.Params{
+			consumerParams := consumer.Params{
 				InternalConsumer: saramaConsumer,
 				ProcessorFactory: *processorFactory,
 				Factory:          metricsFactory,
 				Logger:           logger,
 			}
-			kafkaConsumer, err := spanconsumer.New(consumerParams)
+			spanConsumer, err := consumer.New(consumerParams)
 			if err != nil {
 				logger.Fatal("Unable to set up consumer", zap.Error(err))
 			}
-			kafkaConsumer.Start()
+			spanConsumer.Start()
 
 			hc.Ready()
 			select {
 			case <-signalsChannel:
-				err := kafkaConsumer.Close()
+				logger.Info("Jaeger Ingester is starting to close")
+				err := spanConsumer.Close()
 				if err != nil {
-					logger.Error("Failed to close span writer", zap.Error(err))
+					logger.Error("Failed to close consumer", zap.Error(err))
 				}
-				logger.Info("Jaeger Ingester is finishing")
+				if closer, ok := spanWriter.(io.Closer); ok {
+					err := closer.Close()
+					if err != nil {
+						logger.Error("Failed to close span writer", zap.Error(err))
+					}
+				}
+				logger.Info("Jaeger Ingester has finished closing")
 			}
 			return nil
 		},