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 2020/06/27 00:40:26 UTC
[GitHub] [kafka] guozhangwang opened a new pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
guozhangwang opened a new pull request #8934:
URL: https://github.com/apache/kafka/pull/8934
The intention of using poll(0) is to not block on rebalance but still return some data; however, if there's no fetchable partitions then there's no point of polling with 0ms. What's worse, with poll(0) we may fall into a busy loop since we may advance the system test with too small pace.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#issuecomment-650464132
@hachikuji @ijuma
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#discussion_r450504028
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##########
@@ -416,7 +428,13 @@ boolean joinGroupIfNeeded(final Timer timer) {
}
final RequestFuture<ByteBuffer> future = initiateJoinGroup();
- client.poll(future, timer);
+
+ // if the flat is not set to true; we only try once and not block on the join result
Review comment:
flag?
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1258,11 +1261,19 @@ public void assign(Collection<TopicPartition> partitions) {
}
}
+ private boolean coordinatorNeededForAssignment() {
Review comment:
nit: seems unused?
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1218,17 +1218,20 @@ public void assign(Collection<TopicPartition> partitions) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- // poll for new data until the timeout expires
do {
client.maybeTriggerWakeup();
if (includeMetadataInTimeout) {
- // try to update assignment metadata BUT do not need to block on the timer,
- // since even if we are 1) in the middle of a rebalance or 2) have partitions
- // with unknown starting positions we may still want to return some data
- // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms
- // to never block on completing the rebalance procedure if there's any
- updateAssignmentMetadataIfNeeded(time.timer(0L));
+ // try to update assignment metadata BUT do not need to block on the timer if we still have
+ // some assigned partitions, since even if we are 1) in the middle of a rebalance
+ // or 2) have partitions with unknown starting positions we may still want to return some data
+ // as long as there are some partitions fetchable; NOTE we do not block on rebalancing to complete
+ // if there's one pending if we still have some fetchable partitions
+ if (subscriptions.fetchablePartitions(tp -> true).isEmpty()) {
Review comment:
This call to `fetchablePartitions` is unfortunate. It is an entire pass over all the assigned partitions on every poll. Is there any way it can be avoided?
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##########
@@ -416,7 +428,13 @@ boolean joinGroupIfNeeded(final Timer timer) {
}
final RequestFuture<ByteBuffer> future = initiateJoinGroup();
- client.poll(future, timer);
+
+ // if the flat is not set to true; we only try once and not block on the join result
+ if (waitUntilComplete)
+ client.poll(future, timer);
+ else
+ client.poll(timer);
Review comment:
Do we want this to be `poll(0)`? Otherwise we're still blocking here.
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1218,17 +1218,20 @@ public void assign(Collection<TopicPartition> partitions) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- // poll for new data until the timeout expires
do {
client.maybeTriggerWakeup();
if (includeMetadataInTimeout) {
- // try to update assignment metadata BUT do not need to block on the timer,
- // since even if we are 1) in the middle of a rebalance or 2) have partitions
- // with unknown starting positions we may still want to return some data
- // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms
- // to never block on completing the rebalance procedure if there's any
- updateAssignmentMetadataIfNeeded(time.timer(0L));
+ // try to update assignment metadata BUT do not need to block on the timer if we still have
Review comment:
Just making sure I understand the problem. In the old logic, `updateAssignmentMetadataIfNeeded` never blocks. Even if we have no assignment and do not know the coordinator, we won't block here. That means we should fall through to `pollForFetches`. The poll timeout is set by the following logic:
```java
long pollTimeout = coordinator == null ? timer.remainingMs() :
Math.min(coordinator.timeToNextPoll(timer.currentTimeMs()), timer.remainingMs());
```
I think the only way I can see this logic getting into trouble is if `Heartbeat.timeToNextHeartbeat` returns a small value. So if the coordinator remains unknown for a little while and we need to heartbeat, we might get into a busy loop. Is that about right or are there other cases?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#discussion_r450594795
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1218,17 +1218,20 @@ public void assign(Collection<TopicPartition> partitions) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- // poll for new data until the timeout expires
do {
client.maybeTriggerWakeup();
if (includeMetadataInTimeout) {
- // try to update assignment metadata BUT do not need to block on the timer,
- // since even if we are 1) in the middle of a rebalance or 2) have partitions
- // with unknown starting positions we may still want to return some data
- // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms
- // to never block on completing the rebalance procedure if there's any
- updateAssignmentMetadataIfNeeded(time.timer(0L));
+ // try to update assignment metadata BUT do not need to block on the timer if we still have
+ // some assigned partitions, since even if we are 1) in the middle of a rebalance
+ // or 2) have partitions with unknown starting positions we may still want to return some data
+ // as long as there are some partitions fetchable; NOTE we do not block on rebalancing to complete
+ // if there's one pending if we still have some fetchable partitions
+ if (subscriptions.fetchablePartitions(tp -> true).isEmpty()) {
Review comment:
Agree, I can think of caching a boolean to avoid it.
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1218,17 +1218,20 @@ public void assign(Collection<TopicPartition> partitions) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- // poll for new data until the timeout expires
do {
client.maybeTriggerWakeup();
if (includeMetadataInTimeout) {
- // try to update assignment metadata BUT do not need to block on the timer,
- // since even if we are 1) in the middle of a rebalance or 2) have partitions
- // with unknown starting positions we may still want to return some data
- // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms
- // to never block on completing the rebalance procedure if there's any
- updateAssignmentMetadataIfNeeded(time.timer(0L));
+ // try to update assignment metadata BUT do not need to block on the timer if we still have
Review comment:
My guess is that the `pollDelayMs` from `trySend(timer.currentTimeMs())` is zero and hence
```
long pollTimeout = Math.min(timer.remainingMs(), pollDelayMs);
```
would become zero.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#issuecomment-652573659
> @guozhangwang Looks like the code is not compiling. I think you grabbed `binaryExponentialElectionBackoffMs` from the prototype Raft implementation, but we need to initialize the max value and change names.
Sorry, my bad, I copied the func over just to get the core logic but I forgot to remove the func after I've inlined it inside poll.
My request is to check whether doing this exponential backoff makes sense to you.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji edited a comment on pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
hachikuji edited a comment on pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#issuecomment-652518999
@guozhangwang Looks like the code is not compiling. I think you grabbed `binaryExponentialElectionBackoffMs` from the prototype Raft implementation, but we need to initialize the max value and change names.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#discussion_r450960971
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1218,17 +1218,20 @@ public void assign(Collection<TopicPartition> partitions) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- // poll for new data until the timeout expires
do {
client.maybeTriggerWakeup();
if (includeMetadataInTimeout) {
- // try to update assignment metadata BUT do not need to block on the timer,
- // since even if we are 1) in the middle of a rebalance or 2) have partitions
- // with unknown starting positions we may still want to return some data
- // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms
- // to never block on completing the rebalance procedure if there's any
- updateAssignmentMetadataIfNeeded(time.timer(0L));
+ // try to update assignment metadata BUT do not need to block on the timer if we still have
Review comment:
Hmm, not sure if I buy that. `trySend` would only return a value less than 5s if there was a pending request which needed to be send. However, as soon as it got sent, it would go back to 5s. So I'm not sure if it alone can account for a tight poll loop.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#issuecomment-651409789
> Reading from the JIRA log, this seems not to be the full solution to the reported bug?
Here's my reasoning: if there are still some owned partitions that are fetchable even during a rebalance, then within the while-loop even if we exit the `updateAssignmentMetadataIfNeeded` we can still try to fetch from then and then break out of the while loop. However, if we do not have fetchable partitions then we may be blocked in the while-loop for a long time until the poll timeout expires. Normally that should still be able to exit the busy while-loop eventually since the `timer.update` should still be called in `pollForFetches` within that while loop and hence advance the timer.
It's indeed puzzling to me that why with cooperative rebalance, then we would block inside the while loop forever. I will post this to the JIRA and ask other folks.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang merged pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang merged pull request #8934:
URL: https://github.com/apache/kafka/pull/8934
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#discussion_r450978217
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1218,17 +1218,20 @@ public void assign(Collection<TopicPartition> partitions) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- // poll for new data until the timeout expires
do {
client.maybeTriggerWakeup();
if (includeMetadataInTimeout) {
- // try to update assignment metadata BUT do not need to block on the timer,
- // since even if we are 1) in the middle of a rebalance or 2) have partitions
- // with unknown starting positions we may still want to return some data
- // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms
- // to never block on completing the rebalance procedure if there's any
- updateAssignmentMetadataIfNeeded(time.timer(0L));
+ // try to update assignment metadata BUT do not need to block on the timer if we still have
Review comment:
I did a simple experiment with one broker and one consumer. After joining the group, I did a kill -9 to stop the broker. I had the consumer log the poll timeout that was used in `pollForFetches`. What I saw is this:
```
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#issuecomment-652518999
@guozhangwang Looks like the code is not compiling. I think you grabbed `binaryExponentialElectionBackoffMs` from the Raft implementation, but we need to initialize the max value and change names.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#issuecomment-652202875
@hachikuji Could you take a look at this one since it is a 2.6 blocker?
Another alternative if we are less comfortable with the current exponential timeout fix-forward, is to just revert the behavior and try to complete the metadata update (including join-group) up to the poll timeout.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#discussion_r450978217
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
##########
@@ -1218,17 +1218,20 @@ public void assign(Collection<TopicPartition> partitions) {
throw new IllegalStateException("Consumer is not subscribed to any topics or assigned any partitions");
}
- // poll for new data until the timeout expires
do {
client.maybeTriggerWakeup();
if (includeMetadataInTimeout) {
- // try to update assignment metadata BUT do not need to block on the timer,
- // since even if we are 1) in the middle of a rebalance or 2) have partitions
- // with unknown starting positions we may still want to return some data
- // as long as there are some partitions fetchable; NOTE we always use a timer with 0ms
- // to never block on completing the rebalance procedure if there's any
- updateAssignmentMetadataIfNeeded(time.timer(0L));
+ // try to update assignment metadata BUT do not need to block on the timer if we still have
Review comment:
I did a simple experiment with one broker and one consumer. After the consumer joined the group, I did a kill -9 to stop the broker. I had the consumer log the poll timeout that was used in `pollForFetches`. What I saw is this:
```
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,825] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
[2020-07-07 08:53:10,839] INFO [Consumer clientId=consumer-foo-1, groupId=foo] Polling for fetches with timeout 0 (org.apache.kafka.clients.consumer.KafkaConsumer)
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#issuecomment-655636676
Cherry-picked to 2.6.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #8934: KAFKA-10134: Use long poll if we do not have fetchable partitions
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #8934:
URL: https://github.com/apache/kafka/pull/8934#discussion_r450597622
##########
File path: clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java
##########
@@ -416,7 +428,13 @@ boolean joinGroupIfNeeded(final Timer timer) {
}
final RequestFuture<ByteBuffer> future = initiateJoinGroup();
- client.poll(future, timer);
+
+ // if the flat is not set to true; we only try once and not block on the join result
+ if (waitUntilComplete)
+ client.poll(future, timer);
+ else
+ client.poll(timer);
Review comment:
My rationale after thinking about this, is that it may be better to wait on at least some IOs to be done, especially if there are on-going fetch requests as well: we may complete writing a fetch request, or join / sync group request, or reading a response before returning.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org