You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/07/22 21:59:04 UTC
[GitHub] [kafka] ableegoldman opened a new pull request #11111: HOTFIX: guard against overflow when computing `joinGroupTimeoutMs`
ableegoldman opened a new pull request #11111:
URL: https://github.com/apache/kafka/pull/11111
In older versions of Kafka Streams, the `max.poll.interval.ms` config was overridden by default to `Integer.MAX_VALUE`. Even after we removed this override, users of both the plain consumer client and kafka streams still set the poll interval to MAX_VALUE somewhat often. Unfortunately, this causes an overflow when computing the `joinGroupTimeoutMs` and results in it being set to the `request.timeout.ms` instead, which is much lower.
This can easily make consumers drop out of the group, since they must rejoin now within 30s (by default) yet have no obligation to almost ever call poll() given the high `max.poll.interval.ms`. We just need to check for overflow and fix it to `Integer.MAX_VALUE` when it occurs.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #11111: KAFKA-13126: guard against overflow when computing `joinGroupTimeoutMs`
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11111:
URL: https://github.com/apache/kafka/pull/11111#discussion_r675266851
##########
File path: clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java
##########
@@ -60,17 +60,16 @@ public static boolean awaitReady(KafkaClient client, Node node, Time time, long
throw new IllegalArgumentException("Timeout needs to be greater than 0");
}
long startTime = time.milliseconds();
- long expiryTime = startTime + timeoutMs;
if (isReady(client, node, startTime) || client.ready(node, startTime))
return true;
long attemptStartTime = time.milliseconds();
- while (!client.isReady(node, attemptStartTime) && attemptStartTime < expiryTime) {
+ while (!client.isReady(node, attemptStartTime) && attemptStartTime - startTime < timeoutMs) {
if (client.connectionFailed(node)) {
throw new IOException("Connection to " + node + " failed.");
}
- long pollTimeout = expiryTime - attemptStartTime;
+ long pollTimeout = (startTime - attemptStartTime) + timeoutMs;
Review comment:
Ah, thanks!
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #11111: KAFKA-13126: guard against overflow when computing `joinGroupTimeoutMs`
Posted by GitBox <gi...@apache.org>.
vvcephei commented on a change in pull request #11111:
URL: https://github.com/apache/kafka/pull/11111#discussion_r675220503
##########
File path: clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java
##########
@@ -60,17 +60,16 @@ public static boolean awaitReady(KafkaClient client, Node node, Time time, long
throw new IllegalArgumentException("Timeout needs to be greater than 0");
}
long startTime = time.milliseconds();
- long expiryTime = startTime + timeoutMs;
if (isReady(client, node, startTime) || client.ready(node, startTime))
return true;
long attemptStartTime = time.milliseconds();
- while (!client.isReady(node, attemptStartTime) && attemptStartTime < expiryTime) {
+ while (!client.isReady(node, attemptStartTime) && attemptStartTime - startTime < timeoutMs) {
if (client.connectionFailed(node)) {
throw new IOException("Connection to " + node + " failed.");
}
- long pollTimeout = expiryTime - attemptStartTime;
+ long pollTimeout = (startTime - attemptStartTime) + timeoutMs;
Review comment:
This would still overflow if `timeoutMs` is `MAX_VALUE`, right?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on pull request #11111: KAFKA-13126: guard against overflow when computing `joinGroupTimeoutMs`
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on pull request #11111:
URL: https://github.com/apache/kafka/pull/11111#issuecomment-885959141
Merged to trunk
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman merged pull request #11111: KAFKA-13126: guard against overflow when computing `joinGroupTimeoutMs`
Posted by GitBox <gi...@apache.org>.
ableegoldman merged pull request #11111:
URL: https://github.com/apache/kafka/pull/11111
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] showuon commented on pull request #11111: KAFKA-13126: guard against overflow when computing `joinGroupTimeoutMs`
Posted by GitBox <gi...@apache.org>.
showuon commented on pull request #11111:
URL: https://github.com/apache/kafka/pull/11111#issuecomment-885434828
Failed tests are unrelated.
```
Build / JDK 11 and Scala 2.13 / kafka.api.TransactionsExpirationTest.testBumpTransactionalEpochAfterInvalidProducerIdMapping()
Build / JDK 8 and Scala 2.12 / kafka.api.ConsumerBounceTest.testCloseDuringRebalance()
Build / JDK 8 and Scala 2.12 / kafka.api.ConsumerBounceTest.testCloseDuringRebalance()
Build / JDK 16 and Scala 2.13 / kafka.api.ConsumerBounceTest.testCloseDuringRebalance()
Build / JDK 16 and Scala 2.13 / kafka.api.ConsumerBounceTest.testCloseDuringRebalance()
```
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] ableegoldman commented on a change in pull request #11111: KAFKA-13126: guard against overflow when computing `joinGroupTimeoutMs`
Posted by GitBox <gi...@apache.org>.
ableegoldman commented on a change in pull request #11111:
URL: https://github.com/apache/kafka/pull/11111#discussion_r675241237
##########
File path: clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java
##########
@@ -60,17 +60,16 @@ public static boolean awaitReady(KafkaClient client, Node node, Time time, long
throw new IllegalArgumentException("Timeout needs to be greater than 0");
}
long startTime = time.milliseconds();
- long expiryTime = startTime + timeoutMs;
if (isReady(client, node, startTime) || client.ready(node, startTime))
return true;
long attemptStartTime = time.milliseconds();
- while (!client.isReady(node, attemptStartTime) && attemptStartTime < expiryTime) {
+ while (!client.isReady(node, attemptStartTime) && attemptStartTime - startTime < timeoutMs) {
if (client.connectionFailed(node)) {
throw new IOException("Connection to " + node + " failed.");
}
- long pollTimeout = expiryTime - attemptStartTime;
+ long pollTimeout = (startTime - attemptStartTime) + timeoutMs;
Review comment:
The `startTime` is set once at the beginning of the method while the `attemptStartTime` is initialized just before the first attempt and then updated again after every iteration. So the `attemptStartTime` is always greater than the `startTime` and therefore the quantity being added to the `timeoutMs` here is actually negative.
But I see how that's confusing, I'll refactor the expression to make this more clear
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org