...
|
...
|
@@ -57,6 +57,9 @@ func (c *MessageConsumer) ConsumeClaim(groupSession sarama.ConsumerGroupSession, |
|
|
for message := range groupClaim.Messages() {
|
|
|
logs.Debug("Done Message claimed: timestamp = %v, topic = %s offset = %v value = %v \n",
|
|
|
message.Timestamp, message.Topic, message.Offset, string(message.Value))
|
|
|
for i := range c.beforeHandles {
|
|
|
c.beforeHandles[i](message)
|
|
|
}
|
|
|
groupSession.MarkMessage(message, "")
|
|
|
if topicHandle, err = c.FindTopichandle(groupClaim.Topic()); err != nil {
|
|
|
logs.Error("FindTopichandle err:%s \n", err)
|
...
|
...
|
@@ -65,6 +68,9 @@ func (c *MessageConsumer) ConsumeClaim(groupSession sarama.ConsumerGroupSession, |
|
|
if err = topicHandle(message); err != nil {
|
|
|
logs.Error("Message claimed: kafka消息处理错误 topic =", message.Topic, message.Offset, err)
|
|
|
}
|
|
|
for i := range c.beforeHandles {
|
|
|
c.afterHandles[i](message)
|
|
|
}
|
|
|
}
|
|
|
return nil
|
|
|
}
|
...
|
...
|
|