group consuming: fix commit bug
Noticed while watching __consumer_offsets; kafka doesn't reply to the
commit in the same order that we sent the request, so a simple fix is to
sort everything.
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"regexp"
+ "sort"
"sync"
"time"
@@ -165,8 +166,7 @@ type groupConsumer struct {
autocommitDisable bool
autocommitInterval time.Duration
- // TODO autocommit
- // OnLost (incremental)
+ // TODO OnLost (incremental)
}
// AssignGroup assigns a group to consume from, overriding any prior
@@ -769,6 +769,13 @@ func (g *groupConsumer) updateCommitted(
return
}
+ sort.Slice(req.Topics, func(i, j int) bool {
+ return req.Topics[i].Topic < req.Topics[j].Topic
+ })
+ sort.Slice(resp.Topics, func(i, j int) bool {
+ return resp.Topics[i].Topic < resp.Topics[j].Topic
+ })
+
for i := range resp.Topics {
reqTopic := &req.Topics[i]
respTopic := &resp.Topics[i]
@@ -779,6 +786,13 @@ func (g *groupConsumer) updateCommitted(
continue
}
+ sort.Slice(reqTopic.Partitions, func(i, j int) bool {
+ return reqTopic.Partitions[i].Partition < reqTopic.Partitions[j].Partition
+ })
+ sort.Slice(respTopic.Partitions, func(i, j int) bool {
+ return respTopic.Partitions[i].Partition < respTopic.Partitions[j].Partition
+ })
+
for i := range respTopic.Partitions {
reqPart := &reqTopic.Partitions[i]
respPart := &respTopic.Partitions[i]
