sarama committed offsets don't match java consumer offsets

[kanekv]

When we consume everything from kafka topic, kafka's tool (bin/kafka-consumer-groups.sh) still shows lag of 1 message. If we run java consumer with the same group it consumes 1 remaining message (which was already consumed by sarama consumer).

Does sarama use own offset schema (one less than java)?

Versions

Sarama Version: master
Kafka Version: 0.10
Go Version: 1.6.2

[eapache]

What are you using to commit offsets? Sarama does not commit offsets by default.

[kanekv]

Using PartitionOffsetManager, committing manually.

[eapache]

Huh, dug into this a bit more and found https://github.com/apache/kafka/blob/fa32545442ef6724aa9fb5f4e0e269b0c873288f/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L287-L288.

So in Java-land, you are expected to commit last-message-offset + 1 and then start from that. In sarama we always assumed that you would commit last-message-offset directly and then add +1 to get your next starting point (see https://github.com/Shopify/sarama/blob/366453c3e32f055595f4d5269d8f39aac5fdcdbb/offset_manager.go#L343).

I'm not sure what the correct solution is. We probably shouldn't be adding one in NextOffset - should we instead be adding one in MarkOffset or should we just document it and leave that to the user?

[kanekv]

@eapache I think we should follow what java consumer does, that would potentially allow to mix/interchange consumers, also kafka's built-in reporting tools would look better.

[dtjm]

I think it might be worthwhile to note that this is somewhat of a major API change. If someone recompiles their Kafka consumer with this patch and restarts it, they would end up with 1 duplicate message per partition.

Nevermind, this is already discussed in #713.