Adding a test.

lib/config/config.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strings"
 
 	"github.com/artie-labs/transfer/lib/stringutil"
 
@@ -48,6 +49,10 @@ type Kafka struct {
 	TopicConfigs    []*kafkalib.TopicConfig `yaml:"topicConfigs"`
 }
 
+func (k *Kafka) BootstrapServers() []string {
+	return strings.Split(k.BootstrapServer, ",")
+}
+
 type S3Settings struct {
 	OptionalPrefix     string                   `yaml:"optionalPrefix"`
 	Bucket             string                   `yaml:"bucket"`

lib/config/config_test.go

@@ -38,6 +38,32 @@ func TestBigQuery_DSN(t *testing.T) {
 	assert.Equal(t, "bigquery://project/eu/dataset", b.DSN())
 }
 
+func TestKafka_BootstrapServers(t *testing.T) {
+	type _tc struct {
+		bootstrapServerString    string
+		expectedBootstrapServers []string
+	}
+
+	tcs := []_tc{
+		{
+			bootstrapServerString:    "localhost:9092",
+			expectedBootstrapServers: []string{"localhost:9092"},
+		},
+		{
+			bootstrapServerString:    "a:9092,b:9093,c:9094",
+			expectedBootstrapServers: []string{"a:9092", "b:9093", "c:9094"},
+		},
+	}
+
+	for idx, tc := range tcs {
+		k := Kafka{
+			BootstrapServer: tc.bootstrapServerString,
+		}
+
+		assert.Equal(t, tc.expectedBootstrapServers, k.BootstrapServers(), idx)
+	}
+}
+
 func TestKafka_String(t *testing.T) {
 	k := Kafka{
 		BootstrapServer: "server",

processes/consumer/kafka.go

@@ -3,7 +3,6 @@ package consumer
 import (
 	"context"
 	"crypto/tls"
-	"strings"
 	"sync"
 	"time"
 
@@ -105,11 +104,9 @@ func StartConsumer(ctx context.Context) {
 				GroupID: settings.Config.Kafka.GroupID,
 				Dialer:  dialer,
 				Topic:   topic,
+				Brokers: settings.Config.Kafka.BootstrapServers(),
 			}
-			var brokers []string
-			brokers = append(brokers, strings.Split(settings.Config.Kafka.BootstrapServer, ",")...)
 
-			kafkaCfg.Brokers = brokers
 			kafkaConsumer := kafka.NewReader(kafkaCfg)
 			topicToConsumer.Add(topic, kafkaConsumer)
 			for {