KAFKA-10817; Add clusterId validation to Fetch handling

clients/src/main/java/org/apache/kafka/common/errors/InconsistentClusterIdException.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class InconsistentClusterIdException extends ApiException {
+
+    public InconsistentClusterIdException(String message) {
+        super(message);
+    }
+
+    public InconsistentClusterIdException(String message, Throwable throwable) {
+        super(message, throwable);
+    }
+}

clients/src/main/java/org/apache/kafka/common/protocol/Errors.java

@@ -49,6 +49,7 @@
 import org.apache.kafka.common.errors.InconsistentGroupProtocolException;
 import org.apache.kafka.common.errors.InconsistentTopicIdException;
 import org.apache.kafka.common.errors.InconsistentVoterSetException;
+import org.apache.kafka.common.errors.InconsistentClusterIdException;
 import org.apache.kafka.common.errors.InvalidCommitOffsetSizeException;
 import org.apache.kafka.common.errors.InvalidConfigurationException;
 import org.apache.kafka.common.errors.InvalidFetchSessionEpochException;
@@ -356,7 +357,8 @@ public enum Errors {
         PositionOutOfRangeException::new),
     UNKNOWN_TOPIC_ID(100, "This server does not host this topic ID.", UnknownTopicIdException::new),
     DUPLICATE_BROKER_REGISTRATION(101, "This broker ID is already in use.", DuplicateBrokerRegistrationException::new),
