kirktrue commented on code in PR #17700:
URL: https://github.com/apache/kafka/pull/17700#discussion_r1919106233
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java:
##########
@@ -456,6 +457,36 @@ protected Map<Node, FetchSessionHandler.FetchRequestData>
prepareFetchRequests()
return
fetchable.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e ->
e.getValue().build()));
}
+ private Set<Integer> nodesWithBufferedPartitions(Set<TopicPartition>
buffered, long currentTimeMs) {
+ Set<Integer> nodesWithBufferedPartitions = new HashSet<>();
+
+ for (TopicPartition partition : buffered) {
+ if (!subscriptions.isAssigned(partition)) {
+ // It's possible that a partition with buffered data from a
previous request is now no longer
+ // assigned to the consumer, in which case just skip this
partition.
+ continue;
+ }
+
+ SubscriptionState.FetchPosition position =
subscriptions.position(partition);
Review Comment:
Code has been refactored. I went through three or four different ways of
refactoring, most of which ended up changing more stuff in the _existing_ fetch
code than I liked. The refactoring I did end up with is very minimal. PTAL.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
