AndrewJSchofield commented on code in PR #22720:
URL: https://github.com/apache/kafka/pull/22720#discussion_r3509079830
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumeRequestManager.java:
##########
@@ -248,6 +248,27 @@ public PollResult poll(long currentTimeMs) {
}
});
+ // Iterate over the session handlers again to remove any partitions
which are no longer
+ // being fetched and which also have no acknowledgements to send. This
handles the case
+ // where a node has not otherwise been picked up in this poll, but we
need to send a
+ // ShareFetch to remove the stale partitions from the share session.
+ sessionHandlers.forEach((nodeId, sessionHandler) -> {
+ Node node = cluster.nodeById(nodeId);
+ if (node != null && !handlerMap.containsKey(node) &&
!sessionHandler.sessionPartitionMap().isEmpty()) {
+ if (nodesWithPendingRequests.contains(node.id())) {
+ log.trace("Skipping fetch because previous fetch request
to {} has not been processed", nodeId);
+ } else {
+ Set<TopicPartition> currentPartitionsToFetch = new
HashSet<>(partitionsToFetch());
Review Comment:
I thought about it, but the map is going to be small. I preferred having the
logic separate of the third 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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
