scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-15 16:51:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

FB-1674: Kafka consumer stops reading messages from topic (#2222)

Browse source 
Added condition to first sort by topic, followed by partition and offset.
This commit is contained in:
rachithrr 2022-09-21 12:53:50 -05:00 • committed by Christopher Lowenthal
parent a8709baa8e
commit 6c21910291
1 changed files with 6 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
idk/kafka/source.go
View file

@ -216,19 +216,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)