fix(retry): can not consume the messages in the retry topic (#437)

apache · Mar 3, 2020 · 2c78dae · 2c78dae
1 parent 2af5495
commit 2c78dae
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
@@ -22,6 +22,7 @@ import (
  "fmt"
  "math"
  "strconv"
+ "strings"
  "sync"
  "sync/atomic"
  "time"
@@ -803,6 +804,12 @@ func (pc *pushConsumer) consumeInner(ctx context.Context, subMsgs []*primitive.M
  }
 
  f, exist := pc.consumeFunc.Contains(subMsgs[0].Topic)
+
+ // fix lost retry message
+ if !exist && strings.HasPrefix(subMsgs[0].Topic, internal.RetryGroupTopicPrefix) {
+ f, exist = pc.consumeFunc.Contains(subMsgs[0].GetProperty(primitive.PropertyRetryTopic))
+ }
+
  if !exist {
  return ConsumeRetryLater, fmt.Errorf("the consume callback missing for topic: %s", subMsgs[0].Topic)
  }