consumer.go
func (c *Consumer) MarkOffsets(s *OffsetStash) {
	s.mu.Lock()
	defer s.mu.Unlock()

	for tp, info := range s.offsets {
		if sub := c.subs.Fetch(tp.Topic, tp.Partition); sub != nil {
			sub.MarkOffset(info.Offset, info.Metadata)
		}
		delete(s.offsets, tp)
	}
}
func (c *Consumer) MarkOffsets(s *OffsetStash) {
	s.mu.Lock()
	defer s.mu.Unlock()

	for tp, info := range s.offsets {
		if sub := c.subs.Fetch(tp.Topic, tp.Partition); sub != nil {
			sub.MarkOffset(info.Offset, info.Metadata)
		}
		delete(s.offsets, tp)
	}
}
// MarkOffsets marks stashed offsets as processed.
// See MarkOffset for additional explanation.