artemlivshits commented on code in PR #12956:
URL: https://github.com/apache/kafka/pull/12956#discussion_r1040244300
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java:
##########
@@ -1154,7 +1155,9 @@ Node selectReadReplica(TopicPartition partition, Node
leaderReplica, long curren
} 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);
+ // Note that this condition may happen due to stale metadata,
so we clear preferred replica and
+ // refresh metadata.
+ requestMetadataUpdate(partition);
Review Comment:
This is new logic in addition to https://github.com/apache/kafka/pull/12897.
I'll add a unit test later.
##########
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:
From @hachikuji: 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
