Merge pull request #53 from redpanda-data/nv/repeater-validate-group-…

…name

repeater: validate group name

cmd/kgo-repeater/main.go

@@ -115,12 +115,9 @@ func main() {
 		topicsList = strings.Split(*topic, ",")
 	}
 
-	wConfig := worker.NewWorkerConfig(
-		"kgo", *brokers, *trace, topicsList[0], *linger, *maxBufferedRecords, *useTransactions, *compressionType, *compressiblePayload, *username, *password, *enableTls)
-	opts := wConfig.MakeKgoOpts()
-	opts = append(opts, []kgo.Opt{
-		kgo.ProducerBatchMaxBytes(1024 * 1024),
-	}...)
+	if *group == "" {
+		panic("A consumer group must be provided via the -group flag")
+	}
 
 	dataInFlightPerWorker := (*initialDataMb * 1024 * 1024) / uint64(*workers)
 

pkg/worker/repeater/repeater_worker.go

@@ -276,10 +276,6 @@ func (v *Worker) Init() {
 			kgo.ConsumerGroup(v.config.Group),
 		}...)
 
-		if v.config.Group != "" {
-			opts = append(opts, kgo.ConsumerGroup(v.config.Group))
-		}
-
 		client, err := kgo.NewClient(opts...)
 		util.Chk(err, "unable to initialize client: %v", err)
 		v.client = client