KAFKA-10427: Fetch snapshot

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

@@ -0,0 +1,31 @@
+/*
+ * 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 SnapshotNotFoundException extends ApiException {
+
+    private static final long serialVersionUID = 1;
+
+    public SnapshotNotFoundException(String s) {
+        super(s);
+    }
+
+    public SnapshotNotFoundException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

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

@@ -83,6 +83,8 @@
 import org.apache.kafka.common.message.ExpireDelegationTokenResponseData;
 import org.apache.kafka.common.message.FetchRequestData;
 import org.apache.kafka.common.message.FetchResponseData;
+import org.apache.kafka.common.message.FetchSnapshotRequestData;
+import org.apache.kafka.common.message.FetchSnapshotResponseData;
 import org.apache.kafka.common.message.FindCoordinatorRequestData;
 import org.apache.kafka.common.message.FindCoordinatorResponseData;
 import org.apache.kafka.common.message.HeartbeatRequestData;
@@ -254,7 +256,9 @@ public Struct parseResponse(short version, ByteBuffer buffer) {
     ALTER_ISR(56, "AlterIsr", AlterIsrRequestData.SCHEMAS, AlterIsrResponseData.SCHEMAS),
     UPDATE_FEATURES(57, "UpdateFeatures",
         UpdateFeaturesRequestData.SCHEMAS, UpdateFeaturesResponseData.SCHEMAS, true),
-    ENVELOPE(58, "Envelope", true, false, EnvelopeRequestData.SCHEMAS, EnvelopeResponseData.SCHEMAS);
+    ENVELOPE(58, "Envelope", true, false, EnvelopeRequestData.SCHEMAS, EnvelopeResponseData.SCHEMAS),
+    FETCH_SNAPSHOT(59, "FetchSnapshot", false, false,
+        FetchSnapshotRequestData.SCHEMAS, FetchSnapshotResponseData.SCHEMAS);
 
     private static final ApiKeys[] ID_TO_TYPE;
     private static final int MIN_API_KEY = 0;

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

@@ -95,6 +95,7 @@
 import org.apache.kafka.common.errors.RetriableException;
 import org.apache.kafka.common.errors.SaslAuthenticationException;
 import org.apache.kafka.common.errors.SecurityDisabledException;
+import org.apache.kafka.common.errors.SnapshotNotFoundException;
 import org.apache.kafka.common.errors.StaleBrokerEpochException;
 import org.apache.kafka.common.errors.ThrottlingQuotaExceededException;
 import org.apache.kafka.common.errors.TimeoutException;
@@ -341,7 +342,8 @@ public enum Errors {
     INVALID_UPDATE_VERSION(95, "The given update version was invalid.", InvalidUpdateVersionException::new),
     FEATURE_UPDATE_FAILED(96, "Unable to update finalized features due to an unexpected server error.", FeatureUpdateFailedException::new),
     PRINCIPAL_DESERIALIZATION_FAILURE(97, "Request principal deserialization failed during forwarding. " +
-         "This indicates an internal error on the broker cluster security setup.", PrincipalDeserializationException::new);
+         "This indicates an internal error on the broker cluster security setup.", PrincipalDeserializationException::new),
+    SNAPSHOT_NOT_FOUND(98, "Requested snapshot was not found", SnapshotNotFoundException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

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

@@ -0,0 +1,97 @@
+/*
+ * 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.requests;
+
+import java.util.Collections;
+import java.util.Optional;
+import java.util.function.UnaryOperator;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.message.FetchSnapshotRequestData;
+import org.apache.kafka.common.message.FetchSnapshotResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.Message;
+
+final public class FetchSnapshotRequest extends AbstractRequest {
+    public final FetchSnapshotRequestData data;
+
+    public FetchSnapshotRequest(FetchSnapshotRequestData data) {
+        super(ApiKeys.FETCH_SNAPSHOT, (short) (FetchSnapshotRequestData.SCHEMAS.length - 1));
+        this.data = data;
+    }
+
+    @Override
+    public FetchSnapshotResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        return new FetchSnapshotResponse(
+            new FetchSnapshotResponseData()
+                .setThrottleTimeMs(throttleTimeMs)
+                .setErrorCode(Errors.forException(e).code())
+        );
+    }
+
+    @Override
+    protected Message data() {
+        return data;
+    }
+
+    /**
+     * Creates a FetchSnapshotRequestData with a single PartitionSnapshot for the topic partition.
+     *
+     * The partition index will already be populated when calling operator.
+     *
+     * @param topicPartition the topic partition to include
+     * @param operator unary operator responsible for populating all the appropriate fields
+     * @return the created fetch snapshot request data
+     */
+    public static FetchSnapshotRequestData singleton(
+        TopicPartition topicPartition,
+        UnaryOperator<FetchSnapshotRequestData.PartitionSnapshot> operator
+    ) {
+        FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = operator.apply(
+            new FetchSnapshotRequestData.PartitionSnapshot().setPartition(topicPartition.partition())
+        );
+
+        return new FetchSnapshotRequestData()
+            .setTopics(
+                Collections.singletonList(
+                    new FetchSnapshotRequestData.TopicSnapshot()
+                        .setName(topicPartition.topic())
+                        .setPartitions(Collections.singletonList(partitionSnapshot))
+                )
+            );
+    }
+
+    /**
+     * Finds the PartitionSnapshot for a given topic partition.
+     *
+     * @param data the fetch snapshot request data
+     * @param topicPartition the topic partition to find
+     * @return the request partition snapshot if found, otherwise an empty Optional
+     */
+    public static Optional<FetchSnapshotRequestData.PartitionSnapshot> forTopicPartition(
+        FetchSnapshotRequestData data,
+        TopicPartition topicPartition
+    ) {
+        return data
+            .topics()
+            .stream()
+            .filter(topic -> topic.name().equals(topicPartition.topic()))
+            .flatMap(topic -> topic.partitions().stream())
+            .filter(partition -> partition.partition() == topicPartition.partition())
+            .findAny();
+    }
+}

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

