fix: Kafka should start at LastOffset by default

argoproj-labs · Jul 16, 2021 · 4cdf4bb · 4cdf4bb
1 parent af46d66
commit 4cdf4bb
Showing 1 changed file with 5 additions and 4 deletions.
diff --git a/runner/sidecar/source/kafka/kafka.go b/runner/sidecar/source/kafka/kafka.go
@@ -35,10 +35,11 @@ func New(ctx context.Context, pipelineName, stepName, sourceName string, x dfv1.
 		TLS:       t,
 	}
 	reader := kafka.NewReader(kafka.ReaderConfig{
-		Brokers: x.Brokers,
-		Dialer:  dialer,
-		GroupID: groupName,
-		Topic:   x.Topic,
+		Brokers:     x.Brokers,
+		Dialer:      dialer,
+		GroupID:     groupName,
+		Topic:       x.Topic,
+		StartOffset: kafka.LastOffset,
 	})
 	go wait.JitterUntil(func() {
 		ctx := context.Background()