scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +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 GitHub
parent 91e3b8457a
commit 8b74087c5c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
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

@ -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)