func (c *Consumer) ResetOffset(msg *sarama.ConsumerMessage, metadata string) {
	if sub := c.subs.Fetch(msg.Topic, msg.Partition); sub != nil {
		sub.ResetOffset(msg.Offset, metadata)
	}
}
func (c *Consumer) ResetOffset(msg *sarama.ConsumerMessage, metadata string) {
	if sub := c.subs.Fetch(msg.Topic, msg.Partition); sub != nil {
		sub.ResetOffset(msg.Offset, metadata)
	}
}
// ResetOffset marks the provided message as processed, alongside a metadata string
// that represents the state of the partition consumer at that point in time. The
// metadata string can be used by another consumer to restore that state, so it
// can resume consumption.
//
// Difference between ResetOffset and MarkOffset is that it allows to rewind to an earlier offset