Close partition after ResetOffset

[e11it]

Hi!

I need to reset partition offset and only one working solution that I found is to close partition after reset.

Here is my working example:

config.Group.Mode = cluster.ConsumerModePartitions
...
for {
  select {
  ..
  case part, ok := <-consumer.Partitions():
            if !ok {
                return
            }
            if part.InitialOffset() != 0 {
                part.ResetOffset(-1, "")
                part.Close()
            }
            // start a separate goroutine to consume messages
            go func(pc cluster.PartitionConsumer) {
                for msg := range pc.Messages() {
                    log.Printf("%s/%d/%d\t%s\t%s\n", msg.Topic, msg.Partition, msg.Offset, msg.Key, msg.Value)
                    consumer.MarkOffset(msg, "") // mark message as processed
                }
            }(part)
  } // select
} // for

Am I do it right?

May be ResetOffset() function should close partition itself?

[dim]

@e11it sarama-cluster is about to be deprecated while I am trying to transition everyone over to my latest addition, directly to sarama - https://github.com/Shopify/sarama/pull/1099

[chandradeepak]

@dim what does sarama-cluster will be deprecated? Do we have to start using the new high level consumer API that is merged to sarama in future ? https://github.com/Shopify/sarama/releases/tag/v1.19.0

The new version has the support for the new consumer API. Is there an example on how to use this ?

Thanks for the new consumer API. I saw it is lot of discussion, design and work.

[PoacherBro]

@dim thanks for your great works, I have same question with @chandradeepak, when do we need to switch to sarama? And is there any suggestion for next usage, or any sample? I don't see the update in godoc/sarama. Wish your advise.

[dim]

Take a look at this https://godoc.org/github.com/Shopify/sarama#ConsumerGroup + example below

[PoacherBro]

@dim thanks a lot. So that means we can switch to sarama-1.19.0 now for consumer group rebalance & offset tracking? take over sarama-cluster completely, right?

[dim]

please note https://github.com/bsm/sarama-cluster#deprecation-notice