AndrewJSchofield commented on code in PR #22766:
URL: https://github.com/apache/kafka/pull/22766#discussion_r3585251396
##########
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:
Thanks for the review. I've been reworking this code in parallel to ensure I
clean up as the cluster changes. For example, if a broker is decommissioned
under a running share consumer, it needs to complete pending acks which it will
never be able to send. I'll push a new commit shortly.
--
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]
