lianetm commented on code in PR #22766:
URL: https://github.com/apache/kafka/pull/22766#discussion_r3582776994
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -151,120 +164,121 @@ public PollResult poll(long currentTimeMs) {
return PollResult.EMPTY;
}
+ // Iterate over the partitions to fetch, building a map from partition
to leader node ID
Map<Node, ShareSessionHandler> handlerMap = new HashMap<>();
+ Cluster cluster = metadata.fetch();
Map<String, Uuid> topicIds = metadata.topicIds();
for (TopicPartition partition : partitionsToFetch()) {
- Optional<Node> leaderOpt =
metadata.currentLeader(partition).leader;
+ TopicIdPartition tip =
partitionShareSessionTopicIdMap.get(partition);
+ if (tip == null) {
+ Uuid topicId = topicIds.get(partition.topic());
+ if (topicId == null) {
+ log.debug("Requesting metadata update for partition {}
since topic ID is missing", partition);
+ metadata.requestUpdate(false);
+ continue;
+ }
- if (leaderOpt.isEmpty()) {
- log.debug("Requesting metadata update for partition {} since
current leader node is missing", partition);
- metadata.requestUpdate(false);
- continue;
+ tip = new TopicIdPartition(topicId, partition);
+ partitionShareSessionTopicIdMap.put(partition, tip);
}
- Uuid topicId = topicIds.get(partition.topic());
- if (topicId == null) {
- log.debug("Requesting metadata update for partition {} since
topic ID is missing", partition);
- metadata.requestUpdate(false);
- continue;
+ LeaderIdAndEpoch leader = partitionShareSessionLeaderMap.get(tip);
+ if (leader == null) {
+ Metadata.LeaderAndEpoch leaderOpt =
metadata.currentLeader(partition);
+
+ if (leaderOpt.leader.isEmpty()) {
+ log.debug("Requesting metadata update for partition {}
since current leader node is missing", partition);
+ metadata.requestUpdate(false);
+ continue;
+ }
+
+ partitionShareSessionLeaderMap.put(tip, new
LeaderIdAndEpoch(leaderOpt.leader.get().id(), leaderOpt.epoch.orElse(-1)));
}
+ }
- Node node = leaderOpt.get();
- if (nodesWithPendingRequests.contains(node.id())) {
- log.trace("Skipping fetch for partition {} because previous
fetch request to {} has not been processed", partition, node.id());
+ for (TopicPartition partition : partitionsToFetch()) {
+ TopicIdPartition tip =
partitionShareSessionTopicIdMap.get(partition);
+ if (tip == null) {
+ continue;
+ }
+ LeaderIdAndEpoch leader = partitionShareSessionLeaderMap.get(tip);
+ if (leader == null) {
+ continue;
+ }
+ int nodeId = leader.leaderId;
+ Node node = cluster.nodeById(nodeId);
+ if (node == null) {
+ continue;
Review Comment:
here is the case that we had the leader in the sessionLeaderMap, but the
node is not in the cluster metadata, don't we need to clear our local leader
cache that has stale info? and also request metadata I guess, something changed
from the view we had.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -907,7 +920,14 @@ private void handleShareFetchSuccess(Node fetchTarget,
if (partitionData.currentLeader().leaderId() != -1 &&
partitionData.currentLeader().leaderEpoch() != -1) {
partitionsWithUpdatedLeaderInfo.put(tip.topicPartition(), new
Metadata.LeaderIdAndEpoch(
Optional.of(partitionData.currentLeader().leaderId()),
Optional.of(partitionData.currentLeader().leaderEpoch())));
+
+ maybeUpdateLeaderCache(tip,
partitionData.currentLeader().leaderId(),
partitionData.currentLeader().leaderEpoch());
Review Comment:
could we have the case that the partition is not assigned anymore by the
time we hit this? Metadata does consider the subscription and retains
accordingly (the retainTopic logic), so wondering if we need something similar
to be aware of the subscription while maintaining this cache?
--
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]