-    INCONSISTENT_TOPIC_ID(102, "The log's topic ID did not match the topic ID in the request", InconsistentTopicIdException::new);
+    INCONSISTENT_TOPIC_ID(102, "The log's topic ID did not match the topic ID in the request", InconsistentTopicIdException::new),
+    INCONSISTENT_CLUSTER_ID(103, "The clusterId in the request does not match that found on the server", InconsistentClusterIdException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

clients/src/main/resources/common/message/FetchRequest.json

@@ -50,7 +50,8 @@
   "validVersions": "0-12",
   "flexibleVersions": "12+",
   "fields": [
-    { "name": "ClusterId", "type": "string", "versions": "12+", "nullableVersions": "12+", "default": "null", "taggedVersions": "12+", "tag": 0,
+    { "name": "ClusterId", "type": "string", "versions": "12+", "nullableVersions": "12+", "default": "null",
+      "taggedVersions": "12+", "tag": 0, "ignorable": true,
       "about": "The clusterId if known. This is used to validate metadata fetches prior to broker registration." },
     { "name": "ReplicaId", "type": "int32", "versions": "0+",
       "about": "The broker ID of the follower, of -1 if this request is from a consumer." },

core/src/main/scala/kafka/raft/RaftManager.scala

@@ -200,6 +200,7 @@ class KafkaRaftManager[T](
       metrics,
       expirationService,
       logContext,
+      metaProperties.clusterId.toString,
       OptionalInt.of(config.nodeId),
       raftConfig
     )

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -148,6 +148,7 @@ public class KafkaRaftClient<T> implements RaftClient<T> {
     private final LogContext logContext;
     private final Time time;
     private final int fetchMaxWaitMs;
+    private final String clusterId;
     private final OptionalInt nodeId;
     private final NetworkChannel channel;
     private final ReplicatedLog log;
@@ -184,6 +185,7 @@ public KafkaRaftClient(
         Metrics metrics,
         ExpirationService expirationService,
         LogContext logContext,
+        String clusterId,
         OptionalInt nodeId,
         RaftConfig raftConfig
     ) {
@@ -197,6 +199,7 @@ public KafkaRaftClient(
             metrics,
             expirationService,
             FETCH_MAX_WAIT_MS,
+            clusterId,
             nodeId,
             logContext,
             new Random(),
@@ -214,6 +217,7 @@ public KafkaRaftClient(
         Metrics metrics,
         ExpirationService expirationService,
         int fetchMaxWaitMs,
+        String clusterId,
         OptionalInt nodeId,
         LogContext logContext,
         Random random,
@@ -228,6 +232,7 @@ public KafkaRaftClient(
         this.fetchPurgatory = new ThresholdPurgatory<>(expirationService);
         this.appendPurgatory = new ThresholdPurgatory<>(expirationService);
         this.time = time;
+        this.clusterId = clusterId;
         this.nodeId = nodeId;
         this.metrics = metrics;
         this.fetchMaxWaitMs = fetchMaxWaitMs;
@@ -940,6 +945,14 @@ private FetchResponseData buildEmptyFetchResponse(
         );
     }
 
+    private boolean hasValidClusterId(FetchRequestData request) {
+        // We don't enforce the cluster id if it is not provided.
+        if (request.clusterId() == null) {
+            return true;
+        }
+        return clusterId.equals(request.clusterId());
+    }
+
     /**
      * Handle a Fetch request. The fetch offset and last fetched epoch are always
      * validated against the current log. In the case that they do not match, the response will
@@ -959,6 +972,10 @@ private CompletableFuture<FetchResponseData> handleFetchRequest(
     ) {
         FetchRequestData request = (FetchRequestData) requestMetadata.data;
 
+        if (!hasValidClusterId(request)) {
+            return completedFuture(new FetchResponseData().setErrorCode(Errors.INCONSISTENT_CLUSTER_ID.code()));
+        }
+
         if (!hasValidTopicPartition(request, log.topicPartition())) {
             // Until we support multi-raft, we treat topic partition mismatches as invalid requests
             return completedFuture(new FetchResponseData().setErrorCode(Errors.INVALID_REQUEST.code()));
@@ -1766,6 +1783,7 @@ private FetchRequestData buildFetchRequest() {
         });
         return request
             .setMaxWaitMs(fetchMaxWaitMs)
+            .setClusterId(clusterId.toString())
             .setReplicaId(quorum.localIdOrSentinel());
     }
 

raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java

@@ -74,7 +74,7 @@ public void testFetchRequestOffsetLessThanLogStart() throws Exception {
         // Advance the highWatermark
         context.deliverRequest(context.fetchRequest(epoch, otherNodeId, localLogEndOffset, epoch, 0));
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
         assertEquals(localLogEndOffset, context.client.highWatermark().getAsLong());
 
         OffsetAndEpoch snapshotId = new OffsetAndEpoch(localLogEndOffset, epoch);
@@ -89,7 +89,7 @@ public void testFetchRequestOffsetLessThanLogStart() throws Exception {
         // Send Fetch request less than start offset
         context.deliverRequest(context.fetchRequest(epoch, otherNodeId, 0, epoch, 0));
         context.pollUntilResponse();
-        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchResponse();
+        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchPartitionResponse();
         assertEquals(Errors.NONE, Errors.forCode(partitionResponse.errorCode()));
         assertEquals(epoch, partitionResponse.currentLeader().leaderEpoch());
         assertEquals(localId, partitionResponse.currentLeader().leaderId());
@@ -119,7 +119,7 @@ public void testFetchRequestWithLargerLastFetchedEpoch() throws Exception {
         long localLogEndOffset = context.log.endOffset().offset;
         context.deliverRequest(context.fetchRequest(epoch, otherNodeId, localLogEndOffset, epoch, 0));
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
         assertEquals(localLogEndOffset, context.client.highWatermark().getAsLong());
 
         // Create a snapshot at the high watermark
@@ -135,7 +135,7 @@ public void testFetchRequestWithLargerLastFetchedEpoch() throws Exception {
         // It is an invalid request to send an last fetched epoch greater than the current epoch
         context.deliverRequest(context.fetchRequest(epoch, otherNodeId, oldestSnapshotId.offset + 1, epoch + 1, 0));
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.INVALID_REQUEST, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.INVALID_REQUEST, epoch, OptionalInt.of(localId));
     }
 
     @Test
@@ -162,7 +162,7 @@ public void testFetchRequestTruncateToLogStart() throws Exception {
             context.fetchRequest(epoch, syncNodeId, context.log.endOffset().offset, epoch, 0)
         );
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
         assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong());
 
         // Create a snapshot at the high watermark
@@ -176,7 +176,7 @@ public void testFetchRequestTruncateToLogStart() throws Exception {
             context.fetchRequest(epoch, otherNodeId, oldestSnapshotId.offset + 1, oldestSnapshotId.epoch + 1, 0)
         );
         context.pollUntilResponse();
-        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchResponse();
+        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchPartitionResponse();
         assertEquals(Errors.NONE, Errors.forCode(partitionResponse.errorCode()));
         assertEquals(epoch, partitionResponse.currentLeader().leaderEpoch());
         assertEquals(localId, partitionResponse.currentLeader().leaderId());
@@ -209,7 +209,7 @@ public void testFetchRequestAtLogStartOffsetWithValidEpoch() throws Exception {
             context.fetchRequest(epoch, syncNodeId, context.log.endOffset().offset, epoch, 0)
         );
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
         assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong());
 
         // Create a snapshot at the high watermark
@@ -223,7 +223,7 @@ public void testFetchRequestAtLogStartOffsetWithValidEpoch() throws Exception {
             context.fetchRequest(epoch, otherNodeId, oldestSnapshotId.offset, oldestSnapshotId.epoch, 0)
         );
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
     }
 
     @Test
@@ -251,7 +251,7 @@ public void testFetchRequestAtLogStartOffsetWithInvalidEpoch() throws Exception
             context.fetchRequest(epoch, syncNodeId, context.log.endOffset().offset, epoch, 0)
         );
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
         assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong());
 
         // Create a snapshot at the high watermark
@@ -265,7 +265,7 @@ public void testFetchRequestAtLogStartOffsetWithInvalidEpoch() throws Exception
             context.fetchRequest(epoch, otherNodeId, oldestSnapshotId.offset, oldestSnapshotId.epoch + 1, 0)
         );
         context.pollUntilResponse();
-        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchResponse();
+        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchPartitionResponse();
         assertEquals(Errors.NONE, Errors.forCode(partitionResponse.errorCode()));
         assertEquals(epoch, partitionResponse.currentLeader().leaderEpoch());
         assertEquals(localId, partitionResponse.currentLeader().leaderId());
@@ -298,7 +298,7 @@ public void testFetchRequestWithLastFetchedEpochLessThanOldestSnapshot() throws
             context.fetchRequest(epoch, syncNodeId, context.log.endOffset().offset, epoch, 0)
         );
         context.pollUntilResponse();
-        context.assertSentFetchResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+        context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
         assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong());
 
         // Create a snapshot at the high watermark
@@ -318,7 +318,7 @@ public void testFetchRequestWithLastFetchedEpochLessThanOldestSnapshot() throws
             )
         );
         context.pollUntilResponse();
-        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchResponse();
+        FetchResponseData.FetchablePartitionResponse partitionResponse = context.assertSentFetchPartitionResponse();
         assertEquals(Errors.NONE, Errors.forCode(partitionResponse.errorCode()));
         assertEquals(epoch, partitionResponse.currentLeader().leaderEpoch());
         assertEquals(localId, partitionResponse.currentLeader().leaderId());