Allow marking of earlier offsets #197
Conversation
|
I believe it might be a good idea to combine these with changes from @aravindvs #179 |
consumer.go
Outdated
| // | ||
| // Difference between ResetOffset and MarkOffset is that it allows to rewind to an earlier offset | ||
| func (c *Consumer) ResetOffset(msg *sarama.ConsumerMessage, metadata string) { | ||
| c.subs.Fetch(msg.Topic, msg.Partition).MarkOffset(msg.Offset+1, metadata) |
There was a problem hiding this comment.
Shouldn't that call .ResetOffset(...)
|
@dim I have fixed as per the comment. You were right. Additionally, added some tests. Thank you. |
jerrygb
force-pushed
the
issue-reset-offsets
branch
from
6f93d62
to
3b23565
Compare
This commit is in alignment with the sarama library commits IBM/sarama@b966238#diff-3ca6d659defd100fe2de43adf2b8f41e IBM/sarama@96fa1c8#diff-3ca6d659defd100fe2de43adf2b8f41e
jerrygb
force-pushed
the
issue-reset-offsets
branch
from
3b23565
to
4d12ba4
Compare
|
@dim reopening for travis rebuild. Last build failed, due to seemingly okay https://travis-ci.org/bsm/sarama-cluster/jobs/322296975
|
cluster_test.go
Outdated
| @@ -21,6 +21,7 @@ var ( | |||
| testKafkaRoot = "kafka_2.12-1.0.0" | |||
| testKafkaAddrs = []string{"127.0.0.1:29092"} | |||
| testTopics = []string{"topic-a", "topic-b"} | |||
| testTopicsReset = "topic-c" | |||
There was a problem hiding this comment.
also, no real need to declare this as global, it's only used once
There was a problem hiding this comment.
True and formatted all the files. Sorry about that.
Removed globals assuming we dont need another fuzzing test.
cluster_test.go
Outdated
| @@ -73,6 +74,10 @@ var _ = BeforeSuite(func() { | |||
| "-name", "kafkaServer", "kafka.Kafka", | |||
| testDataDir("server.properties"), | |||
| ) | |||
| if _, err := os.Stat(testKafkaData); err == nil { | |||
There was a problem hiding this comment.
No need for that clause, instead just the below will do:
Expect(os.RemoveAll(testKafkaData)).To(Succeed())
consumer_test.go
Outdated
| @@ -315,6 +340,94 @@ var _ = Describe("Consumer", func() { | |||
| Expect(uniques).To(HaveLen(15000)) | |||
| }) | |||
|
|
|||
| It("should consume/commit/reset/resume", func() { | |||
There was a problem hiding this comment.
so I am not really sure we need another fuzzing test for this. I don't see the added value. what you have done in https://github.com/bsm/sarama-cluster/pull/197/files#diff-e85d03976d9766c4d00ac176511d4f06R188 seems enough, please remove that test.
There was a problem hiding this comment.
Removed the test. This was meant to check that we do actually rewind and, whether the first offset consumed is actually the one we expected. The earlier fuzzing test actually doesnt deal specifically with start offsets, right?
Rather it deals with consuming atleast fixed number of unique entries (partition/offset pairs)?
I have removed it as you advised. But, let me know if we need check the start offsets when we start consumption.
|
Thank you @dim. Fixed as per comments. Also removed the additional fuzzing test. This was meant to check whether we do actually rewind properly and check for the the first offset consumed. I suppose the earlier fuzzing test actually deal specifically with start offsets, right? For now, I have removed it as you advised. But, let me know if you need to me add those back in. |
This commit is in alignment with the sarama library commits to reset offsets
IBM/sarama@b966238#diff-3ca6d659defd100fe2de43adf2b8f41e
IBM/sarama@96fa1c8#diff-3ca6d659defd100fe2de43adf2b8f41e