KAFKA-13111: Re-evaluate Fetch Sessions when using topic IDs

clients/src/main/java/org/apache/kafka/clients/FetchSessionHandler.java

@@ -18,6 +18,7 @@
 package org.apache.kafka.clients;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.requests.FetchMetadata;
@@ -39,6 +40,7 @@
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import static org.apache.kafka.common.requests.FetchMetadata.INVALID_SESSION_ID;
 
@@ -536,7 +538,8 @@ public boolean handleResponse(FetchResponse response, short version) {
             }
             return false;
         }
-        Set<TopicPartition> topicPartitions = response.responseData(sessionTopicNames, version).keySet();
+        Set<TopicPartition> topicPartitions = response.responseData(sessionTopicNames, version).keySet().stream().map(
+                TopicIdPartition::topicPartition).collect(Collectors.toSet());
         if (nextMetadata.isFull()) {
             if (topicPartitions.isEmpty() && response.throttleTimeMs() > 0) {
                 // Normally, an empty full fetch response would be invalid.  However, KIP-219

clients/src/main/java/org/apache/kafka/clients/consumer/internals/Fetcher.java

@@ -38,6 +38,7 @@
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.CorruptRecordException;
 import org.apache.kafka.common.errors.InvalidTopicException;
@@ -299,12 +300,12 @@ public void onSuccess(ClientResponse resp) {
                                 return;
                             }
 
-                            Map<TopicPartition, FetchResponseData.PartitionData> responseData = response.responseData(data.topicNames(), resp.requestHeader().apiVersion());
-                            Set<TopicPartition> partitions = new HashSet<>(responseData.keySet());
+                            Map<TopicIdPartition, FetchResponseData.PartitionData> responseData = response.responseData(data.topicNames(), resp.requestHeader().apiVersion());
+                            Set<TopicPartition> partitions = new HashSet<>(responseData.keySet().stream().map(TopicIdPartition::topicPartition).collect(Collectors.toSet()));
                             FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions);
 
-                            for (Map.Entry<TopicPartition, FetchResponseData.PartitionData> entry : responseData.entrySet()) {
-                                TopicPartition partition = entry.getKey();
+                            for (Map.Entry<TopicIdPartition, FetchResponseData.PartitionData> entry : responseData.entrySet()) {
+                                TopicPartition partition = entry.getKey().topicPartition();
                                 FetchRequest.PartitionData requestData = data.sessionPartitions().get(partition);
                                 if (requestData == null) {
                                     String message;

clients/src/main/java/org/apache/kafka/common/requests/FetchRequest.java

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.common.IsolationLevel;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.errors.UnknownTopicIdException;
 import org.apache.kafka.common.message.FetchRequestData;
@@ -47,8 +48,8 @@ public class FetchRequest extends AbstractRequest {
     public static final long INVALID_LOG_START_OFFSET = -1L;
 
     private final FetchRequestData data;
-    private volatile LinkedHashMap<TopicPartition, PartitionData> fetchData = null;
-    private volatile List<TopicPartition> toForget = null;
+    private volatile LinkedHashMap<TopicIdPartition, PartitionData> fetchData = null;
+    private volatile List<TopicIdPartition> toForget = null;
 
     // This is an immutable read-only structures derived from FetchRequestData
     private final FetchMetadata metadata;
@@ -315,7 +316,7 @@ public int maxBytes() {
     // For versions < 13, builds the partitionData map using only the FetchRequestData.
     // For versions 13+, builds the partitionData map using both the FetchRequestData and a mapping of topic IDs to names.
     // Throws UnknownTopicIdException for versions 13+ if the topic ID was unknown to the server.
-    public Map<TopicPartition, PartitionData> fetchData(Map<Uuid, String> topicNames) throws UnknownTopicIdException {
+    public Map<TopicIdPartition, PartitionData> fetchData(Map<Uuid, String> topicNames) throws UnknownTopicIdException {
         if (fetchData == null) {
             synchronized (this) {
                 if (fetchData == null) {
@@ -328,22 +329,18 @@ public Map<TopicPartition, PartitionData> fetchData(Map<Uuid, String> topicNames
                         } else {
                             name = topicNames.get(fetchTopic.topicId());
                         }
-                        if (name != null) {
-                            // If topic name is resolved, simply add to fetchData map
-                            fetchTopic.partitions().forEach(fetchPartition ->
-                                    fetchData.put(new TopicPartition(name, fetchPartition.partition()),
-                                            new PartitionData(
-                                                    fetchPartition.fetchOffset(),
-                                                    fetchPartition.logStartOffset(),
-                                                    fetchPartition.partitionMaxBytes(),
-                                                    optionalEpoch(fetchPartition.currentLeaderEpoch()),
-                                                    optionalEpoch(fetchPartition.lastFetchedEpoch())
-                                            )
-                                    )
-                            );
-                        } else {
-                            throw new UnknownTopicIdException(String.format("Topic Id %s in FetchRequest was unknown to the server", fetchTopic.topicId()));
-                        }
+                        // If topic name is resolved, simply add to fetchData map
+                        fetchTopic.partitions().forEach(fetchPartition ->
+                                fetchData.put(new TopicIdPartition(fetchTopic.topicId(), new TopicPartition(name, fetchPartition.partition())),
+                                        new PartitionData(
+                                                fetchPartition.fetchOffset(),
+                                                fetchPartition.logStartOffset(),
+                                                fetchPartition.partitionMaxBytes(),
+                                                optionalEpoch(fetchPartition.currentLeaderEpoch()),
+                                                optionalEpoch(fetchPartition.lastFetchedEpoch())
+                                        )
+                                )
+                        );
                     });
                 }
             }
@@ -352,7 +349,7 @@ public Map<TopicPartition, PartitionData> fetchData(Map<Uuid, String> topicNames
     }
 
     // For versions 13+, throws UnknownTopicIdException if the topic ID was unknown to the server.
-    public List<TopicPartition> forgottenTopics(Map<Uuid, String> topicNames) throws UnknownTopicIdException {
+    public List<TopicIdPartition> forgottenTopics(Map<Uuid, String> topicNames) throws UnknownTopicIdException {
         if (toForget == null) {
             synchronized (this) {
                 if (toForget == null) {
@@ -367,7 +364,7 @@ public List<TopicPartition> forgottenTopics(Map<Uuid, String> topicNames) throws
                         if (name == null) {
                             throw new UnknownTopicIdException(String.format("Topic Id %s in FetchRequest was unknown to the server", forgottenTopic.topicId()));
                         }
-                        forgottenTopic.partitions().forEach(partitionId -> toForget.add(new TopicPartition(name, partitionId)));
+                        forgottenTopic.partitions().forEach(partitionId -> toForget.add(new TopicIdPartition(forgottenTopic.topicId(), new TopicPartition(name, partitionId))));
                     });
                 }
             }

clients/src/main/java/org/apache/kafka/common/requests/FetchResponse.java

@@ -17,6 +17,7 @@
 package org.apache.kafka.common.requests;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicIdPartition;
 import org.apache.kafka.common.Uuid;
 import org.apache.kafka.common.message.FetchResponseData;
 import org.apache.kafka.common.protocol.ApiKeys;
@@ -72,7 +73,7 @@ public class FetchResponse extends AbstractResponse {
 
     private final FetchResponseData data;
     // we build responseData when needed.
-    private volatile LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData = null;
+    private volatile LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> responseData = null;
 
     @Override
     public FetchResponseData data() {
@@ -96,7 +97,7 @@ public Errors error() {
         return Errors.forCode(data.errorCode());
     }
 
-    public LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData(Map<Uuid, String> topicNames, short version) {
+    public LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> responseData(Map<Uuid, String> topicNames, short version) {
         if (responseData == null) {
             synchronized (this) {
                 if (responseData == null) {
@@ -110,7 +111,7 @@ public LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseDa
                         }
                         if (name != null) {
                             topicResponse.partitions().forEach(partition ->
-                                    responseData.put(new TopicPartition(name, partition.partitionIndex()), partition));
+                                    responseData.put(new TopicIdPartition(topicResponse.topicId(), new TopicPartition(name, partition.partitionIndex())), partition));
                         }
                     });
                 }
@@ -154,16 +155,14 @@ public Set<Uuid> topicIds() {
      *
      * @param version       The version of the response to use.
      * @param partIterator  The partition iterator.
-     * @param topicIds      The mapping from topic name to topic ID.
      * @return              The response size in bytes.
      */
     public static int sizeOf(short version,
-                             Iterator<Map.Entry<TopicPartition,
-                             FetchResponseData.PartitionData>> partIterator,
-                             Map<String, Uuid> topicIds) {
+                             Iterator<Map.Entry<TopicIdPartition,
+                             FetchResponseData.PartitionData>> partIterator) {
         // Since the throttleTimeMs and metadata field sizes are constant and fixed, we can
         // use arbitrary values here without affecting the result.
-        FetchResponseData data = toMessage(Errors.NONE, 0, INVALID_SESSION_ID, partIterator, topicIds);
+        FetchResponseData data = toMessage(Errors.NONE, 0, INVALID_SESSION_ID, partIterator);
         ObjectSerializationCache cache = new ObjectSerializationCache();
         return 4 + data.size(cache, version);
     }
@@ -226,40 +225,55 @@ public static int recordsSize(FetchResponseData.PartitionData partition) {
     public static FetchResponse of(Errors error,
                                    int throttleTimeMs,
                                    int sessionId,
-                                   LinkedHashMap<TopicPartition, FetchResponseData.PartitionData> responseData,
-                                   Map<String, Uuid> topicIds) {
-        return new FetchResponse(toMessage(error, throttleTimeMs, sessionId, responseData.entrySet().iterator(), topicIds));
+                                   LinkedHashMap<TopicIdPartition, FetchResponseData.PartitionData> responseData) {
+        return new FetchResponse(toMessage(error, throttleTimeMs, sessionId, responseData.entrySet().iterator()));
+    }
+
+    private static boolean matchingTopic(FetchResponseData.FetchableTopicResponse previousTopic, TopicIdPartition currentTopic) {
+        if (previousTopic == null)
+            return false;
+        if (!previousTopic.topicId().equals(Uuid.ZERO_UUID))
+            return previousTopic.topicId().equals(currentTopic.topicId());
+        else
+            return previousTopic.topic().equals(currentTopic.topicPartition().topic());
+
     }
 
     private static FetchResponseData toMessage(Errors error,
                                                int throttleTimeMs,
                                                int sessionId,
-                                               Iterator<Map.Entry<TopicPartition, FetchResponseData.PartitionData>> partIterator,
-                                               Map<String, Uuid> topicIds) {
+                                               Iterator<Map.Entry<TopicIdPartition, FetchResponseData.PartitionData>> partIterator) {
         List<FetchResponseData.FetchableTopicResponse> topicResponseList = new ArrayList<>();
-        partIterator.forEachRemaining(entry -> {
+        boolean hasInconsistentTopicId = false;
+        while (partIterator.hasNext()) {
+            Map.Entry<TopicIdPartition, FetchResponseData.PartitionData> entry = partIterator.next();
             FetchResponseData.PartitionData partitionData = entry.getValue();
             // Since PartitionData alone doesn't know the partition ID, we set it here
-            partitionData.setPartitionIndex(entry.getKey().partition());
+            partitionData.setPartitionIndex(entry.getKey().topicPartition().partition());
             // We have to keep the order of input topic-partition. Hence, we batch the partitions only if the last
             // batch is in the same topic group.
+            if (partitionData.errorCode() == Errors.INCONSISTENT_TOPIC_ID.code()) {
+                hasInconsistentTopicId = true;
+            }
             FetchResponseData.FetchableTopicResponse previousTopic = topicResponseList.isEmpty() ? null
-                : topicResponseList.get(topicResponseList.size() - 1);
-            if (previousTopic != null && previousTopic.topic().equals(entry.getKey().topic()))
+                    : topicResponseList.get(topicResponseList.size() - 1);
+            if (matchingTopic(previousTopic, entry.getKey()))
                 previousTopic.partitions().add(partitionData);
             else {
                 List<FetchResponseData.PartitionData> partitionResponses = new ArrayList<>();
                 partitionResponses.add(partitionData);
                 topicResponseList.add(new FetchResponseData.FetchableTopicResponse()
-                    .setTopic(entry.getKey().topic())
-                    .setTopicId(topicIds.getOrDefault(entry.getKey().topic(), Uuid.ZERO_UUID))
-                    .setPartitions(partitionResponses));
+                        .setTopic(entry.getKey().topicPartition().topic())
+                        .setTopicId(entry.getKey().topicId())
+                        .setPartitions(partitionResponses));
             }
-        });
+        }
+
+        short errorCode = hasInconsistentTopicId ? Errors.INCONSISTENT_TOPIC_ID.code() : error.code();
 
         return new FetchResponseData()
             .setThrottleTimeMs(throttleTimeMs)
-            .setErrorCode(error.code())
+            .setErrorCode(errorCode)
             .setSessionId(sessionId)
             .setResponses(topicResponseList);
     }