fix: Kafka pending wrong when re-using pipeline

argoproj-labs · Jun 21, 2021 · 3c0d169 · 3c0d169
1 parent 9fefcb6
commit 3c0d169
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/runner/sidecar/sources.go b/runner/sidecar/sources.go
@@ -141,7 +141,7 @@ func connectKafkaSource(ctx context.Context, x *dfv1.Kafka, sourceName string, f
 		logger.Info("closing kafka admin client", "source", sourceName)
 		return adminClient.Close()
 	})
-	groupName := pipelineName + "-" + stepName
+	groupName := pipelineName + "-" + stepName + "-source-" + sourceName + "-" + x.Topic
 	group, err := sarama.NewConsumerGroup(x.Brokers, groupName, config)
 	if err != nil {
 		return fmt.Errorf("failed to create kafka consumer group: %w", err)