作者 唐旭辉

kafka 消息提交当前位移

... ... @@ -81,7 +81,7 @@ spec:
- name: KAFKA_HOST
value: ""
- name: KAFKA_CONSUMER_ID
value: "partnermg_test"
value: "partnermg_prd"
volumes:
- name: accesslogs
emptyDir: {}
... ...
... ... @@ -61,9 +61,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)
} else {
groupSession.MarkMessage(message, "")
}
groupSession.MarkMessage(message, "")
}
return nil
}
... ...