From 8b74087c5c73de53d7cda41789e028117eba7e7f Mon Sep 17 00:00:00 2001 From: rachithrr Date: Wed, 21 Sep 2022 12:53:50 -0500 Subject: [PATCH] FB-1674: Kafka consumer stops reading messages from topic (#2222) Added condition to first sort by topic, followed by partition and offset. --- idk/kafka/source.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/idk/kafka/source.go b/idk/kafka/source.go index 85b143c48..b276f4e7b 100644 --- a/idk/kafka/source.go +++ b/idk/kafka/source.go @@ -217,19 +217,23 @@ func (r *Record) Commit(ctx context.Context) error { } } sort.Slice(section, func(i, j int) bool { - if section[i].Partition != section[j].Partition { + if *section[i].Topic != *section[j].Topic { + return *section[i].Topic < *section[j].Topic + } else if section[i].Partition != section[j].Partition { return section[i].Partition < section[j].Partition } return section[i].Offset > section[j].Offset }) p := int32(-1) + s := "" r.src.highmarks = r.src.highmarks[:0] // sort by increasing partition, decreasing offset for _, x := range section { - if p != x.Partition { + if s != *x.Topic || p != x.Partition { r.src.highmarks = append(r.src.highmarks, x) } + s = *x.Topic p = x.Partition } committedOffsets, err := r.src.CommitMessages(r.src.highmarks)