KAFKA-10427: Fetch snapshot

clients/src/main/java/org/apache/kafka/common/errors/PositionOutOfRangeException.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 PositionOutOfRangeException extends ApiException {
+
+    private static final long serialVersionUID = 1;
+
+    public PositionOutOfRangeException(String s) {
+        super(s);
+    }
+
+    public PositionOutOfRangeException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+}

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

@@ -96,7 +96,8 @@ public enum ApiKeys {
     DESCRIBE_QUORUM(ApiMessageType.DESCRIBE_QUORUM, true, RecordBatch.MAGIC_VALUE_V0, false, false),
     ALTER_ISR(ApiMessageType.ALTER_ISR, true),
     UPDATE_FEATURES(ApiMessageType.UPDATE_FEATURES, false, true),
-    ENVELOPE(ApiMessageType.ENVELOPE, true, RecordBatch.MAGIC_VALUE_V0, false, false);
+    ENVELOPE(ApiMessageType.ENVELOPE, true, RecordBatch.MAGIC_VALUE_V0, false, false),
+    FETCH_SNAPSHOT(ApiMessageType.FETCH_SNAPSHOT, false, RecordBatch.MAGIC_VALUE_V0, false, false);
 
     // The generator ensures every `ApiMessageType` has a unique id
     private static final Map<Integer, ApiKeys> ID_TO_TYPE = Arrays.stream(ApiKeys.values())

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

@@ -34,10 +34,9 @@
 import org.apache.kafka.common.errors.DuplicateSequenceException;
 import org.apache.kafka.common.errors.ElectionNotNeededException;
 import org.apache.kafka.common.errors.EligibleLeadersNotAvailableException;
-import org.apache.kafka.common.errors.FencedInstanceIdException;
 import org.apache.kafka.common.errors.FeatureUpdateFailedException;
+import org.apache.kafka.common.errors.FencedInstanceIdException;
 import org.apache.kafka.common.errors.FencedLeaderEpochException;
-import org.apache.kafka.common.errors.InvalidUpdateVersionException;
 import org.apache.kafka.common.errors.FetchSessionIdNotFoundException;
 import org.apache.kafka.common.errors.GroupAuthorizationException;
 import org.apache.kafka.common.errors.GroupIdNotFoundException;
@@ -56,6 +55,7 @@
 import org.apache.kafka.common.errors.InvalidPartitionsException;
 import org.apache.kafka.common.errors.InvalidPidMappingException;
 import org.apache.kafka.common.errors.InvalidPrincipalTypeException;
+import org.apache.kafka.common.errors.InvalidProducerEpochException;
 import org.apache.kafka.common.errors.InvalidReplicaAssignmentException;
 import org.apache.kafka.common.errors.InvalidReplicationFactorException;
 import org.apache.kafka.common.errors.InvalidRequestException;
@@ -65,6 +65,7 @@
 import org.apache.kafka.common.errors.InvalidTopicException;
 import org.apache.kafka.common.errors.InvalidTxnStateException;
 import org.apache.kafka.common.errors.InvalidTxnTimeoutException;
+import org.apache.kafka.common.errors.InvalidUpdateVersionException;
 import org.apache.kafka.common.errors.KafkaStorageException;
 import org.apache.kafka.common.errors.LeaderNotAvailableException;
 import org.apache.kafka.common.errors.ListenerNotFoundException;
@@ -83,6 +84,7 @@
 import org.apache.kafka.common.errors.OperationNotAttemptedException;
 import org.apache.kafka.common.errors.OutOfOrderSequenceException;
 import org.apache.kafka.common.errors.PolicyViolationException;
+import org.apache.kafka.common.errors.PositionOutOfRangeException;
 import org.apache.kafka.common.errors.PreferredLeaderNotAvailableException;
 import org.apache.kafka.common.errors.PrincipalDeserializationException;
 import org.apache.kafka.common.errors.ProducerFencedException;
@@ -95,6 +97,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;
@@ -115,7 +118,6 @@
 import org.apache.kafka.common.errors.UnsupportedForMessageFormatException;
 import org.apache.kafka.common.errors.UnsupportedSaslMechanismException;
 import org.apache.kafka.common.errors.UnsupportedVersionException;
-import org.apache.kafka.common.errors.InvalidProducerEpochException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -343,7 +345,12 @@ 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),
+    POSITION_OUT_OF_RANGE(
+        99,
+        "Requested position is not greater than or equal to zero, and less than the size of the snapshot.",
+        PositionOutOfRangeException::new);
 
     private static final Logger log = LoggerFactory.getLogger(Errors.class);
 

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

@@ -272,6 +272,8 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
                 return UpdateFeaturesRequest.parse(buffer, apiVersion);
             case ENVELOPE:
                 return EnvelopeRequest.parse(buffer, apiVersion);
+            case FETCH_SNAPSHOT:
+                return FetchSnapshotRequest.parse(buffer, apiVersion);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
                         "code should be updated to do so.", apiKey));

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

@@ -229,6 +229,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, ByteBuffer response
                 return UpdateFeaturesResponse.parse(responseBuffer, version);
             case ENVELOPE:
                 return EnvelopeResponse.parse(responseBuffer, version);
+            case FETCH_SNAPSHOT:
+                return FetchSnapshotResponse.parse(responseBuffer, version);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                         "code should be updated to do so.", apiKey));

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

@@ -0,0 +1,122 @@
+/*
+ * 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.nio.ByteBuffer;
+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.ApiMessage;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+
+final public class FetchSnapshotRequest extends AbstractRequest {
+    public final FetchSnapshotRequestData data;
+
+    public FetchSnapshotRequest(FetchSnapshotRequestData data, short version) {
+        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
+    public ApiMessage 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();
+    }
+
+    public static FetchSnapshotRequest parse(ByteBuffer buffer, short version) {
+        return new FetchSnapshotRequest(new FetchSnapshotRequestData(new ByteBufferAccessor(buffer), version), version);
+    }
+
+    public static class Builder extends AbstractRequest.Builder<FetchSnapshotRequest> {
+        private final FetchSnapshotRequestData data;
+
+        public Builder(FetchSnapshotRequestData  data) {
+            super(ApiKeys.FETCH_SNAPSHOT);
+            this.data = data;
+        }
+
+        @Override
+        public FetchSnapshotRequest build(short version) {
+            return new FetchSnapshotRequest(data, version);
+        }
+
+        @Override
+        public String toString() {
+            return data.toString();
+        }
+    }
+}