RivenSun2 edited a comment on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-948271872
Hi @guozhangwang Thank you for your reply，I agree with most of your points, `KIP-580` is also a great optimization, I hope it can be completed as soon as possible. But I still suggest that the `commitOffsets` in `onJoinPrepare` only need to be done once in one `consumer.poll`, and it should not be necessary to call the poll method multiple times to maintain `async commitOffsets` until it commits successfully or times out. That would be more complicated. And if the asynchronous submission fails all the time, we need to consider how to control a timeout period during multiple calls to `consumer.poll`. The `rebalance.timeout` you mentioned, I understand it should be applied here, to set a timeout for `commitOffset` during multiple polls. I think it’s a weird practice to maintain a timeout for asynchronous operations. At the same time, its effect is actually equivalent to only do `commitOffset` once, setting a timeout `MIN (poll timer, the configured timer)` for this commit, and letting customers to consider how to better adjust these parameters `poll timer`, `the configured timer`. If we only did it once, what would be the worst result. Partition re-allocation results in repeated consumption of some messages. If we inform customers that this may happen, and provide parameters for customers to adjust by themselves, I think this is acceptable. my suggestion is: 1. `OnJoinPrepare` provides parameters for customers to choose whether to submit asynchronously or synchronously. `auto.commit.way.in.rebalance` Type: string Default: sync Valid Values: [async, sync] If it is submitted asynchronously, no other parameters need to be adjusted 2. If `auto.commit.way.in.rebalance`=sync, Provide `max.offset.auto.commit.in.rebalance.ms` parameter for customers to adjust, `max.offset.auto.commit.in.rebalance.ms` Type: long Default: 5000 Valid Values: [0,max.poll.interval.ms) Use `MIN(poll timer, this configured timer)` to replace `time.timer(rebalanceConfig.rebalanceTimeoutMs)` in `onJoinPrepare` WDYT? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
