Expose MarkOffset on PartitionConsumer

[gaffneyc]

I'm prototyping a worker that uses PartitionConsumers to ensure there is a Go routine running for each subscribed partition. Minor annoyance is that I need to pass a reference to the top level Consumer to the workers in order to mark offsets. If a version of MarkOffset was exposed on PartitionConsumer it would alleviate the need.

It looks like there is already MarkOffset(int64, string) on PartitionConsumer but it's not exposed as part of the interface and seems to expect msg.Offset + 1 which could cause confusion if exposed as is.

Rough sketch of what I'm doing

func main() {
  consumer, _ := cluster.NewConsumer(...)

  go func() {
    for part := range consumer.Partitions() {
      go worker(consumer, part)
    }
  }
}

func worker(consumer *cluster.Consumer, part cluster.PartitionConsumer) {
  for msg := part.Messages() {
    // process

    // If only this could be part.MarkOffset(msg, "")
    consumer.MarkOffset(msg, "")
  }
}

I'm also not sure if this approach is strictly necessary. So if I'm overthinking it I would appreciate some course correction.

[dim]

I looked at the code and I don't see an immediate problem with exposing it, but do you mind calling part.MarkOffset(msg.Offset, "") instead?

[gaffneyc]

I'm fine with calling part.MarkOffset(msg.Offset, "") the only concern on my end is that consumer.MarkOffset(msg, "") calls part.MarkOffset(msg.Offset + 1, "") see here.

Is msg.Offset + 1 correct in this case? Is MarkOffset marking the specified offset as completed or is it marking the next offset as where to resume from?

[dim]

Continuing in #209