dajac commented on a change in pull request #11331:
URL: https://github.com/apache/kafka/pull/11331#discussion_r743764793
##########
File path:
clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java
##########
@@ -285,52 +268,57 @@ public FetchRequestData build() {
if (nextMetadata.isFull()) {
if (log.isDebugEnabled()) {
log.debug("Built full fetch {} for node {} with {}.",
- nextMetadata, node,
partitionsToLogString(next.keySet()));
+ nextMetadata, node,
topicPartitionsToLogString(next.keySet()));
}
sessionPartitions = next;
next = null;
// Only add topic IDs to the session if we are using topic IDs.
if (canUseTopicIds) {
- sessionTopicIds = topicIds;
- sessionTopicNames = new HashMap<>(topicIds.size());
- topicIds.forEach((name, id) -> sessionTopicNames.put(id,
name));
+ Map<Uuid, Set<String>> newTopicNames =
sessionPartitions.entrySet().stream().collect(Collectors.groupingByConcurrent(entry
-> entry.getValue().topicId,
+ Collectors.mapping(entry ->
entry.getKey().topic(), Collectors.toSet())));
+
+ sessionTopicNames = new HashMap<>(newTopicNames.size());
+ // There should only be one topic name per topic ID.
+ newTopicNames.forEach((topicId, topicNamesSet) ->
topicNamesSet.forEach(topicName -> sessionTopicNames.put(topicId, topicName)));
} else {
- sessionTopicIds = new HashMap<>();
- sessionTopicNames = new HashMap<>();
+ sessionTopicNames = Collections.emptyMap();
}
- topicIds = null;
Map<TopicPartition, PartitionData> toSend =
- Collections.unmodifiableMap(new
LinkedHashMap<>(sessionPartitions));
- Map<String, Uuid> toSendTopicIds =
- Collections.unmodifiableMap(new
HashMap<>(sessionTopicIds));
- Map<Uuid, String> toSendTopicNames =
- Collections.unmodifiableMap(new
HashMap<>(sessionTopicNames));
- return new FetchRequestData(toSend, Collections.emptyList(),
toSend, toSendTopicIds, toSendTopicNames, nextMetadata, canUseTopicIds);
+ Collections.unmodifiableMap(new
LinkedHashMap<>(sessionPartitions));
+ return new FetchRequestData(toSend, Collections.emptyList(),
Collections.emptyList(), toSend, nextMetadata, canUseTopicIds);
}
- List<TopicPartition> added = new ArrayList<>();
- List<TopicPartition> removed = new ArrayList<>();
- List<TopicPartition> altered = new ArrayList<>();
+ List<TopicIdPartition> added = new ArrayList<>();
+ List<TopicIdPartition> removed = new ArrayList<>();
+ List<TopicIdPartition> altered = new ArrayList<>();
+ List<TopicIdPartition> replaced = new ArrayList<>();
for (Iterator<Entry<TopicPartition, PartitionData>> iter =
- sessionPartitions.entrySet().iterator(); iter.hasNext();
) {
+ sessionPartitions.entrySet().iterator(); iter.hasNext(); ) {
Entry<TopicPartition, PartitionData> entry = iter.next();
TopicPartition topicPartition = entry.getKey();
PartitionData prevData = entry.getValue();
PartitionData nextData = next.remove(topicPartition);
if (nextData != null) {
- if (!prevData.equals(nextData)) {
+ // We basically check if the new partition had the same
topic ID. If not,
+ // we add it to the "replaced" set.
+ if (!prevData.topicId.equals(nextData.topicId) &&
!prevData.topicId.equals(Uuid.ZERO_UUID)) {
+ // Re-add the replaced partition to the end of 'next'
+ next.put(topicPartition, nextData);
+ entry.setValue(nextData);
+ replaced.add(new TopicIdPartition(prevData.topicId,
topicPartition));
+ } else if (!prevData.equals(nextData)) {
// Re-add the altered partition to the end of 'next'
next.put(topicPartition, nextData);
entry.setValue(nextData);
- altered.add(topicPartition);
+ altered.add(new TopicIdPartition(nextData.topicId,
topicPartition));
}
} else {
// Remove this partition from the session.
iter.remove();
// Indicate that we no longer want to listen to this
partition.
- removed.add(topicPartition);
+ removed.add(new TopicIdPartition(prevData.topicId,
topicPartition));
// If we do not have this topic ID in the builder or the
session, we can not use topic IDs.
- if (canUseTopicIds &&
!topicIds.containsKey(topicPartition.topic()) &&
!sessionTopicIds.containsKey(topicPartition.topic()))
+ if (canUseTopicIds && prevData.topicId == Uuid.ZERO_UUID)
Review comment:
Should we use `equals` instead of `==`? We use `equals` at L304 btw.
--
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
