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

checkstyle/suppressions.xml

@@ -68,7 +68,7 @@
               files="(Utils|Topic|KafkaLZ4BlockOutputStream|AclData|JoinGroupRequest).java"/>
 
     <suppress checks="CyclomaticComplexity"
-              files="(ConsumerCoordinator|Fetcher|KafkaProducer|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords).java"/>
+              files="(ConsumerCoordinator|Fetcher|KafkaProducer|ConfigDef|KerberosLogin|AbstractRequest|AbstractResponse|Selector|SslFactory|SslTransportLayer|SaslClientAuthenticator|SaslClientCallbackHandler|SaslServerAuthenticator|AbstractCoordinator|TransactionManager|AbstractStickyAssignor|DefaultSslEngineFactory|Authorizer|RecordAccumulator|MemoryRecords|FetchSessionHandler).java"/>
 
     <suppress checks="JavaNCSS"
               files="(AbstractRequest|AbstractResponse|KerberosLogin|WorkerSinkTaskTest|TransactionManagerTest|SenderTest|KafkaAdminClient|ConsumerCoordinatorTest|KafkaAdminClientTest|KafkaRaftClientTest).java"/>

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;
@@ -110,6 +111,7 @@
 import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
 
 /**
  * This class manages the fetching process with the brokers.
@@ -262,7 +264,7 @@ public synchronized int sendFetches() {
                 maxVersion = ApiKeys.FETCH.latestVersion();
             }
             final FetchRequest.Builder request = FetchRequest.Builder
-                    .forConsumer(maxVersion, this.maxWaitMs, this.minBytes, data.toSend(), data.topicIds())
+                    .forConsumer(maxVersion, this.maxWaitMs, this.minBytes, data.toSend())
                     .isolationLevel(isolationLevel)
                     .setMaxBytes(this.maxBytes)
                     .metadata(data.metadata())
@@ -292,19 +294,18 @@ public void onSuccess(ClientResponse resp) {
                             }
                             if (!handler.handleResponse(response, resp.requestHeader().apiVersion())) {
                                 if (response.error() == Errors.FETCH_SESSION_TOPIC_ID_ERROR
-                                        || response.error() == Errors.UNKNOWN_TOPIC_ID
                                         || response.error() == Errors.INCONSISTENT_TOPIC_ID) {
                                     metadata.requestUpdate();
                                 }
                                 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(handler.sessionTopicNames(), 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()) {
+                                TopicIdPartition partition = entry.getKey();
                                 FetchRequest.PartitionData requestData = data.sessionPartitions().get(partition);
                                 if (requestData == null) {
                                     String message;
@@ -330,7 +331,7 @@ public void onSuccess(ClientResponse resp) {
                                     Iterator<? extends RecordBatch> batches = FetchResponse.recordsOrFail(partitionData).batches().iterator();
                                     short responseVersion = resp.requestHeader().apiVersion();
 
-                                    completedFetches.add(new CompletedFetch(partition, partitionData,
+                                    completedFetches.add(new CompletedFetch(partition.topicPartition(), partitionData,
                                             metricAggregator, batches, fetchOffset, responseVersion));
                                 }
                             }
@@ -824,6 +825,14 @@ static boolean hasUsableOffsetForLeaderEpochVersion(NodeApiVersions nodeApiVersi
         return OffsetsForLeaderEpochRequest.supportsTopicPermission(apiVersion.maxVersion());
     }
 
+    static boolean hasUsableTopicIdFetchRequestVersion(NodeApiVersions nodeApiVersions) {
+        ApiVersion apiVersion = nodeApiVersions.apiVersion(ApiKeys.FETCH);
+        if (apiVersion == null)
+            return false;
+
+        return apiVersion.maxVersion() >= 13;
+    }
+
     /**
      * For each partition which needs validation, make an asynchronous request to get the end-offsets for the partition
      * with the epoch less than or equal to the epoch the partition last saw.
@@ -1239,7 +1248,15 @@ private Map<Node, FetchSessionHandler.FetchRequestData> prepareFetchRequests() {
                     fetchable.put(node, builder);
                 }
 
-                builder.add(partition, topicIds.getOrDefault(partition.topic(), Uuid.ZERO_UUID), new FetchRequest.PartitionData(position.offset,
+                // One option is to store this in the builder and only check on the creation of the builder
+                NodeApiVersions nodeApiVersions = apiVersions.get(node.idString());
+                if (nodeApiVersions == null) {
+                    client.tryConnect(node);
+                    return emptyMap();
+                }
+                Uuid topicId = hasUsableTopicIdFetchRequestVersion(nodeApiVersions) ? topicIds.getOrDefault(partition.topic(), Uuid.ZERO_UUID) : Uuid.ZERO_UUID;
+
+                builder.add(new TopicIdPartition(topicId, partition), new FetchRequest.PartitionData(position.offset,
                     FetchRequest.INVALID_LOG_START_OFFSET, this.fetchSize,
                     position.currentLeader.epoch, Optional.empty()));
 
@@ -1353,6 +1370,9 @@ private CompletedFetch initializeCompletedFetch(CompletedFetch nextCompletedFetc
             } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
                 log.warn("Received unknown topic or partition error in fetch for partition {}", tp);
                 this.metadata.requestUpdate();
+            } else if (error == Errors.UNKNOWN_TOPIC_ID) {
+                log.warn("Received unknown topic ID error in fetch for partition {}", tp);
+                this.metadata.requestUpdate();
             } else if (error == Errors.OFFSET_OUT_OF_RANGE) {
                 Optional<Integer> clearedReplicaId = subscriptions.clearPreferredReadReplica(tp);
                 if (!clearedReplicaId.isPresent()) {

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

@@ -120,6 +120,13 @@ public FetchMetadata nextCloseExisting() {
         return new FetchMetadata(sessionId, INITIAL_EPOCH);
     }
 
+    /**
+     * Return the metadata for the next closed session response.
+     */
+    public FetchMetadata closeExisting() {
+        return new FetchMetadata(sessionId, FINAL_EPOCH);
+    }
+
     /**
      * Return the metadata for the next full fetch request.
      */