Change marking message logic
@@ -17,7 +17,6 @@ import (
"context"
"errors"
"fmt"
- "sort"
"strconv"
"sync"
"time"
@@ -118,21 +117,18 @@ func (consumer *consumer) doBulkCallback(session sarama.ConsumerGroupSession, me
Entries: messageValues,
}
responses, err := handler(session.Context(), &event)
- sort.SliceStable(responses, func(i, j int) bool {
- return responses[i].EntryID < responses[j].EntryID
- })
if err != nil {
for i, resp := range responses {
- id, e := strconv.Atoi(resp.EntryID)
- if e != nil {
- id = i
- }
+ if resp.EntryID == messageValues[i].EntryID {
if resp.Error == nil {
- session.MarkMessage(messages[id], "")
+ session.MarkMessage(messages[i], "")
} else {
break
}
+ } else {
+ return errors.New("entry id mismatch while processing bulk messages")
+ }
}
} else {
for _, message := range messages {
