Add producer package

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

pkg/kafka/config/config.go → pkg/kafka/config/producer/config.go

@@ -12,22 +12,22 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package config
+package producer
 
 import (
 	"github.com/Shopify/sarama"
 )
 
+// Builder builds a new kafka producer
+type Builder interface {
+	NewProducer() (sarama.AsyncProducer, error)
+}
+
 // Configuration describes the configuration properties needed to create a Kafka producer
 type Configuration struct {
 	Brokers []string
 }
 
-// ProducerBuilder builds a new kafka producer
-type ProducerBuilder interface {
-	NewProducer() (sarama.AsyncProducer, error)
-}
-
 // NewProducer creates a new asynchronous kafka producer
 func (c *Configuration) NewProducer() (sarama.AsyncProducer, error) {
 	saramaConfig := sarama.NewConfig()

plugin/storage/kafka/factory.go

@@ -23,7 +23,7 @@ import (
 	"github.com/uber/jaeger-lib/metrics"
 	"go.uber.org/zap"
 
-	"github.com/jaegertracing/jaeger/pkg/kafka/config"
+	"github.com/jaegertracing/jaeger/pkg/kafka/config/producer"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
@@ -35,9 +35,9 @@ type Factory struct {
 	metricsFactory metrics.Factory
 	logger         *zap.Logger
 
-	config     config.ProducerBuilder
 	producer   sarama.AsyncProducer
 	marshaller Marshaller
+	producer.Builder
 }
 
 // NewFactory creates a new Factory.
@@ -53,16 +53,16 @@ func (f *Factory) AddFlags(flagSet *flag.FlagSet) {
 // InitFromViper implements plugin.Configurable
 func (f *Factory) InitFromViper(v *viper.Viper) {
 	f.options.InitFromViper(v)
-	f.config = &f.options.config
+	f.Builder = &f.options.config
 }
 
 // Initialize implements storage.Factory
 func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger) error {
 	f.metricsFactory, f.logger = metricsFactory, logger
-	logger.Info("Kafka storage configuration",
-		zap.Any("producer config", f.config),
+	logger.Info("Kafka producer builder",
+		zap.Any("producer builder", f.Builder),
 		zap.Any("topic", f.options.topic))
-	p, err := f.config.NewProducer()
+	p, err := f.NewProducer()
 	if err != nil {
 		return err
 	}

plugin/storage/kafka/factory_test.go

@@ -25,7 +25,7 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/pkg/config"
-	kafkaConfig "github.com/jaegertracing/jaeger/pkg/kafka/config"
+	kafkaConfig "github.com/jaegertracing/jaeger/pkg/kafka/config/producer"
 	"github.com/jaegertracing/jaeger/storage"
 )
 
@@ -51,13 +51,13 @@ func TestKafkaFactory(t *testing.T) {
 	command.ParseFlags([]string{})
 	f.InitFromViper(v)
 
-	f.config = &mockProducerBuilder{
+	f.Builder = &mockProducerBuilder{
 		err: errors.New("made-up error"),
 		t:   t,
 	}
 	assert.EqualError(t, f.Initialize(metrics.NullFactory, zap.NewNop()), "made-up error")
 
-	f.config = &mockProducerBuilder{t: t}
+	f.Builder = &mockProducerBuilder{t: t}
 	assert.NoError(t, f.Initialize(metrics.NullFactory, zap.NewNop()))
 	assert.IsType(t, &protobufMarshaller{}, f.marshaller)
 
@@ -86,7 +86,7 @@ func TestKafkaFactoryEncoding(t *testing.T) {
 			command.ParseFlags([]string{"--kafka.encoding=" + test.encoding})
 			f.InitFromViper(v)
 
-			f.config = &mockProducerBuilder{t: t}
+			f.Builder = &mockProducerBuilder{t: t}
 			assert.NoError(t, f.Initialize(metrics.NullFactory, zap.NewNop()))
 			assert.IsType(t, test.marshaller, f.marshaller)
 		})
@@ -99,6 +99,6 @@ func TestKafkaFactoryMarshallerErr(t *testing.T) {
 	command.ParseFlags([]string{"--kafka.encoding=bad-input"})
 	f.InitFromViper(v)
 
-	f.config = &mockProducerBuilder{t: t}
+	f.Builder = &mockProducerBuilder{t: t}
 	assert.Error(t, f.Initialize(metrics.NullFactory, zap.NewNop()))
 }

plugin/storage/kafka/options.go

@@ -21,7 +21,7 @@ import (
 
 	"github.com/spf13/viper"
 
-	"github.com/jaegertracing/jaeger/pkg/kafka/config"
+	"github.com/jaegertracing/jaeger/pkg/kafka/config/producer"
 )
 
 const (
@@ -40,7 +40,7 @@ const (
 
 // Options stores the configuration options for Kafka
 type Options struct {
-	config   config.Configuration
+	config   producer.Configuration
 	topic    string
 	encoding string
 }
@@ -64,7 +64,7 @@ func (opt *Options) AddFlags(flagSet *flag.FlagSet) {
 
 // InitFromViper initializes Options with properties from viper
 func (opt *Options) InitFromViper(v *viper.Viper) {
-	opt.config = config.Configuration{
+	opt.config = producer.Configuration{
 		Brokers: strings.Split(v.GetString(configPrefix+suffixBrokers), ","),
 	}
 	opt.topic = v.GetString(configPrefix + suffixTopic)