jolshan commented on a change in pull request #9944:
URL: https://github.com/apache/kafka/pull/9944#discussion_r565612146
##########
File path:
clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java
##########
@@ -186,23 +241,37 @@ public String toString() {
* incremental fetch requests (see below).
*/
private LinkedHashMap<TopicPartition, PartitionData> next;
+ private Map<String, Uuid> topicIds;
+ private Map<Uuid, String> topicNames;
+ private Map<String, Integer> partitionsPerTopic;
private final boolean copySessionPartitions;
Builder() {
this.next = new LinkedHashMap<>();
+ this.topicIds = new HashMap<>();
+ this.topicNames = new HashMap<>();
+ this.partitionsPerTopic = new HashMap<>();
this.copySessionPartitions = true;
}
Builder(int initialSize, boolean copySessionPartitions) {
this.next = new LinkedHashMap<>(initialSize);
+ this.topicIds = new HashMap<>(initialSize);
+ this.topicNames = new HashMap<>(initialSize);
+ this.partitionsPerTopic = new HashMap<>(initialSize);
this.copySessionPartitions = copySessionPartitions;
}
/**
* Mark that we want data from this partition in the upcoming fetch.
*/
- public void add(TopicPartition topicPartition, PartitionData data) {
- next.put(topicPartition, data);
+ public void add(TopicPartition topicPartition, Uuid id, PartitionData
data) {
+ if (next.put(topicPartition, data) == null)
+ partitionsPerTopic.merge(topicPartition.topic(), 1, (prev,
next) -> prev + next);
Review comment:
I think I may want to do this in a simpler way. I want to keep track if
we have IDs for all the topics and I'm not sure if there is a better way to
figure out when a topic is no longer in a session besides checking all the
topic partitions.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
