hachikuji commented on a change in pull request #10952:
URL: https://github.com/apache/kafka/pull/10952#discussion_r668324802
##########
File path: clients/src/main/java/org/apache/kafka/clients/Metadata.java
##########
@@ -377,6 +397,12 @@ else if (metadata.error() ==
Errors.TOPIC_AUTHORIZATION_FAILED)
log.debug("Updating last seen epoch for partition {} from {}
to epoch {} from new metadata", tp, currentEpoch, newEpoch);
lastSeenLeaderEpochs.put(tp, newEpoch);
return Optional.of(partitionMetadata);
+ // If the topic ID changed, updated the metadata
+ } else if (changedTopicId) {
+ log.debug("Topic ID changed, so this topic must have been
recreated. " +
+ "Removing last seen epoch {} for the old partition {}
and adding epoch {} from new metadata", currentEpoch, tp, newEpoch);
+ lastSeenLeaderEpochs.put(tp, newEpoch);
+ return Optional.of(partitionMetadata);
Review comment:
Leaving this comment here for lack of an alternative location. This
patch takes a good first step in improving consumer behavior for the topic
recreation case. At least we are able to detect and discard the old epoch
state. In fact, it does a little more than that since, combined with the fetch
validation logic, we are likely to detect that the old fetch position is no
longer valid. Most likely this case would get raised to the user as a
`LogTruncationException`, which might not be ideal, but at least is
justifiable. However, it doesn't quite close the door on reuse of the fetch
position since it may remain valid on the recreated topic. For the full
solution, we probably need to track topicId in SubscriptionState as well so
that we can force an offset reset whenever the topicId changes. I think it
makes sense to do this in https://issues.apache.org/jira/browse/KAFKA-12975.
##########
File path: clients/src/main/java/org/apache/kafka/clients/Metadata.java
##########
@@ -391,10 +393,15 @@ else if (metadata.error() ==
Errors.TOPIC_AUTHORIZATION_FAILED)
int newEpoch = partitionMetadata.leaderEpoch.get();
// If the received leader epoch is at least the same as the
previous one, update the metadata
Integer currentEpoch = lastSeenLeaderEpochs.get(tp);
- if (currentEpoch == null || newEpoch >= currentEpoch ||
changedTopicId) {
+ if (currentEpoch == null || newEpoch >= currentEpoch) {
log.debug("Updating last seen epoch for partition {} from {}
to epoch {} from new metadata", tp, currentEpoch, newEpoch);
lastSeenLeaderEpochs.put(tp, newEpoch);
return Optional.of(partitionMetadata);
+ } else if (changedTopicId) {
+ log.debug("Topic ID changed, so this topic must have been
recreated. " +
Review comment:
nit: seems like it would be more useful for the log message to indicate
the topic ids that changed instead of the unrelated epochs.
##########
File path: clients/src/main/java/org/apache/kafka/clients/MetadataCache.java
##########
@@ -130,13 +150,30 @@ MetadataCache mergeWith(String newClusterId,
Set<String> addInvalidTopics,
Set<String> addInternalTopics,
Node newController,
+ Map<String, Uuid> topicIds,
BiPredicate<String, Boolean> retainTopic) {
Predicate<String> shouldRetainTopic = topic -> retainTopic.test(topic,
internalTopics.contains(topic));
Map<TopicPartition, PartitionMetadata> newMetadataByPartition = new
HashMap<>(addPartitions.size());
+ Map<String, Uuid> newTopicIds = new HashMap<>(topicIds.size());
+
+ // We want the most recent topic ID. We add the old one here for
retained topics and then update with newest information in the MetadataResponse
+ // we add if a new topic ID is added or remove if the request did not
support topic IDs for this topic.
+ for (Map.Entry<String, Uuid> entry : this.topicIds.entrySet()) {
+ if (shouldRetainTopic.test(entry.getKey())) {
+ newTopicIds.put(entry.getKey(), entry.getValue());
+ }
+ }
+
for (PartitionMetadata partition : addPartitions) {
newMetadataByPartition.put(partition.topicPartition, partition);
+ Uuid id = topicIds.get(partition.topic());
+ if (id != null)
+ newTopicIds.put(partition.topic(), id);
+ else
+ // Remove if the latest metadata does not have a topic ID
Review comment:
Does this still make sense in the context of 3.0, which does not have
topicId fetch logic?
##########
File path: clients/src/main/java/org/apache/kafka/clients/MetadataCache.java
##########
@@ -93,6 +104,13 @@ private MetadataCache(String clusterId,
return Optional.ofNullable(metadataByPartition.get(topicPartition));
}
+ Uuid topicId(String topicName) {
+ return topicIds.get(topicName);
+ }
+ String topicName(Uuid topicId) {
Review comment:
nit: add newline between methods
--
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
