allowing publisher to read from one project and publish to another

pubsub/kafka/config.go

@@ -4,6 +4,7 @@ import (
 	"strings"
 
 	"github.com/NYTimes/gizmo/config"
+	"github.com/Shopify/sarama"
 )
 
 // Config holds the basic information for working with Kafka.
@@ -18,6 +19,9 @@ type Config struct {
 	Topic     string `envconfig:"KAFKA_TOPIC"`
 
 	MaxRetry int `envconfig:"KAFKA_MAX_RETRY"`
+
+	// Config is a sarama config struct for more control over the underlying Kafka client.
+	Config *sarama.Config
 }
 
 // LoadConfigFromEnv will attempt to load an Kafka object

pubsub/kafka/kafka.go

@@ -35,9 +35,13 @@ func NewPublisher(cfg *Config) (pubsub.Publisher, error) {
 	}
 	p.topic = cfg.Topic
 
-	sconfig := sarama.NewConfig()
-	sconfig.Producer.Retry.Max = cfg.MaxRetry
-	sconfig.Producer.RequiredAcks = RequiredAcks
+	sconfig := cfg.Config
+	if sconfig == nil {
+		sconfig = sarama.NewConfig()
+		sconfig.Producer.Retry.Max = cfg.MaxRetry
+		sconfig.Producer.RequiredAcks = RequiredAcks
+	}
+	// we always want successes to return
 	sconfig.Producer.Return.Successes = true
 	p.producer, err = sarama.NewSyncProducer(cfg.BrokerHosts, sconfig)
 	return p, err
@@ -130,7 +134,11 @@ func NewSubscriber(cfg *Config, offsetProvider func() int64, offsetBroadcast fun
 	}
 	s.topic = cfg.Topic
 
-	sconfig := sarama.NewConfig()
+	sconfig := cfg.Config
+	if sconfig == nil {
+		sconfig = sarama.NewConfig()
+	}
+	// we always want to see errors, no matter what
 	sconfig.Consumer.Return.Errors = true
 	s.cnsmr, err = sarama.NewConsumer(cfg.BrokerHosts, sconfig)
 	return s, err