[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-1033416495 Merged to trunk, thanks @RivenSun2 ! -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-1033414741 Yeah I think I agree with you now, the failed cases for 2.12 is `connect.integration` and for 2.13 is `DynamicBrokerReconfiguration` which is known to be a bit flaky. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-1033152747 > @RivenSun2 , I checked the build results after your rebase, there's a test ConsumerBounceTest.testClose that failed twice with NPE. That looks suspicious and didn't fail in trunk branch. Could you please take a look? Thanks. +1. It seems relevant to the recent commit? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-1031034049 @RivenSun2 Thanks for the final commits and thanks to @showuon for making more passes as well. I just did another pass and it looks good to me too! I tried to resolve the conflicts myself and merge but seems it's a bit complicated. @RivenSun2 could you rebase again (sorry!!) and then I will merge. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-1005381862 Retriggering jenkins tests. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-960425136 Restarted jenkins build. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-948851978 Thanks everyone for the valuable discussions, this is a tricky issue indeed and worth some more clarifications, on the status quo. In general there are a lot of issues we've observed with coupling the rebalance protocol with the `poll` calls, and in the long run we are thinking to move the rebalance as well as fetching records completely into the background thread, and hence caller `poll` would only be getting the buffered records --- I will submit a JIRA ticket for this soon --- but as for now, what I'm thinking is a short-term fix, and hence the goal is to fix it with as less scope and complexity as possible. The status quo is that, when we are about to enter the rebalance we try to commit sync with a `rebalance.timeout`. That means: 1) the time blocked on commit maybe LARGER than the timeout passed into `poll`. This is the main issue we'd like to address. 2) if the commit exhausted the timeout but still cannot succeed, we would "give up" with a warning log and still continue the rebalance protocol with the risk that the new host of those partitions may fetch stale committed offsets (or even no offsets at all, if there's no committed offsets before the rebalance happened). This is not ideal, but okay according to the at least once semantics --- note, with exactly once semantics we do not rely on consumer to commit offsets so EOS is not impacted by this behavior. So any solution we may propose, as long as it would not make further regression, would be fine. Now here are my thoughts: * it's risky to just always upper bound the commit timeout with the `poll` parameter value, since in practice many users may call `poll(0)` or with a very small millisecond parameter. This would mean that, we may be much more aggressive on giving up and continue, i.e. making 2) above more exaggerate. Imagine if there's a leader migration happening just at the same time of the rebalance, we would very easily give up the commit even with just one round-trip failure. * it's also risky to always upper bound the commit timeout with the configured rebalance regardless of the `poll` timeout, which we have been discussed many times above and we all know its impact now. So what's left seems to be that, we still make a reasonably "best effort" to commit offsets before continue the rebalance protocol, and that "best effort" should be somewhat irrelevant to the poll timeout, but also not blocking more than the timeout value itself. Hence it occurs to me that, the best effort for now could be that we still do not block for longer than the poll timeout --- i.e. we may even return early --- but we do not give up the commit either, and hence here async commit where could be potentially completed across multiple `poll` calls would be reasonable. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-948136951 @RivenSun2 I talked to @hachikuji offline about the best options to fix it in near term, and we feel the async-commit approach may be more appropriate here. But you'd need to be careful about not just trying once and give up immediately to continue the rebalance. Here's the status quo: * We would try commit upto the configured rebalance.timeout, and if we exhaust that timeout but still cannot succeed (like in this case, keep getting unknown topic partition error), we would just log it and continue the rebalance. * Note that we have a flag `needsJoinPrepare` in AbstractCoordinator which is set before the `onJoinPrepare` call, which means that if the call itself throws out error, upon the next `poll` we would not try to trigger `onJoinPrepare` again. So to make async-commit work, here's a rough sketch of what we'd need to do: * We keep a reference of the last commit response future sent as part of the `onJoinPrepare`. * In `maybeAutoCommitOffsetsSync`, as we would rename it to `maybeAutoCommitOffsetsAsync`, we check if the response future is `null` or not; if it is `null` we just send out the request and get hold on the `future`. And then we call the networkClient.poll once and see if the `future` is completed. If yes and there's no error, we return `true` from `maybeAutoCommitOffsetsAsync` indicating it has suceeded, otherwise we return `false`. * When `maybeAutoCommitOffsetsAsync` returns false, the `onJoinPrepare` would return false immediately as well, and the caller would then reset the `needsJoinPrepare` flag so that next time it would still trigger `onJoinPrepare`. And then return to the `poll` call. By doing that, the `poll` call would not block on commit, but would return immediately after just one trial of the commit request, and the user may potentially call `poll` multiple times in order to complete the commit as part of the `onJoinPrepare` to continue the rebalance, but it would help resolving the longer than `poll` timeout blocking issues. As for the backing off, let's delegate that to KIP-580. 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-948033052 Hi @RivenSun2 just to clarify my original comments here :) * I totally agree with you that we should NOT trigger `maybeAutoCommitOffsetsSync` inside `onJoinPrepare` with `time.timer(rebalanceConfig.rebalanceTimeoutMs` in any of the proposed options we've discussed. On the other hand, we do need to commit sync the offsets before rejoining the group since the consumer may got those partitions reassigned to others in the group, and hence it's important that the offsets are committed already so that others getting the partitions can read the committed offsets and start at the right position. If we allow async commit during rebalance, then we still need to wait for that commit to complete before we can continue to send the join-group request. * What I was wondering is, what timeout value we should use when triggering `maybeAutoCommitOffsetsSync` inside `onJoinPrepare`. The key point here is that, that timeout value should no larger than the `poll()` call's own passed in timeout itself, so it should be in the form of `MIN(poll timer, the configured time)` --- and again, that's why I was thinking that we should keep passing the poll timer down into the callees, since later when we remove the deprecated call we would not need to pass in the boolean flag anyways. The `configured time` itself, could be either 1) we use a separate config as you suggested, or 2) we piggy-back on the existing configs still, e.g. on `DEFAULT_API_TIMEOUT_MS_CONFIG` (since semantically that defines how long a blocking call, like commitSync should be take at max). The semantics is basically that, the poll call should definitely return within the specified timeout parameter value, no matter if it is blocked on committing offsets, or other things li ke fetching. * Now for `commitSync` called by users, my rationale is the same as `poll`: say, if the user calls `consumer.commitSync` or `consumer.poll` with a very large timer anyways, then it is the intentional behavior that we should try to succeed that request for at least that amount of time. Here I think the reason with short backoff time it would mean a lot of requests sent to the brokers during that period of time hence wasted broker's CPUs, is actually due to the lack of https://issues.apache.org/jira/browse/KAFKA-9800. But even with that, I'd consider that users still have a lot of different ways to bombard a brokers other than `commitSync` with manually specified non-existent topic partitions, or with `allConsumed` partitions that would not be refreshed yet. So personally, I'd suggest we make the following changes as a fix: 1) pass in the timer to `onJoinPrepare`, and inside it, trigger `maybeAutoCommitOffsetsSync` with the time as `MIN(passed-in poll timer, configured timeout)`. 2) complete https://issues.apache.org/jira/browse/KAFKA-9800 (KIP-580) to reduce the bombarding effect of retriable requests. In the future as we have KRaft for metadata propagation, we can then reason about whether `UnknownTopicOrPartition` error is certain, or not, and then treat it as fatal rather than retriable. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-941700809 -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[GitHub] [kafka] guozhangwang commented on pull request #11340: KAFKA-13310 : KafkaConsumer cannot jump out of the poll method, and the…
guozhangwang commented on pull request #11340: URL: https://github.com/apache/kafka/pull/11340#issuecomment-941700809 BTW I have not thoroughly gone through the added testing, I also agree with @showuon 's comments about the test coverage -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
