ShivsundarR commented on code in PR #22720:
URL: https://github.com/apache/kafka/pull/22720#discussion_r3507606088
##########
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:
Is it possible to maybe include this logic the previous loop (the one
processing the unsent acknowledgements) just after we do the acknowledgements
check to avoid passing over the main map twice?
Since we are having the `!handlerMap.containsKey(node)` check here, it
should not coincide with any partitions added in that acknowledgements loop
pass.
--
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]
