kirktrue commented on code in PR #14406:
URL: https://github.com/apache/kafka/pull/14406#discussion_r1347874026
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -693,15 +1015,14 @@ private boolean refreshCommittedOffsetsIfNeeded(Timer
timer) {
log.debug("Refreshing committed offsets for partitions {}",
initializingPartitions);
try {
- final Map<TopicPartition, OffsetAndMetadata> offsets =
eventHandler.addAndGet(new OffsetFetchApplicationEvent(initializingPartitions),
timer);
+ final Map<TopicPartition, OffsetAndMetadata> offsets =
applicationEventHandler.addAndGet(new
OffsetFetchApplicationEvent(initializingPartitions), timer);
Review Comment:
TODO: find case where we _might_ be sending RPCs with empty sets.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/PrototypeAsyncConsumer.java:
##########
@@ -622,12 +837,18 @@ public void assign(Collection<TopicPartition> partitions)
{
throw new IllegalArgumentException("Topic partitions to assign
to cannot have null or empty topic");
}
- // TODO: implementation of refactored Fetcher will be included in
forthcoming commits.
- // fetcher.clearBufferedDataForUnassignedPartitions(partitions);
+ // Clear the buffered data which are not a part of newly assigned
topics
+ final Set<TopicPartition> currentTopicPartitions = new HashSet<>();
+
+ for (TopicPartition tp : subscriptions.assignedPartitions()) {
+ if (partitions.contains(tp))
+ currentTopicPartitions.add(tp);
+ }
+
+ fetchBuffer.retainAll(currentTopicPartitions);
- // assignment change event will trigger autocommit if it is configured
and the group id is specified. This is
- // to make sure offsets of topic partitions the consumer is
unsubscribing from are committed since there will
- // be no following rebalance
+ // make sure the offsets of topic partitions the consumer is
unsubscribing from
+ // are committed since there will be no following rebalance
Review Comment:
TODO: fix the comment in the code and explain about the
ApplicationEventProcessor.
--
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
