dapr
/
components-contrib
mirror of https://github.com/dapr/components-contrib.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Change marking message logic 

Signed-off-by: Deepanshu Agarwal <deepanshu.agarwal1984@gmail.com>
This commit is contained in:
Deepanshu Agarwal 2022-09-14 13:01:30 +05:30
parent e483dfe334
commit 31a38e5aaa
1 changed files with 7 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
internal/component/kafka/consumer.go
View File

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