kakfa: properly start at last offset when using a consumer group (#3629)

Author: blotus

go.mod

@@ -68,7 +68,7 @@ require (
 	github.com/prometheus/prom2json v1.3.0
 	github.com/r3labs/diff/v2 v2.14.1
 	github.com/sanity-io/litter v1.5.5
-	github.com/segmentio/kafka-go v0.4.45
+	github.com/segmentio/kafka-go v0.4.48
 	github.com/shirou/gopsutil/v3 v3.23.5
 	github.com/sirupsen/logrus v1.9.3
 	github.com/slack-go/slack v0.16.0

go.sum

@@ -632,8 +632,8 @@ github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/sanity-io/litter v1.5.5 h1:iE+sBxPBzoK6uaEP5Lt3fHNgpKcHXc/A2HGETy0uJQo=
 github.com/sanity-io/litter v1.5.5/go.mod h1:9gzJgR2i4ZpjZHsKvUXIRQVk7P+yM3e+jAF7bU2UI5U=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
-github.com/segmentio/kafka-go v0.4.45 h1:prqrZp1mMId4kI6pyPolkLsH6sWOUmDxmmucbL4WS6E=
-github.com/segmentio/kafka-go v0.4.45/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
+github.com/segmentio/kafka-go v0.4.48 h1:9jyu9CWK4W5W+SroCe8EffbrRZVqAOkuaLd/ApID4Vs=
+github.com/segmentio/kafka-go v0.4.48/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/sergi/go-diff v1.3.1 h1:xkr+Oxo4BOQKmkn/B9eMK0g5Kg/983T9DqqPHwYqD+8=
 github.com/sergi/go-diff v1.3.1/go.mod h1:aMJSSKb2lpPvRNec0+w3fl7LP9IOFzdc9Pa4NFbPK1I=

pkg/acquisition/modules/kafka/kafka.go

@@ -157,8 +157,12 @@ func (k *KafkaSource) Dump() any {
 }
 
 func (k *KafkaSource) ReadMessage(ctx context.Context, out chan types.Event) error {
-	// Start processing from latest Offset
-	k.Reader.SetOffsetAt(ctx, time.Now())
+	if k.Config.GroupID == "" {
+		err := k.Reader.SetOffset(kafka.LastOffset)
+		if err != nil {
+			return fmt.Errorf("while setting offset for reader on topic '%s': %w", k.Config.Topic, err)
+		}
+	}
 
 	for {
 		k.logger.Tracef("reading message from topic '%s'", k.Config.Topic)
@@ -279,6 +283,7 @@ func (kc *KafkaConfiguration) NewDialer() (*kafka.Dialer, error) {
 		}
 		dialer.TLS = tlsConfig
 	}
+
 	return dialer, nil
 }
 
@@ -297,6 +302,8 @@ func (kc *KafkaConfiguration) NewReader(dialer *kafka.Dialer, logger *log.Entry)
 
 	if kc.GroupID != "" {
 		rConf.GroupID = kc.GroupID
+		// kafka-go does not support calling SetOffset while using a consumer group
+		rConf.StartOffset = kafka.LastOffset
 	} else if kc.Partition != 0 {
 		rConf.Partition = kc.Partition
 	} else {