@@ -0,0 +1,124 @@
+/*
+ * 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.requests;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.UnaryOperator;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.message.FetchSnapshotResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.Message;
+
+final public class FetchSnapshotResponse extends AbstractResponse {
+    public final FetchSnapshotResponseData data;
+
+    public FetchSnapshotResponse(FetchSnapshotResponseData data) {
+        super(ApiKeys.FETCH_SNAPSHOT);
+
+        this.data = data;
+    }
+
+    @Override
+    public Map<Errors, Integer> errorCounts() {
+        Map<Errors, Integer> errors = new HashMap<>();
+
+        Errors topLevelError = Errors.forCode(data.errorCode());
+        if (topLevelError != Errors.NONE) {
+            errors.put(topLevelError, 1);
+        }
+
+        for (FetchSnapshotResponseData.TopicSnapshot topicResponse : data.topics()) {
+            for (FetchSnapshotResponseData.PartitionSnapshot partitionResponse : topicResponse.partitions()) {
+                errors.compute(Errors.forCode(partitionResponse.errorCode()),
+                    (error, count) -> count == null ? 1 : count + 1);
+            }
+        }
+
+        return errors;
+    }
+
+    @Override
+    public int throttleTimeMs() {
+        return data.throttleTimeMs();
+    }
+
+    @Override
+    protected Message data() {
+        return data;
+    }
+
+    /**
+     * Creates a FetchSnapshotResponseData with a top level error.
+     *
+     * @param error the top level error
+     * @return the created fetch snapshot response data
+     */
+    public static FetchSnapshotResponseData withTopError(Errors error) {
+        return new FetchSnapshotResponseData().setErrorCode(error.code());
+    }
+
+    /**
+     * Creates a FetchSnapshotResponseData with a single PartitionSnapshot for the topic partition.
+     *
+     * The partition index will already by populated when calling operator.
+     *
+     * @param topicPartition the topic partition to include
+     * @param operator unary operator responsible for populating all of the appropriate fields
+     * @return the created fetch snapshot response data
+     */
+    public static FetchSnapshotResponseData singleton(
+        TopicPartition topicPartition,
+        UnaryOperator<FetchSnapshotResponseData.PartitionSnapshot> operator
+    ) {
+        FetchSnapshotResponseData.PartitionSnapshot partitionSnapshot = operator.apply(
+            new FetchSnapshotResponseData.PartitionSnapshot().setIndex(topicPartition.partition())
+        );
+
+        return new FetchSnapshotResponseData()
+            .setTopics(
+                Collections.singletonList(
+                    new FetchSnapshotResponseData.TopicSnapshot()
+                        .setName(topicPartition.topic())
+                        .setPartitions(Collections.singletonList(partitionSnapshot))
+                )
+            );
+    }
+
+    /**
+     * Finds the PartitionSnapshot for a given topic partition.
+     *
+     * @param data the fetch snapshot response data
+     * @param topicPartition the topic partition to find
+     * @return the response partition snapshot if found, otherwise an empty Optional
+     */
+    public static Optional<FetchSnapshotResponseData.PartitionSnapshot> forTopicPartition(
+        FetchSnapshotResponseData data,
+        TopicPartition topicPartition
+    ) {
+        return data
+            .topics()
+            .stream()
+            .filter(topic -> topic.name().equals(topicPartition.topic()))
+            .flatMap(topic -> topic.partitions().stream())
+            .filter(parition -> parition.index() == topicPartition.partition())
+            .findAny();
+    }
+}

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

