You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/11/03 18:59:56 UTC
[GitHub] [kafka] jsancio opened a new pull request #9553: KAFKA-10427: Fetch snapshot
jsancio opened a new pull request #9553:
URL: https://github.com/apache/kafka/pull/9553
Implements the code necessary for the leader to response to fetch snapshot requests and for the follower to fetch snapshots. It depends on PR https://github.com/apache/kafka/pull/9512 and that PR should be reviewed first. Included are the following changes.
Leader Changes:
1. Raft leader response to FetchSnapshot request by reading the local snapshot and sending the requested bytes in the response. This implementation currently copies the bytes to memory. This will be fixed in a future PR.
Follower Changes:
1. Raft followers will start fetching snapshot if the leader sends a Fetch response that includes a SnapshotId.
2. Raft followers send FetchSnapshot requests if there is a pending download. The same timer is used for both Fetch and FetchSnapshot requests.
3. Raft follower handle FetchSnapshot responses by comping the bytes to the pending SnapshotWriter. This implementation doesn't fix the replicated log after the snapshot has been downloaded. This will be implemented in a future PR.
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547437816
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
+ ByteBuffer buffer = ByteBuffer.allocate(Math.min(data.maxBytes(), maxSnapshotSize));
+ snapshot.read(buffer, partitionSnapshot.position());
Review comment:
Good catch. In the KIP we added `POSITION_OUT_OF_RANGE` which I am not using any where in this PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547391686
##########
File path: raft/src/main/java/org/apache/kafka/raft/LeaderState.java
##########
@@ -287,4 +287,7 @@ public String name() {
return "Leader";
}
+ @Override
+ public void close() {}
Review comment:
Maybe we could add a default no-op implementation to EpochState?
##########
File path: 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) {
Review comment:
nit: maybe `withTopLevelError`?
##########
File path: clients/src/main/resources/common/message/FetchSnapshotResponse.json
##########
@@ -0,0 +1,59 @@
+// 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": "response",
+ "name": "FetchSnapshotResponse",
+ "validVersions": "0",
+ "flexibleVersions": "0+",
+ "fields": [
+ { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
+ "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+", "ignorable": false,
+ "about": "The top level response error code." },
+ { "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": "Index", "type": "int32", "versions": "0+",
+ "about": "The partition index." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+",
+ "about": "The error code, or 0 if there was no fetch error." },
+ { "name": "SnapshotId", "type": "SnapshotId", "versions": "0+",
+ "about": "The snapshot endOffset and epoch fetched",
+ "fields": [
+ { "name": "EndOffset", "type": "int64", "versions": "0+" },
+ { "name": "Epoch", "type": "int32", "versions": "0+" }
+ ]},
+ { "name": "CurrentLeader", "type": "LeaderIdAndEpoch",
+ "versions": "0+", "taggedVersions": "0+", "tag": 0, "fields": [
+ { "name": "LeaderId", "type": "int32", "versions": "0+",
+ "about": "The ID of the current leader or -1 if the leader is unknown."},
+ { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+ "about": "The latest known leader epoch"}
+ ]},
+ { "name": "Size", "type": "int64", "versions": "0+",
+ "about": "The total size of the snapshot." },
+ { "name": "Position", "type": "int64", "versions": "0+",
+ "about": "The starting byte position within the snapshot included in the Bytes field." },
+ { "name": "Bytes", "type": "bytes", "versions": "0+", "zeroCopy": true,
Review comment:
Can you remind me if we are planning to change the type to "records"? I don't think we will get the benefit of `sendfile` unless we do so.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
+ ByteBuffer buffer = ByteBuffer.allocate(Math.min(data.maxBytes(), maxSnapshotSize));
+ snapshot.read(buffer, partitionSnapshot.position());
+ buffer.flip();
+
+ long snapshotSize = snapshot.sizeInBytes();
+
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ addQuorumLeader(responsePartitionSnapshot)
+ .snapshotId()
+ .setEndOffset(snapshotId.offset)
+ .setEpoch(snapshotId.epoch);
+
+ return responsePartitionSnapshot
+ .setSize(snapshotSize)
+ .setPosition(partitionSnapshot.position())
+ .setBytes(buffer);
+ }
+ );
+ }
+ }
+
+ private boolean handleFetchSnapshotResponse(
+ RaftResponse.Inbound responseMetadata,
+ long currentTimeMs
+ ) throws IOException {
+ FetchSnapshotResponseData data = (FetchSnapshotResponseData) responseMetadata.data;
+ Errors topLevelError = Errors.forCode(data.errorCode());
+ if (topLevelError != Errors.NONE) {
+ // TODO: check what values this expression returns
Review comment:
Address TODO?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -941,6 +949,8 @@ private FetchResponseData tryCompleteFetchRequest(
) {
Optional<Errors> errorOpt = validateLeaderOnlyRequest(request.currentLeaderEpoch());
if (errorOpt.isPresent()) {
+ // TODO: The replica should return what information it knows about the current epoch and
Review comment:
Isn't this already done by `buildFetchResponse`?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
Review comment:
nit: I guess you could use `leaderValidation.ifPresent`
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1037,6 +1047,35 @@ private boolean handleFetchResponse(
logger.info("Truncated to offset {} from Fetch response from leader {}",
truncationOffset, quorum.leaderIdOrNil());
});
+ } else if (partitionResponse.snapshotId().epoch() >= 0 ||
+ partitionResponse.snapshotId().endOffset() >= 0) {
+ // The leader is asking us to fetch a snapshot
+
+ if (partitionResponse.snapshotId().epoch() < 0) {
+ throw new KafkaException(
Review comment:
Hmm.. The leader has sent a bad response. I think logging an error and retrying might be better than crashing the followers.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
+ ByteBuffer buffer = ByteBuffer.allocate(Math.min(data.maxBytes(), maxSnapshotSize));
+ snapshot.read(buffer, partitionSnapshot.position());
Review comment:
Is it worth validating that `partitionSnapshot.position()` is non-negative?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1629,16 +1872,29 @@ private long pollFollowerAsVoter(FollowerState state, long currentTimeMs) throws
transitionToCandidate(currentTimeMs);
return 0L;
} else {
- long backoffMs = maybeSendRequest(
- currentTimeMs,
- state.leaderId(),
- this::buildFetchRequest
- );
+ long backoffMs;
+ if (state.fetchingSnapshot().isPresent()) {
Review comment:
Can we move this logic to `pollFollower` somehow?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
Review comment:
Just checking my understanding. This patch adds the logic to respond to the snapshot id from a fetch response and to handle send/handle snapshots when needed. However, since it does not contain the logic to set the snapshot id in the fetch request handler, none of this logic will get exercised by the simulation test. Is that right?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
+ ByteBuffer buffer = ByteBuffer.allocate(Math.min(data.maxBytes(), maxSnapshotSize));
+ snapshot.read(buffer, partitionSnapshot.position());
+ buffer.flip();
+
+ long snapshotSize = snapshot.sizeInBytes();
+
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ addQuorumLeader(responsePartitionSnapshot)
+ .snapshotId()
+ .setEndOffset(snapshotId.offset)
+ .setEpoch(snapshotId.epoch);
+
+ return responsePartitionSnapshot
+ .setSize(snapshotSize)
+ .setPosition(partitionSnapshot.position())
+ .setBytes(buffer);
+ }
+ );
+ }
+ }
+
+ private boolean handleFetchSnapshotResponse(
+ RaftResponse.Inbound responseMetadata,
+ long currentTimeMs
+ ) throws IOException {
+ FetchSnapshotResponseData data = (FetchSnapshotResponseData) responseMetadata.data;
+ Errors topLevelError = Errors.forCode(data.errorCode());
+ if (topLevelError != Errors.NONE) {
+ // TODO: check what values this expression returns
+ return handleTopLevelError(topLevelError, responseMetadata);
+ }
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return false;
+ }
+
+ Optional<FetchSnapshotResponseData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotResponse
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ return false;
+ }
+
+ FetchSnapshotResponseData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+
+ FetchSnapshotResponseData.LeaderIdAndEpoch currentLeaderIdAndEpoch = partitionSnapshot.currentLeader();
+ OptionalInt responseLeaderId = optionalLeaderId(currentLeaderIdAndEpoch.leaderId());
+ int responseEpoch = currentLeaderIdAndEpoch.leaderEpoch();
+ Errors error = Errors.forCode(partitionSnapshot.errorCode());
+
+ Optional<Boolean> handled = maybeHandleCommonResponse(
+ error, responseLeaderId, responseEpoch, currentTimeMs);
+ if (handled.isPresent()) {
+ // TODO: check what values this expression returns
+ return handled.get();
+ }
+
+ FollowerState state = quorum.followerStateOrThrow();
+
+ if (Errors.forCode(partitionSnapshot.errorCode()) == Errors.SNAPSHOT_NOT_FOUND ||
+ partitionSnapshot.snapshotId().endOffset() < 0 ||
+ partitionSnapshot.snapshotId().epoch() < 0) {
+
+ /* The leader deleted the snapshot before the follower could download it. Start over by
Review comment:
A log message would probably be helpful. It's probably worth doing one full pass over the logic here to see where we could add extra logging.
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -0,0 +1,765 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.memory.MemoryPool;
+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.protocol.Errors;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.requests.FetchSnapshotRequest;
+import org.apache.kafka.common.requests.FetchSnapshotResponse;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.raft.internals.StringSerde;
+import org.apache.kafka.snapshot.RawSnapshotReader;
+import org.apache.kafka.snapshot.RawSnapshotWriter;
+import org.apache.kafka.snapshot.SnapshotWriter;
+import org.apache.kafka.snapshot.SnapshotWriterTest;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Disabled;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+final public class KafkaRaftClientSnapshotTest {
+ @Test
+ public void testMissingFetchSnapshotRequest() throws Exception {
+ int localId = 0;
+ int epoch = 2;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ new OffsetAndEpoch(0, 0),
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context.assertSentFetchSnapshotResponse(context.metadataPartition).get();
+ assertEquals(Errors.SNAPSHOT_NOT_FOUND, Errors.forCode(response.errorCode()));
+ }
+
+ @Test
+ public void testUnknownFetchSnapshotRequest() throws Exception {
+ int localId = 0;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+ int epoch = 2;
+ TopicPartition topicPartition = new TopicPartition("unknown", 0);
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ topicPartition,
+ epoch,
+ new OffsetAndEpoch(0, 0),
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context.assertSentFetchSnapshotResponse(topicPartition).get();
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.forCode(response.errorCode()));
+ }
+
+ @Test
+ public void testFetchSnapshotRequestAsLeader() throws Exception {
+ int localId = 0;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+ int epoch = 2;
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(0, 0);
+ List<String> records = Arrays.asList("foo", "bar");
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ try (SnapshotWriter<String> snapshot = context.client.createSnapshot(snapshotId)) {
+ snapshot.append(records);
+ snapshot.freeze();
+ }
+
+ try (RawSnapshotReader snapshot = context.log.readSnapshot(snapshotId).get()) {
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context
+ .assertSentFetchSnapshotResponse(context.metadataPartition)
+ .get();
+
+ assertEquals(Errors.NONE, Errors.forCode(response.errorCode()));
+ assertEquals(snapshot.sizeInBytes(), response.size());
+ assertEquals(0, response.position());
+ assertEquals(snapshot.sizeInBytes(), response.bytes().remaining());
+
+ ByteBuffer buffer = ByteBuffer.allocate(Math.toIntExact(snapshot.sizeInBytes()));
+ snapshot.read(buffer, 0);
+ buffer.flip();
+
+ assertEquals(buffer.slice(), response.bytes());
+ }
+ }
+
+ @Test
+ public void testPartialFetchSnapshotRequestAsLeader() throws Exception {
+ int localId = 0;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+ int epoch = 2;
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(0, 0);
+ List<String> records = Arrays.asList("foo", "bar");
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ try (SnapshotWriter<String> snapshot = context.client.createSnapshot(snapshotId)) {
+ snapshot.append(records);
+ snapshot.freeze();
+ }
+
+ try (RawSnapshotReader snapshot = context.log.readSnapshot(snapshotId).get()) {
+ // Fetch half of the snapshot
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Math.toIntExact(snapshot.sizeInBytes() / 2),
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context
+ .assertSentFetchSnapshotResponse(context.metadataPartition)
+ .get();
+
+ assertEquals(Errors.NONE, Errors.forCode(response.errorCode()));
+ assertEquals(snapshot.sizeInBytes(), response.size());
+ assertEquals(0, response.position());
+ assertEquals(snapshot.sizeInBytes() / 2, response.bytes().remaining());
+
+ ByteBuffer snapshotBuffer = ByteBuffer.allocate(Math.toIntExact(snapshot.sizeInBytes()));
+ snapshot.read(snapshotBuffer, 0);
+ snapshotBuffer.flip();
+
+ ByteBuffer responseBuffer = ByteBuffer.allocate(Math.toIntExact(snapshot.sizeInBytes()));
+ responseBuffer.put(response.bytes());
+
+ ByteBuffer expectedBytes = snapshotBuffer.duplicate();
+ expectedBytes.limit(Math.toIntExact(snapshot.sizeInBytes() / 2));
+
+ assertEquals(expectedBytes, responseBuffer.duplicate().flip());
+
+ // Fetch the remainder of the snapshot
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ responseBuffer.position()
+ )
+ );
+
+ context.client.poll();
+
+ response = context.assertSentFetchSnapshotResponse(context.metadataPartition).get();
+ assertEquals(Errors.NONE, Errors.forCode(response.errorCode()));
+ assertEquals(snapshot.sizeInBytes(), response.size());
+ assertEquals(responseBuffer.position(), response.position());
+ assertEquals(snapshot.sizeInBytes() - (snapshot.sizeInBytes() / 2), response.bytes().remaining());
+
+ responseBuffer.put(response.bytes());
+ assertEquals(snapshotBuffer, responseBuffer.flip());
+ }
+ }
+
+ @Test
+ public void testFetchSnapshotRequestAsFollower() throws IOException {
+ int localId = 0;
+ int leaderId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, leaderId);
+ int epoch = 2;
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(0, 0);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withElectedLeader(epoch, leaderId)
+ .build();
+
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context.assertSentFetchSnapshotResponse(context.metadataPartition).get();
+ assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, Errors.forCode(response.errorCode()));
+ assertEquals(epoch, response.currentLeader().leaderEpoch());
+ assertEquals(leaderId, response.currentLeader().leaderId());
+ }
+
+ @Disabled
+ @Test
+ public void testFetchSnapshotRequestWithOlderEpoch() throws IOException {
+ assertTrue(false);
Review comment:
I guess you're still planning to implement these?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547475978
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1037,6 +1047,35 @@ private boolean handleFetchResponse(
logger.info("Truncated to offset {} from Fetch response from leader {}",
truncationOffset, quorum.leaderIdOrNil());
});
+ } else if (partitionResponse.snapshotId().epoch() >= 0 ||
+ partitionResponse.snapshotId().endOffset() >= 0) {
+ // The leader is asking us to fetch a snapshot
+
+ if (partitionResponse.snapshotId().epoch() < 0) {
+ throw new KafkaException(
Review comment:
Yeah. I am not sure. If these exception are thrown then it means that the epoch is "valid" but the end offset is not "valid" or vice versa. I think this could happen because of either a buggy remote replica or corrupted data in the remote replica.
I guess you can argue that it is safe to keep retrying.
In the case that the `Fetch` never succeed do we want to transition to the candidate the state? If so, let me add a test that confirms that.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547548831
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1037,6 +1047,35 @@ private boolean handleFetchResponse(
logger.info("Truncated to offset {} from Fetch response from leader {}",
truncationOffset, quorum.leaderIdOrNil());
});
+ } else if (partitionResponse.snapshotId().epoch() >= 0 ||
+ partitionResponse.snapshotId().endOffset() >= 0) {
+ // The leader is asking us to fetch a snapshot
+
+ if (partitionResponse.snapshotId().epoch() < 0) {
+ throw new KafkaException(
Review comment:
Hmm.. I would rather not handle this case specially by transitioning to a candidate. It's not a crazy idea, but I think we should attempt that in a separate PR and consider errors more holistically.
I would suggest that we log an error saying that the remote replica seemed to return an invalid response and just keep fetching. Then a user can see the log message and restart the remote replica.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547412746
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
Review comment:
Correct. I am handling this case in https://issues.apache.org/jira/browse/KAFKA-10761. I think that after implementing that issue we should have an end-to-end working Raft Client with snapshot support that we test in the simulations.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
hachikuji commented on pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#issuecomment-751924971
I manually reran tests since it has been a week since the last update. Here is a link to the results since they have not been automatically linked: https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-9553/14/.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547485023
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -0,0 +1,765 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.memory.MemoryPool;
+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.protocol.Errors;
+import org.apache.kafka.common.record.CompressionType;
+import org.apache.kafka.common.requests.FetchSnapshotRequest;
+import org.apache.kafka.common.requests.FetchSnapshotResponse;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.raft.internals.StringSerde;
+import org.apache.kafka.snapshot.RawSnapshotReader;
+import org.apache.kafka.snapshot.RawSnapshotWriter;
+import org.apache.kafka.snapshot.SnapshotWriter;
+import org.apache.kafka.snapshot.SnapshotWriterTest;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Disabled;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+final public class KafkaRaftClientSnapshotTest {
+ @Test
+ public void testMissingFetchSnapshotRequest() throws Exception {
+ int localId = 0;
+ int epoch = 2;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ new OffsetAndEpoch(0, 0),
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context.assertSentFetchSnapshotResponse(context.metadataPartition).get();
+ assertEquals(Errors.SNAPSHOT_NOT_FOUND, Errors.forCode(response.errorCode()));
+ }
+
+ @Test
+ public void testUnknownFetchSnapshotRequest() throws Exception {
+ int localId = 0;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+ int epoch = 2;
+ TopicPartition topicPartition = new TopicPartition("unknown", 0);
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ topicPartition,
+ epoch,
+ new OffsetAndEpoch(0, 0),
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context.assertSentFetchSnapshotResponse(topicPartition).get();
+ assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION, Errors.forCode(response.errorCode()));
+ }
+
+ @Test
+ public void testFetchSnapshotRequestAsLeader() throws Exception {
+ int localId = 0;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+ int epoch = 2;
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(0, 0);
+ List<String> records = Arrays.asList("foo", "bar");
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ try (SnapshotWriter<String> snapshot = context.client.createSnapshot(snapshotId)) {
+ snapshot.append(records);
+ snapshot.freeze();
+ }
+
+ try (RawSnapshotReader snapshot = context.log.readSnapshot(snapshotId).get()) {
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context
+ .assertSentFetchSnapshotResponse(context.metadataPartition)
+ .get();
+
+ assertEquals(Errors.NONE, Errors.forCode(response.errorCode()));
+ assertEquals(snapshot.sizeInBytes(), response.size());
+ assertEquals(0, response.position());
+ assertEquals(snapshot.sizeInBytes(), response.bytes().remaining());
+
+ ByteBuffer buffer = ByteBuffer.allocate(Math.toIntExact(snapshot.sizeInBytes()));
+ snapshot.read(buffer, 0);
+ buffer.flip();
+
+ assertEquals(buffer.slice(), response.bytes());
+ }
+ }
+
+ @Test
+ public void testPartialFetchSnapshotRequestAsLeader() throws Exception {
+ int localId = 0;
+ Set<Integer> voters = Utils.mkSet(localId, localId + 1);
+ int epoch = 2;
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(0, 0);
+ List<String> records = Arrays.asList("foo", "bar");
+
+ RaftClientTestContext context = RaftClientTestContext.initializeAsLeader(localId, voters, epoch);
+
+ try (SnapshotWriter<String> snapshot = context.client.createSnapshot(snapshotId)) {
+ snapshot.append(records);
+ snapshot.freeze();
+ }
+
+ try (RawSnapshotReader snapshot = context.log.readSnapshot(snapshotId).get()) {
+ // Fetch half of the snapshot
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Math.toIntExact(snapshot.sizeInBytes() / 2),
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context
+ .assertSentFetchSnapshotResponse(context.metadataPartition)
+ .get();
+
+ assertEquals(Errors.NONE, Errors.forCode(response.errorCode()));
+ assertEquals(snapshot.sizeInBytes(), response.size());
+ assertEquals(0, response.position());
+ assertEquals(snapshot.sizeInBytes() / 2, response.bytes().remaining());
+
+ ByteBuffer snapshotBuffer = ByteBuffer.allocate(Math.toIntExact(snapshot.sizeInBytes()));
+ snapshot.read(snapshotBuffer, 0);
+ snapshotBuffer.flip();
+
+ ByteBuffer responseBuffer = ByteBuffer.allocate(Math.toIntExact(snapshot.sizeInBytes()));
+ responseBuffer.put(response.bytes());
+
+ ByteBuffer expectedBytes = snapshotBuffer.duplicate();
+ expectedBytes.limit(Math.toIntExact(snapshot.sizeInBytes() / 2));
+
+ assertEquals(expectedBytes, responseBuffer.duplicate().flip());
+
+ // Fetch the remainder of the snapshot
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ responseBuffer.position()
+ )
+ );
+
+ context.client.poll();
+
+ response = context.assertSentFetchSnapshotResponse(context.metadataPartition).get();
+ assertEquals(Errors.NONE, Errors.forCode(response.errorCode()));
+ assertEquals(snapshot.sizeInBytes(), response.size());
+ assertEquals(responseBuffer.position(), response.position());
+ assertEquals(snapshot.sizeInBytes() - (snapshot.sizeInBytes() / 2), response.bytes().remaining());
+
+ responseBuffer.put(response.bytes());
+ assertEquals(snapshotBuffer, responseBuffer.flip());
+ }
+ }
+
+ @Test
+ public void testFetchSnapshotRequestAsFollower() throws IOException {
+ int localId = 0;
+ int leaderId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, leaderId);
+ int epoch = 2;
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(0, 0);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withElectedLeader(epoch, leaderId)
+ .build();
+
+ context.deliverRequest(
+ fetchSnapshotRequest(
+ context.metadataPartition,
+ epoch,
+ snapshotId,
+ Integer.MAX_VALUE,
+ 0
+ )
+ );
+
+ context.client.poll();
+
+ FetchSnapshotResponseData.PartitionSnapshot response = context.assertSentFetchSnapshotResponse(context.metadataPartition).get();
+ assertEquals(Errors.NOT_LEADER_OR_FOLLOWER, Errors.forCode(response.errorCode()));
+ assertEquals(epoch, response.currentLeader().leaderEpoch());
+ assertEquals(leaderId, response.currentLeader().leaderId());
+ }
+
+ @Disabled
+ @Test
+ public void testFetchSnapshotRequestWithOlderEpoch() throws IOException {
+ assertTrue(false);
Review comment:
Yes. Doing that now...
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547481211
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
+ ByteBuffer buffer = ByteBuffer.allocate(Math.min(data.maxBytes(), maxSnapshotSize));
+ snapshot.read(buffer, partitionSnapshot.position());
+ buffer.flip();
+
+ long snapshotSize = snapshot.sizeInBytes();
+
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ addQuorumLeader(responsePartitionSnapshot)
+ .snapshotId()
+ .setEndOffset(snapshotId.offset)
+ .setEpoch(snapshotId.epoch);
+
+ return responsePartitionSnapshot
+ .setSize(snapshotSize)
+ .setPosition(partitionSnapshot.position())
+ .setBytes(buffer);
+ }
+ );
+ }
+ }
+
+ private boolean handleFetchSnapshotResponse(
+ RaftResponse.Inbound responseMetadata,
+ long currentTimeMs
+ ) throws IOException {
+ FetchSnapshotResponseData data = (FetchSnapshotResponseData) responseMetadata.data;
+ Errors topLevelError = Errors.forCode(data.errorCode());
+ if (topLevelError != Errors.NONE) {
+ // TODO: check what values this expression returns
Review comment:
Done. Excuse the noise.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547472266
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -941,6 +949,8 @@ private FetchResponseData tryCompleteFetchRequest(
) {
Optional<Errors> errorOpt = validateLeaderOnlyRequest(request.currentLeaderEpoch());
if (errorOpt.isPresent()) {
+ // TODO: The replica should return what information it knows about the current epoch and
Review comment:
Yes. This was a note to me to confirm this as it wasn't clear to me when I first read the code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r538060397
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
+ ByteBuffer buffer = ByteBuffer.allocate(Math.min(data.maxBytes(), maxSnapshotSize));
+ snapshot.read(buffer, partitionSnapshot.position());
+ buffer.flip();
+
+ long snapshotSize = snapshot.sizeInBytes();
+
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ addQuorumLeader(responsePartitionSnapshot)
+ .snapshotId()
+ .setEndOffset(snapshotId.offset)
+ .setEpoch(snapshotId.epoch);
+
+ return responsePartitionSnapshot
+ .setSize(snapshotSize)
+ .setPosition(partitionSnapshot.position())
+ .setBytes(buffer);
+ }
+ );
+ }
+ }
+
+ private boolean handleFetchSnapshotResponse(
+ RaftResponse.Inbound responseMetadata,
+ long currentTimeMs
+ ) throws IOException {
+ FetchSnapshotResponseData data = (FetchSnapshotResponseData) responseMetadata.data;
+ Errors topLevelError = Errors.forCode(data.errorCode());
+ if (topLevelError != Errors.NONE) {
+ // TODO: check what values this expression returns
+ return handleTopLevelError(topLevelError, responseMetadata);
+ }
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return false;
+ }
+
+ Optional<FetchSnapshotResponseData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotResponse
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ return false;
+ }
+
+ FetchSnapshotResponseData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+
+ FetchSnapshotResponseData.LeaderIdAndEpoch currentLeaderIdAndEpoch = partitionSnapshot.currentLeader();
+ OptionalInt responseLeaderId = optionalLeaderId(currentLeaderIdAndEpoch.leaderId());
+ int responseEpoch = currentLeaderIdAndEpoch.leaderEpoch();
+ Errors error = Errors.forCode(partitionSnapshot.errorCode());
+
+ Optional<Boolean> handled = maybeHandleCommonResponse(
+ error, responseLeaderId, responseEpoch, currentTimeMs);
+ if (handled.isPresent()) {
+ // TODO: check what values this expression returns
+ return handled.get();
+ }
+
+ FollowerState state = quorum.followerStateOrThrow();
+
+ if (Errors.forCode(partitionSnapshot.errorCode()) == Errors.SNAPSHOT_NOT_FOUND ||
+ partitionSnapshot.snapshotId().endOffset() < 0 ||
+ partitionSnapshot.snapshotId().epoch() < 0) {
+
+ /* The leader deleted the snapshot before the follower could download it. Start over by
+ * reseting the fetching snapshot state and sending another fetch request.
+ */
+ state.setFetchingSnapshot(Optional.empty());
+ state.resetFetchTimeout(currentTimeMs);
+ return true;
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+
+ RawSnapshotWriter snapshot;
+ if (state.fetchingSnapshot().isPresent()) {
+ snapshot = state.fetchingSnapshot().get();
+ } else {
+ throw new IllegalStateException("Received unexpected fetch snapshot response: " + partitionSnapshot);
+ }
+
+ if (!snapshot.snapshotId().equals(snapshotId)) {
+ throw new IllegalStateException(String.format("Received fetch snapshot response with an invalid id. Expected %s; Received %s", snapshot.snapshotId(), snapshotId));
+ }
+ if (snapshot.sizeInBytes() != partitionSnapshot.position()) {
+ throw new IllegalStateException(String.format("Received fetch snapshot response with an invalid position. Expected %s; Received %s", snapshot.sizeInBytes(), partitionSnapshot.position()));
+ }
+
+ snapshot.append(partitionSnapshot.bytes());
+
+ if (snapshot.sizeInBytes() == partitionSnapshot.size()) {
+ // Finished fetching the snapshot.
+ snapshot.freeze();
+ state.setFetchingSnapshot(Optional.empty());
Review comment:
The follower needs to update the log start offset and log end offset after it has successfully fetched a snapshot. I want to implement this part in this JIRA: https://issues.apache.org/jira/browse/KAFKA-10820
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r549053981
##########
File path: clients/src/main/resources/common/message/FetchSnapshotResponse.json
##########
@@ -0,0 +1,59 @@
+// 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": "response",
+ "name": "FetchSnapshotResponse",
+ "validVersions": "0",
+ "flexibleVersions": "0+",
+ "fields": [
+ { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
+ "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+", "ignorable": false,
+ "about": "The top level response error code." },
+ { "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": "Index", "type": "int32", "versions": "0+",
+ "about": "The partition index." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+",
+ "about": "The error code, or 0 if there was no fetch error." },
+ { "name": "SnapshotId", "type": "SnapshotId", "versions": "0+",
+ "about": "The snapshot endOffset and epoch fetched",
+ "fields": [
+ { "name": "EndOffset", "type": "int64", "versions": "0+" },
+ { "name": "Epoch", "type": "int32", "versions": "0+" }
+ ]},
+ { "name": "CurrentLeader", "type": "LeaderIdAndEpoch",
+ "versions": "0+", "taggedVersions": "0+", "tag": 0, "fields": [
+ { "name": "LeaderId", "type": "int32", "versions": "0+",
+ "about": "The ID of the current leader or -1 if the leader is unknown."},
+ { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+ "about": "The latest known leader epoch"}
+ ]},
+ { "name": "Size", "type": "int64", "versions": "0+",
+ "about": "The total size of the snapshot." },
+ { "name": "Position", "type": "int64", "versions": "0+",
+ "about": "The starting byte position within the snapshot included in the Bytes field." },
+ { "name": "Bytes", "type": "bytes", "versions": "0+", "zeroCopy": true,
Review comment:
@dengziming, Thanks for the comment. The `FetchSnapshotRequest` differs from `FetchRequest` in that `Position` is a byte offset. Because of this we don't need an `Offset` to `Byte` index.
If we implement option 1 in my comment above, we can still use `Records` by making sure that:
1. The `leaders` send at least one entire `RecordBatch` in the `FetchSnapshotResponse`.
2. The `followers` only write complete `RecordBatch`es by not writing bytes sent by the leader that are not complete `RecordBatch`.
This guarantees that the `Position` sent in `FetchSnapshotRequest` is `RecordBatch` aligned and it always makes progress because the leader sends at least one complete `RecordBatch`.
Having said that, I think we should try and implement option 2. Unfortunately, the Jira https://issues.apache.org/jira/browse/KAFKA-10694 doesn't document option 2.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r537993156
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1037,6 +1047,14 @@ private boolean handleFetchResponse(
logger.info("Truncated to offset {} from Fetch response from leader {}",
truncationOffset, quorum.leaderIdOrNil());
});
+ } else if (partitionResponse.snapshotId().epoch() >= 0 && partitionResponse.snapshotId().endOffset() >= 0) {
Review comment:
Fix this logic to be more conservative. It should be a fatal error if one is negative and the other is positive.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547574619
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1055,14 +1055,14 @@ private boolean handleFetchResponse(
partitionResponse.snapshotId().endOffset(),
partitionResponse.snapshotId().epoch()
);
- return false;
+ return true;
Review comment:
Returning `false` seemed more appropriate. That ensures we will get a backoff before the next retry.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547591383
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1055,14 +1055,14 @@ private boolean handleFetchResponse(
partitionResponse.snapshotId().endOffset(),
partitionResponse.snapshotId().epoch()
);
- return false;
+ return true;
Review comment:
Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #9553:
URL: https://github.com/apache/kafka/pull/9553
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547556738
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1037,6 +1047,35 @@ private boolean handleFetchResponse(
logger.info("Truncated to offset {} from Fetch response from leader {}",
truncationOffset, quorum.leaderIdOrNil());
});
+ } else if (partitionResponse.snapshotId().epoch() >= 0 ||
+ partitionResponse.snapshotId().endOffset() >= 0) {
+ // The leader is asking us to fetch a snapshot
+
+ if (partitionResponse.snapshotId().epoch() < 0) {
+ throw new KafkaException(
Review comment:
> I would suggest that we log an error saying that the remote replica seemed to return an invalid response and just keep fetching. Then a user can see the log message and restart the remote replica.
Yeah. This is what I implemented and added a test for it. In other words.
1. Log an error message
2. Tell the raft client that the response was handle successfully but the fetch timer was not reset
In practice this results in the follower continuing to send `Fetch` requests. After `fetchTimeoutMs` the follower will transition to candidate as the existing client code does. See https://github.com/apache/kafka/pull/9553/files#diff-86474ad1438150630c21b29a3da2f6dd79d1357e33ac034f00e5fcef0f2e889cR350
Let me know if this is what you were thinking.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547554416
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
+ ByteBuffer buffer = ByteBuffer.allocate(Math.min(data.maxBytes(), maxSnapshotSize));
+ snapshot.read(buffer, partitionSnapshot.position());
Review comment:
Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547470779
##########
File path: clients/src/main/resources/common/message/FetchSnapshotResponse.json
##########
@@ -0,0 +1,59 @@
+// 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": "response",
+ "name": "FetchSnapshotResponse",
+ "validVersions": "0",
+ "flexibleVersions": "0+",
+ "fields": [
+ { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
+ "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+", "ignorable": false,
+ "about": "The top level response error code." },
+ { "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": "Index", "type": "int32", "versions": "0+",
+ "about": "The partition index." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+",
+ "about": "The error code, or 0 if there was no fetch error." },
+ { "name": "SnapshotId", "type": "SnapshotId", "versions": "0+",
+ "about": "The snapshot endOffset and epoch fetched",
+ "fields": [
+ { "name": "EndOffset", "type": "int64", "versions": "0+" },
+ { "name": "Epoch", "type": "int32", "versions": "0+" }
+ ]},
+ { "name": "CurrentLeader", "type": "LeaderIdAndEpoch",
+ "versions": "0+", "taggedVersions": "0+", "tag": 0, "fields": [
+ { "name": "LeaderId", "type": "int32", "versions": "0+",
+ "about": "The ID of the current leader or -1 if the leader is unknown."},
+ { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+ "about": "The latest known leader epoch"}
+ ]},
+ { "name": "Size", "type": "int64", "versions": "0+",
+ "about": "The total size of the snapshot." },
+ { "name": "Position", "type": "int64", "versions": "0+",
+ "about": "The starting byte position within the snapshot included in the Bytes field." },
+ { "name": "Bytes", "type": "bytes", "versions": "0+", "zeroCopy": true,
Review comment:
Yeah. I created this issue: https://issues.apache.org/jira/browse/KAFKA-10694
I think we have two options:
1. Use "records" as you have suggested.
2. Similar to "records" introduce a new type called for example `Memory` that has two implementations one that encapsulates `ByteBuffer` and another that encapsulates a file and a start position. This is similar to "records", `MemoryRecords` and `FileRecords` without the requirement that the start needs to be a valid `RecordBatch`.
I want to play around with this soon and restart the conversation on that issue and PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547481788
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1629,16 +1872,29 @@ private long pollFollowerAsVoter(FollowerState state, long currentTimeMs) throws
transitionToCandidate(currentTimeMs);
return 0L;
} else {
- long backoffMs = maybeSendRequest(
- currentTimeMs,
- state.leaderId(),
- this::buildFetchRequest
- );
+ long backoffMs;
+ if (state.fetchingSnapshot().isPresent()) {
Review comment:
I cleaned this up by moving it to a method we can reuse. Let me know if that addresses your concern.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547483452
##########
File path: raft/src/main/java/org/apache/kafka/raft/LeaderState.java
##########
@@ -287,4 +287,7 @@ public String name() {
return "Leader";
}
+ @Override
+ public void close() {}
Review comment:
I prefer to have each of the epoch states explicitly opt out of this `close` method. This makes it clear that this state doesn't have any resource that it wishes to clean/close during a transition. Instead of future code changes forgetting to override this method. What do you think?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547424374
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
Review comment:
Hmm. I don't think so. `ifPresent` returns `void` and this part of the code wants to return a `FetchSnapshotResponseData` if there was a validation error. We can do a `map` followed by a `isPresent` and `get` but I think this is more consistent with the rest of the code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r547446688
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -1101,6 +1140,174 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
);
}
+ private FetchSnapshotResponseData handleFetchSnapshotRequest(
+ RaftRequest.Inbound requestMetadata
+ ) throws IOException {
+ FetchSnapshotRequestData data = (FetchSnapshotRequestData) requestMetadata.data;
+
+ if (data.topics().size() != 1 && data.topics().get(0).partitions().size() != 1) {
+ return FetchSnapshotResponse.withTopError(Errors.INVALID_REQUEST);
+ }
+
+ Optional<FetchSnapshotRequestData.PartitionSnapshot> partitionSnapshotOpt = FetchSnapshotRequest
+ .forTopicPartition(data, log.topicPartition());
+ if (!partitionSnapshotOpt.isPresent()) {
+ // The Raft client assumes that there is only one topic partition.
+ TopicPartition unknownTopicPartition = new TopicPartition(
+ data.topics().get(0).name(),
+ data.topics().get(0).partitions().get(0).partition()
+ );
+
+ return FetchSnapshotResponse.singleton(
+ unknownTopicPartition,
+ responsePartitionSnapshot -> {
+ return responsePartitionSnapshot
+ .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code());
+ }
+ );
+ }
+
+ FetchSnapshotRequestData.PartitionSnapshot partitionSnapshot = partitionSnapshotOpt.get();
+ Optional<Errors> leaderValidation = validateLeaderOnlyRequest(
+ partitionSnapshot.currentLeaderEpoch()
+ );
+ if (leaderValidation.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(leaderValidation.get().code());
+ }
+ );
+ }
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(
+ partitionSnapshot.snapshotId().endOffset(),
+ partitionSnapshot.snapshotId().epoch()
+ );
+ Optional<RawSnapshotReader> snapshotOpt = log.readSnapshot(snapshotId);
+ if (!snapshotOpt.isPresent()) {
+ return FetchSnapshotResponse.singleton(
+ log.topicPartition(),
+ responsePartitionSnapshot -> {
+ return addQuorumLeader(responsePartitionSnapshot)
+ .setErrorCode(Errors.SNAPSHOT_NOT_FOUND.code());
+ }
+ );
+ }
+
+ try (RawSnapshotReader snapshot = snapshotOpt.get()) {
+ int maxSnapshotSize;
+ try {
+ maxSnapshotSize = Math.toIntExact(snapshot.sizeInBytes());
+ } catch (ArithmeticException e) {
+ maxSnapshotSize = Integer.MAX_VALUE;
+ }
+
+ // TODO: Make sure that we also limit based on the fetch max bytes configuration
Review comment:
I created this issue: https://issues.apache.org/jira/browse/KAFKA-10884
We can fix this for Raft Client's implementation of both `Fetch` and `FetchSnapshot` at the same time.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dengziming commented on a change in pull request #9553: KAFKA-10427: Fetch snapshot
Posted by GitBox <gi...@apache.org>.
dengziming commented on a change in pull request #9553:
URL: https://github.com/apache/kafka/pull/9553#discussion_r548960063
##########
File path: clients/src/main/resources/common/message/FetchSnapshotResponse.json
##########
@@ -0,0 +1,59 @@
+// 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": "response",
+ "name": "FetchSnapshotResponse",
+ "validVersions": "0",
+ "flexibleVersions": "0+",
+ "fields": [
+ { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
+ "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+", "ignorable": false,
+ "about": "The top level response error code." },
+ { "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": "Index", "type": "int32", "versions": "0+",
+ "about": "The partition index." },
+ { "name": "ErrorCode", "type": "int16", "versions": "0+",
+ "about": "The error code, or 0 if there was no fetch error." },
+ { "name": "SnapshotId", "type": "SnapshotId", "versions": "0+",
+ "about": "The snapshot endOffset and epoch fetched",
+ "fields": [
+ { "name": "EndOffset", "type": "int64", "versions": "0+" },
+ { "name": "Epoch", "type": "int32", "versions": "0+" }
+ ]},
+ { "name": "CurrentLeader", "type": "LeaderIdAndEpoch",
+ "versions": "0+", "taggedVersions": "0+", "tag": 0, "fields": [
+ { "name": "LeaderId", "type": "int32", "versions": "0+",
+ "about": "The ID of the current leader or -1 if the leader is unknown."},
+ { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+ "about": "The latest known leader epoch"}
+ ]},
+ { "name": "Size", "type": "int64", "versions": "0+",
+ "about": "The total size of the snapshot." },
+ { "name": "Position", "type": "int64", "versions": "0+",
+ "about": "The starting byte position within the snapshot included in the Bytes field." },
+ { "name": "Bytes", "type": "bytes", "versions": "0+", "zeroCopy": true,
Review comment:
@jsancio Can we add an index file to the snapshot file so that the `FetchSnaphotResponseData` can also satisfy the requirement that the start is a valid `RecordBatch`? I can help with some work of this issue: https://issues.apache.org/jira/browse/KAFKA-10694.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org