Manual commit multiple partitions consumer #137
Conversation
|
@twmb I would still like to ask you about this one even though @JacobSMoller closed this PR. From the logs I see Essentially we are using a goroutine per partition and we manually commit offsets. When a rebalance occurs and a worker is calling onrevoke for some partitions, the remaining goroutine consumers most often throws It seems to me like franz-go is trying to commit offsets, while the group is not synced. Is this expected? I have no idea what the expected behavior should be here. The problem seems to increase with when latency to the kafka brokers are high. On low latency clusters the problem disappears. What are your two cents on this issue, is this expected or should I look further into this trying to provide a better example reproducing the error? |
|
@JacobSMoller was this accidentally closed? I'm always open to new examples. @brunsgaard how often are you running into this error? If it's what I think it is, I expect this to be rare. If it isn't rare, then I need to look into this more. Currently, committing offsets does not block rebalancing, so what can happen is that you issue a commit immediately before a rebalance which grabs the group's current generation, and then a rebalance starts and finishes before the commit is issued, and then the commit when finally issued uses an old generation. A way to fix this internally would be to block rebalances during any commit. If this is what you're running into, I'd favor blocking rebalances during commit, because a few other people have ran into this as well. edit: similar thing to the first error, REBALANCE_IN_PROGRESS, in this case the commit happened during the rebalance rather than after. |
|
Hi wasn't accidental, just realized i didn't quite have the time to clean it up properly and put in the description @brunsgaard has now added, at the time i opened this :). We are seeing the ILLEGAL_GENERATION error quite consistently, when running our code locally from Europe towards a kafka cluster in the US, which has quite high latency. If we run the code from a k8s cluster in a US region we don't see it. |
JacobSMoller
changed the title
|
Our workloads essentially reads in 10000 elements per partition and puts them in a batch and forwards them to a grpc endpoint before committing. In out test cases we are reading from the start of the topic, so there is a lot of data and thus we commit very frequently, which would not be the case in production. (but it also happens with @JacobSMoller's example code that does not commit that often, hmm) I dont understand the generel impact and consequences of blocking rebalances during commit, but It sounds like a good idea. What are next steps @twmb? @JacobSMoller I suggest adding a done channel to the pconsumer so we can make sure current work is finished and committed before onrewokes/lost returns. |
|
Did a bit more testing to reproduce the ILLEGAL_GENERATION, using the example code from this PR. With minor changes in adding a comment above <-pc.done on line 84, enabling SASL to connect to our cluster and added the kgo.WithLogger option to get you some logs. Full DEBUG log here: https://gist.github.com/JacobSMoller/d3a2688ff3383a44f360e19066c97266 And INFO log version of the same run below Hope it helps understanding the behaviour we are talking mentioning with the ILLEGAL_GENERATION error. |
|
So from those logs, what I can see:
Committing should probably block joining a group, or wait until after a group rejoin has finished. If committing is blocked on a rebalance, that's probably fine because onRevoked can be used to fence out committing for records that the client no longer owns. Also I wonder if the per-partition consuming pattern can be made easier. |
@twmb is this something you will take on?
It could make sense, but I also like that franz-go is a bit more transparent and less magic compared to e.g. Sarama. That said I think it makes sense to "package" a default |
|
After looking at it, I think it would be easier to have the client reissue the commit if it errors with Perhaps, if the commit fails with |
See #137. Cooperative consumers can consume during rebalancing. If they commit at the start of a rebalance that ends after, then the commit will fail with ILLEGAL_GENERATION (or sometimes, REBALANCE_IN_PROGRESS). For cooperative specifically, if the commit fails but the consumer still owns all partitions being committed, we now retry the commit once. This should help alleviate commit errors that well written consumers are currently running into. We retry up to twice because the rebalancing when cooperative results in two rebalances. A third failure is more unexpected.
|
If possible, can you try 9b8da34? I plan to tag this in v1.4 this next week. |
|
@twmb I can still reproduce after doing a Debug logs for a run using the new commit. https://gist.github.com/JacobSMoller/44cf79f61cbf305ddd7c0f0210319ba4 INFO logs |
|
@twmb I think it would be more correct to block joining a group (even it is more complicated implementation wise) for three reasons With the re-issue commit after join approach:
Any feedback on these concerns? |
|
Those logs have shown I have one bit of logic wrong and I'd like to fix it. Every commit was happening 3x -- I stayed inside a conditional which reset the bool being used to track whether a recommit was necessary.
This shouldn't be the case, unless you committed to a partition that was lost.
Yes, true. The fundamental problem is that clients have to make a choice during implementation: should the client wait for the user to process records and commit before rebalancing, or should the client rebalance and punt all responsibilities for when to commit to the user? The first choice poses the risk for the member to be booted from the group if processing records is so long that a rebalance causes the member to be kicked from the group (this is the sarama choice), the second choice makes it a lot harder for end users to guard against rebalances under the hood but ensures rebalancing will never be blocked on the application (this is the franz-go choice). The no-wait-for-user also makes duplicates more likely around rebalances: you may have polled records, and then a rebalance happened, and now something else will consume those same records. If the rebalance waited for an end user, then you could In the non-cooperative world, a rebalance that waits for applications is also problematic because rebalances are slow and there would be a large stop-the-world effect as all members of the group waits for the slowest to finish what it's doing and rejoin. In a cooperative world, waiting for an end user is of less effect because consuming continues during the rebalance. In thinking about it more though, if a user can always guarantee that processing records is quick (and the user usually can), franz-go may be taking an overly paranoid approach by rebalancing under the hood transparently. I'm not sure what it would take to add an option, Another thought is that with concurrent rebalancing, perhaps I should embed the group generation as a hidden field in a record and use that as the generation when committing, rather than using the group's current generation. This would guard against accidental rewinds when an application loses a partition across a rebalance. |
See #137. Cooperative consumers can consume during rebalancing. If they commit at the start of a rebalance that ends after, then the commit will fail with ILLEGAL_GENERATION (or sometimes, REBALANCE_IN_PROGRESS). For cooperative specifically, if the commit fails but the consumer still owns all partitions being committed, we now retry the commit once. This should help alleviate commit errors that well written consumers are currently running into. We retry up to twice because the rebalancing when cooperative results in two rebalances. A third failure is more unexpected.
|
If you could try bf0e5d7, that should have been the more proper fix (I think). |
See #137. Cooperative consumers can consume during rebalancing. If they commit at the start of a rebalance that ends after, then the commit will fail with ILLEGAL_GENERATION (or sometimes, REBALANCE_IN_PROGRESS). For cooperative specifically, if the commit fails but the consumer still owns all partitions being committed, we now retry the commit once. This should help alleviate commit errors that well written consumers are currently running into. We retry up to twice because the rebalancing when cooperative results in two rebalances. A third failure is more unexpected.
One strength of Sarama's consumer group is that it is easier to reason about where record processing logic falls within the group management lifecycle: if you are processing records, you can be sure that the group will not rebalance underneath you. This is also a risk, though, if your processing logic is so long that your group member is booted from the group. This client took the opposite approach of separating the group management logic from the consuming logic. This essentially eliminated the risk of long processing booting the member, but made it much more difficult to reason about when to commit. There are plenty of disclaimers about potential duplicates, and any non-transactional consumer should expect duplicates at some point, but if a user wants to opt for the simpler approach of consuming & processing within one group generation, we should support that. In fact, if a user's processing loop is very fast, we really should encourage it. Helps #137.
|
Besides the prior commit, what do you think of this? 0c8c1a6 I'm not 100% on the naming yet but I'm going to sleep on it, I haven't thought of better API names all day, so this may be the final API. Also, I evaluated having that
So, I'll probably add another example once all this is over (or perhaps we can redux this PR?) and call it good. |
One strength of Sarama's consumer group is that it is easier to reason about where record processing logic falls within the group management lifecycle: if you are processing records, you can be sure that the group will not rebalance underneath you. This is also a risk, though, if your processing logic is so long that your group member is booted from the group. This client took the opposite approach of separating the group management logic from the consuming logic. This essentially eliminated the risk of long processing booting the member, but made it much more difficult to reason about when to commit. There are plenty of disclaimers about potential duplicates, and any non-transactional consumer should expect duplicates at some point, but if a user wants to opt for the simpler approach of consuming & processing within one group generation, we should support that. In fact, if a user's processing loop is very fast, we really should encourage it. Helps #137.
|
@JacobSMoller bf0e5d7 maps to 748509a |
|
@twmb using bf0e5d7 and 0c8c1a6 we can no longer reproduce the error. What do you think about the example code using the lost/assigned lock around Thanks a lot for your quick work on this. Let me know if you have some changes to the example PS. Do you have an ETA on v1.4 ? |
| } | ||
| }) | ||
| }) | ||
| s.mu.Lock() |
There was a problem hiding this comment.
If I'm reading this correctly, this lock is to ensure onLost is not running?
There was a problem hiding this comment.
From the main comments, yes. Perhaps a comment here on that would be good.
There was a problem hiding this comment.
onLost, is having <-pc.done that ensures the records are committed before the goroutine returns, thus before onLost returns. Thus committing work in progress before rebalance is allowed. Does it makes sense?
There was a problem hiding this comment.
I'm planning to merge this then add followup commit that adds a few more comments and potentially avoids EachTopic (which internally allocates a map). I'm pretty sure on the new API naming.
There was a problem hiding this comment.
We are unsure if that makes sense our self. Please advice :)
There was a problem hiding this comment.
I'll merge this now and touch up a few things (file reorganization now that there are two similar examples, potentially avoid FetchTopics, add an example using AutoCommitMarks). Will ping here once done.
There was a problem hiding this comment.
I see why this lock is necessary, and I think I may switch BlockRebalanceOnPoll to also block polling while user provided OnAssiged / OnRevoked / OnLost are called.
There was a problem hiding this comment.
Please check https:/twmb/franz-go/blob/v1.4/examples/goroutine_per_partition_consuming/manual_commit/main.go to see what this example has become :)
I think it's a good bit simpler now to avoid locking.
| // Mimick work to happen before committing records | ||
| time.Sleep(time.Duration(rand.Intn(150)+100) * time.Millisecond) | ||
| fmt.Printf("Some sort of work done, about to commit t %s p %d\n", topic, partition) | ||
| err := cl.CommitRecords(context.Background(), recs...) |
There was a problem hiding this comment.
I wonder if you would prefer MarkCommitRecords here, and AutoCommitMarks? Perhaps I need another, CommitMarksCommitUncommittedOffsets does exactly this.
This would allow more batching for your commits, vs. each commit here being blocking.
Not sure, what do you think? This is fine too.
There was a problem hiding this comment.
In our "real" usecase we dont commit that often, we rollup 10000 messages over a period of minutes so the commit per partition is called infrequently. For this example i like we stick to the commit per []*kgo.records if your prefer to use the markcommit strategy for this example we will do that. We are aware committing too often is a undesirable pattern.
|
@twmb I would also like to call out "like jacob" the way you have worked with us on this issue/PR, It over and beyond thank you. |
|
Thanks! As mentioned, I'll merge this now and touch up a few things (file reorganization now that there are two similar examples, potentially avoid FetchTopics, add an example using AutoCommitMarks). Will ping here once done. |
See #137. Cooperative consumers can consume during rebalancing. If they commit at the start of a rebalance that ends after, then the commit will fail with ILLEGAL_GENERATION (or sometimes, REBALANCE_IN_PROGRESS). For cooperative specifically, if the commit fails but the consumer still owns all partitions being committed, we now retry the commit once. This should help alleviate commit errors that well written consumers are currently running into. We retry up to twice because the rebalancing when cooperative results in two rebalances. A third failure is more unexpected.
One strength of Sarama's consumer group is that it is easier to reason about where record processing logic falls within the group management lifecycle: if you are processing records, you can be sure that the group will not rebalance underneath you. This is also a risk, though, if your processing logic is so long that your group member is booted from the group. This client took the opposite approach of separating the group management logic from the consuming logic. This essentially eliminated the risk of long processing booting the member, but made it much more difficult to reason about when to commit. There are plenty of disclaimers about potential duplicates, and any non-transactional consumer should expect duplicates at some point, but if a user wants to opt for the simpler approach of consuming & processing within one group generation, we should support that. In fact, if a user's processing loop is very fast, we really should encourage it. Helps #137.
|
I'm changing the logic that I introduced for this patch, #410 is likely better. |
|
Thanks for the heads up on the change. |
Slightly modified version of https:/twmb/franz-go/tree/master/examples/goroutine_per_partition_consuming
Using manual commits.
During testing of this i noticed that when adding new consumers I see errors from
err := cl.CommitRecords(context.Background(), recs...)The errors where mainly
But also included
There did not seem to be a clear pattern in the errors