@@ -78,6 +78,13 @@
           { "name": "LeaderEpoch", "type": "int32", "versions": "12+", "default": "-1",
             "about": "The latest known leader epoch"}
         ]},
+        { "name": "SnapshotId", "type": "SnapshotId",
+          "versions": "12+", "taggedVersions": "12+", "tag": 2,
+          "about": "In the case of fetching an offset less than the LogStartOffset, this is the end offset and epoch that should be used in the FetchSnapshot request.",
+          "fields": [
+            { "name": "EndOffset", "type": "int64", "versions": "0+", "default": "-1" },
+            { "name": "Epoch", "type": "int32", "versions": "0+", "default": "-1" }
+        ]},
         { "name": "AbortedTransactions", "type": "[]AbortedTransaction", "versions": "4+", "nullableVersions": "4+", "ignorable": true,
           "about": "The aborted transactions.",  "fields": [
           { "name": "ProducerId", "type": "int64", "versions": "4+", "entityType": "producerId",

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

@@ -0,0 +1,50 @@
+// 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.
+
+{
+  "apiKey": 59,
+  "type": "request",
+  "name": "FetchSnapshotRequest",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "ClusterId", "type": "string", "versions": "0+", "nullableVersions": "0+", "default": "null", "taggedVersions": "0+", "tag": 0,
+      "about": "The clusterId if known, this is used to validate metadata fetches prior to broker registration" },
+    { "name": "ReplicaId", "type": "int32", "versions": "0+", "default": "-1",
+      "about": "The broker ID of the follower" },
+    { "name": "MaxBytes", "type": "int32", "versions": "0+", "default": "0x7fffffff",
+      "about": "The maximum bytes to fetch from all of the snapshots" },
+    { "name": "Topics", "type": "[]TopicSnapshot", "versions": "0+",
+      "about": "The topics to fetch", "fields": [
+      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+        "about": "The name of the topic to fetch" },
+      { "name": "Partitions", "type": "[]PartitionSnapshot", "versions": "0+",
+        "about": "The partitions to fetch", "fields": [
+        { "name": "Partition", "type": "int32", "versions": "0+",
+          "about": "The partition index" },
+        { "name": "CurrentLeaderEpoch", "type": "int32", "versions": "0+",
+          "about": "The current leader epoch of the partition, -1 for unknown leader epoch" },
+        { "name": "SnapshotId", "type": "SnapshotId", "versions": "0+",
+          "about": "The snapshot endOffset and epoch to fetch",
+          "fields": [
+          { "name": "EndOffset", "type": "int64", "versions": "0+" },
+          { "name": "Epoch", "type": "int32", "versions": "0+" }
+        ]},
+        { "name": "Position", "type": "int64", "versions": "0+",
+          "about": "The byte position within the snapshot to start fetching from" }
+      ]}
+    ]}
+  ]
+}