Allow the Consumer to disable auto-commit offsets #1164
Conversation
|
Hi, I also need to disable the autocommit but I need to commit when I want and at closing. |
|
Given the new consumer groups one also needs to escape in the Close() of the offset manager as it requires MainLoop to run... |
|
@taiyang-li I'll merge upstream and see if I is compatible with the latest tag, 1.23.1 - if so, I'll reissue this PR. |
ghost
added
the
…p if commit offsets are disabled
|
cla is signed. im unsure why only one of the travis builds fail, and not all.. |
|
hey @kjelle is it feasible to write a test for it? |
I am adding tests to check if setting the |
…mit.Enable. Renamed Consumer.Offsets.CommitInteval to Consumer.Offsets.AutoCommit.Inteval
|
@varun06 I discovered some problems while writing the unit-tests. Updates are in kjelle:master repo. Tests are good. I will run it in our production line to see how it behaves. It should not differ. |
…lushToBroker to keep mainLoop
|
@kjelle That's why I asked for test :) Thanks for doing that and finding the issue. |
|
@sam-obeid can you please have a look again? |
|
@varun06 given the amount of commits I would suggest a merge squash when it comes to that :) |
|
Done, Please test and let me know how it goes. |
This reverts commit 72a629d.
| @@ -338,8 +338,15 @@ type Config struct { | |||
| // offsets. This currently requires the manual use of an OffsetManager | |||
| // but will eventually be automated. | |||
| Offsets struct { | |||
| // How frequently to commit updated offsets. Defaults to 1s. | |||
| CommitInterval time.Duration | |||
There was a problem hiding this comment.
ohh shit, this was a breaking change, i.e: it breaks people using sarama-cluster with the latest version of sarama.
../../../../go/pkg/mod/github.com/bsm/[email protected]+incompatible/consumer.go:452:59: c.client.config.Config.Consumer.Offsets.CommitInterval undefined (type struct { AutoCommit struct { Enable bool; Interval time.Duration }; Initial int64; Retention time.Duration; Retry struct { Max int } } has no field or method CommitInterval)
fix breaking API change added in #1164 Fixes: - bsm/sarama-cluster#308 - bsm/sarama-cluster#307 - lovoo/goka#211
| func (om *offsetManager) flushToBroker() { | ||
| if !om.conf.Consumer.Offsets.AutoCommit.Enable { |
There was a problem hiding this comment.
Glad this line is fixed, it actually blocks all calls to Commit(), however sometimes we do wish to commit manually.
|
I'm sorry if this PR caused a lot of hazzle. I commit offsets manually by making my own offset manager sending a sarama.OffsetCommitRequest with my own block containing topic, partition and offset. I then send it to the correct broker coordinating that partition. If Sarama now supports manual offset commit I could move to that instead. |
Adding configuration
Consumer.Offsets.Enable(defaulttrue)It blocks the call to
flushToBrokerinOffsetManager