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 2022/11/23 02:13:05 UTC
[GitHub] [kafka] jeffkbkim opened a new pull request, #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata [WIP]
jeffkbkim opened a new pull request, #12897:
URL: https://github.com/apache/kafka/pull/12897
The consumer (fetcher) refreshes the preferred read replica only on three conditions:
1. the consumer receives an OFFSET_OUT_OF_RANGE error
2. the follower does not exist in the client's metadata (i.e., offline)
3. after metadata.max.age.ms (5 min default)
For other errors, it will continue to reach to the possibly unavailable follower and only after 5 minutes will it refresh the preferred read replica and go back to the leader.
A specific example is when a partition is reassigned. the consumer will get NOT_LEADER_OR_FOLLOWER which triggers a metadata update but the preferred read replica will not be refreshed as the follower is still online. it will continue to reach out to the old follower until the preferred read replica expires.
the consumer can instead refresh its preferred read replica whenever it makes a metadata update request. so when the consumer receives i.e. NOT_LEADER_OR_FOLLOWER it can find the new preferred read replica without waiting for the expiration.
### 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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac closed pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
dajac closed pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
URL: https://github.com/apache/kafka/pull/12897
--
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] jeffkbkim commented on a diff in pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata [WIP]
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on code in PR #12897:
URL: https://github.com/apache/kafka/pull/12897#discussion_r1033847041
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java:
##########
@@ -5157,11 +5225,6 @@ private <K, V> void buildFetcher(MetricConfig metricConfig,
subscriptionState, logContext);
}
- private void buildFetcher(SubscriptionState subscriptionState, LogContext logContext) {
Review Comment:
this is unused
--
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] dajac commented on a diff in pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
dajac commented on code in PR #12897:
URL: https://github.com/apache/kafka/pull/12897#discussion_r1035961196
##########
clients/src/main/java/org/apache/kafka/common/Cluster.java:
##########
@@ -253,7 +253,11 @@ public Node nodeById(int id) {
public Optional<Node> nodeIfOnline(TopicPartition partition, int id) {
Node node = nodeById(id);
PartitionInfo partitionInfo = partition(partition);
- if (node != null && partitionInfo != null && !Arrays.asList(partitionInfo.offlineReplicas()).contains(node)) {
+
+ if (node != null && partitionInfo != null &&
+ !Arrays.asList(partitionInfo.offlineReplicas()).contains(node) &&
+ Arrays.asList(partitionInfo.replicas()).contains(node)) {
Review Comment:
Could we add or update a test case for this as well?
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestTestUtils.java:
##########
@@ -208,10 +209,10 @@ public static MetadataResponse metadataUpdateWith(final String clusterId,
for (int i = 0; i < numPartitions; i++) {
TopicPartition tp = new TopicPartition(topic, i);
Node leader = nodes.get(i % nodes.size());
- List<Integer> replicaIds = Collections.singletonList(leader.id());
+ List<Integer> replicaIds = nodes.stream().map(Node::id).collect(Collectors.toList());
Review Comment:
Why do we need to change this? Is it OK to change for all existing usages?
--
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] jeffkbkim commented on a diff in pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on code in PR #12897:
URL: https://github.com/apache/kafka/pull/12897#discussion_r1035045272
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java:
##########
@@ -346,8 +346,10 @@ public void onFailure(RuntimeException e) {
FetchSessionHandler handler = sessionHandler(fetchTarget.id());
if (handler != null) {
handler.handleError(e);
+ handler.sessionTopicPartitions().forEach(subscriptions::clearPreferredReadReplica);
}
} finally {
+ metadata.requestUpdate();
Review Comment:
i did look at the code path but did not validate via a network client, thanks for the find!
--
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] jeffkbkim commented on a diff in pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on code in PR #12897:
URL: https://github.com/apache/kafka/pull/12897#discussion_r1036383573
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestTestUtils.java:
##########
@@ -208,10 +209,10 @@ public static MetadataResponse metadataUpdateWith(final String clusterId,
for (int i = 0; i < numPartitions; i++) {
TopicPartition tp = new TopicPartition(topic, i);
Node leader = nodes.get(i % nodes.size());
- List<Integer> replicaIds = Collections.singletonList(leader.id());
+ List<Integer> replicaIds = nodes.stream().map(Node::id).collect(Collectors.toList());
Review Comment:
tests were failing, specifically the ones that were testing preferred read replica since it was using 2 nodes instead of 1. replica=[0,1] is expected with the new nodeIfOnline changes.
this provides the expected behavior: all nodes that are passed in are part of the replica and isr set.
--
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] jeffkbkim commented on pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
jeffkbkim commented on PR #12897:
URL: https://github.com/apache/kafka/pull/12897#issuecomment-1331446867
thanks @dajac for the review. as discussed offline:
`nodeIfOnline` will be updated to also check whether the node is within the given partition info replica set. this will help clear the preferred read replica in cases where a partition is reassigned. this will resolve the given example (topic reassignment) as well as cases where the follower falls out of ISR due to a network problem.
however, we will still keep the changes to clear the preferred read replica on errors so that the consumer relies on the leader for up-to-date preferred read replica.
--
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] hachikuji commented on a diff in pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12897:
URL: https://github.com/apache/kafka/pull/12897#discussion_r1038530374
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestTestUtils.java:
##########
@@ -208,10 +209,10 @@ public static MetadataResponse metadataUpdateWith(final String clusterId,
for (int i = 0; i < numPartitions; i++) {
TopicPartition tp = new TopicPartition(topic, i);
Node leader = nodes.get(i % nodes.size());
- List<Integer> replicaIds = Collections.singletonList(leader.id());
+ List<Integer> replicaIds = nodes.stream().map(Node::id).collect(Collectors.toList());
Review Comment:
Leaving the replication factor as implicit seems less than ideal. Perhaps we could make it an explicit argument?
--
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] dajac commented on a diff in pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
dajac commented on code in PR #12897:
URL: https://github.com/apache/kafka/pull/12897#discussion_r1034773867
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java:
##########
@@ -346,8 +346,10 @@ public void onFailure(RuntimeException e) {
FetchSessionHandler handler = sessionHandler(fetchTarget.id());
if (handler != null) {
handler.handleError(e);
+ handler.sessionTopicPartitions().forEach(subscriptions::clearPreferredReadReplica);
}
} finally {
+ metadata.requestUpdate();
Review Comment:
In `NetworkClient.processDisconnection`, it seems that we request refreshing metadata upon connection issues.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java:
##########
@@ -1944,4 +1946,9 @@ private Set<String> topicsForPartitions(Collection<TopicPartition> partitions) {
return partitions.stream().map(TopicPartition::topic).collect(Collectors.toSet());
}
+ private void prepareMetadataRefresh(TopicPartition topicPartition) {
Review Comment:
nit: `requestMetadataUpdateAndClearPreferredReadReplica` or `requestMetadataUpdate`?
--
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] dajac commented on pull request #12897: KAFKA-14379: consumer should refresh preferred read replica on update metadata
Posted by GitBox <gi...@apache.org>.
dajac commented on PR #12897:
URL: https://github.com/apache/kafka/pull/12897#issuecomment-1330712535
> A specific example is when a partition is reassigned. the consumer will get NOT_LEADER_OR_FOLLOWER which triggers a metadata update but the preferred read replica will not be refreshed as the follower is still online. it will continue to reach out to the old follower until the preferred read replica expires.
Is this really true? The replica selection is done as follow when the fetch request is constructed.
```
Node selectReadReplica(TopicPartition partition, Node leaderReplica, long currentTimeMs) {
Optional<Integer> nodeId = subscriptions.preferredReadReplica(partition, currentTimeMs);
if (nodeId.isPresent()) {
Optional<Node> node = nodeId.flatMap(id -> metadata.fetch().nodeIfOnline(partition, id));
if (node.isPresent()) {
return node.get();
} else {
log.trace("Not fetching from {} for partition {} since it is marked offline or is missing from our metadata," +
" using the leader instead.", nodeId, partition);
subscriptions.clearPreferredReadReplica(partition);
return leaderReplica;
}
} else {
return leaderReplica;
}
}
```
And this selection process resets the preferred read replica if `metadata.fetch().nodeIfOnline(partition, id)` returns an empty optional.
```
public Optional<Node> nodeIfOnline(TopicPartition partition, int id) {
Node node = nodeById(id);
PartitionInfo partitionInfo = partition(partition);
if (node != null && partitionInfo != null && !Arrays.asList(partitionInfo.offlineReplicas()).contains(node)) {
return Optional.of(node);
} else {
return Optional.empty();
}
}
```
I guess that the issue is in `nodeIfOnline` which only look into offline replicas. When a replica is moved to another broker, it is not in the offline replicas any longer. `nodeIfOnline` should also consider wether the node is a valid replica at all. Should we also fix this?
--
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