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 2021/04/25 02:38:08 UTC
[GitHub] [kafka] feyman2016 opened a new pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
feyman2016 opened a new pull request #10593:
URL: https://github.com/apache/kafka/pull/10593
*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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619988848
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
Thanks, previous I thought the quorum epoch is the leader epoch cache as a mistake~
Updated the PR, in the jira, one thing I'm not sure about is that:
> 2. The epoch of the snapshot is equal to the quorum epoch.
I think the snapshotId's epoch <= quorum epoch should be fine?
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619988848
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
Thanks, previous I thought the quorum epoch is the leader epoch cache as a mistake~
Updated the PR, one thing I'm not sure about is that:
> The epoch of the snapshot is equal to the quorum epoch.
I think the snapshotId's epoch <= quorum epoch should be fine?
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r643850904
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2276,6 +2277,24 @@ public void resign(int epoch) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset < snapshotId.offset) {
+ throw new IllegalArgumentException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
Review comment:
Fixed~
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r620232416
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
Yeah, This is not strictly required for correctness. Oh, I see, the check in 2280 is checking that the `epoch > current epoch`.
I mistakenly read it as `epoch != current epoch`. If we perform this check we are basically saying that the caller of `createSnapshot` needs to catch up to the current quorum epoch before it can generate a snapshot. Yes, I think `epoch <= quorum epoch` is fine. Let me think about it and I'll update the Jira.
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r640805609
##########
File path: raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java
##########
@@ -963,6 +952,20 @@ static void verifyLeaderChangeMessage(
new HashSet<>(leaderChangeMessage.grantingVoters()));
}
+ public void advanceLeaderHighWatermarkToEndOffset() throws InterruptedException {
+ Integer replicaId = null;
+ for (Integer voter: voters) {
+ if (voter != localId.getAsInt()) {
+ replicaId = voter;
+ break;
+ }
+ }
+ deliverRequest(fetchRequest(currentEpoch(), replicaId, log.endOffset().offset, currentEpoch(), 0));
+ pollUntilResponse();
+ assertSentFetchPartitionResponse(Errors.NONE, currentEpoch(), localId);
+ assertEquals(log.endOffset().offset, client.highWatermark().getAsLong());
+ }
Review comment:
I needed something similar in a PR I am currently working and this is what I have if you want to adapt it to your PR:
```java
public void advanceLocalLeaderHighWatermarkToLogEndOffset() throws InterruptedException {
assertEquals(localId, currentLeader());
long localLogEndOffset = log.endOffset().offset;
Set<Integer> followers = voters.stream().filter(voter -> voter != localId.getAsInt()).collect(Collectors.toSet());
// Send a request from every follower
for (int follower : followers) {
deliverRequest(
fetchRequest(currentEpoch(), follower, localLogEndOffset, currentEpoch(), 0)
);
pollUntilResponse();
assertSentFetchPartitionResponse(Errors.NONE, currentEpoch(), localId);
}
pollUntil(() -> OptionalLong.of(localLogEndOffset).equals(client.highWatermark()));
}
```
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623574263
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterTest.java
##########
@@ -38,7 +38,7 @@
@Test
public void testWritingSnapshot() throws IOException {
- OffsetAndEpoch id = new OffsetAndEpoch(10L, 3);
+ OffsetAndEpoch id = new OffsetAndEpoch(1L, 1);
Review comment:
I checked, it's because the client become leader and append one leaderChangeMessage, so the high-watermark is `LogOffsetMetadata(offset=1, metadata=Optional[MockOffsetMetadata(id=1)])` , detailed stacktrace as below:
```
at org.apache.kafka.raft.KafkaRaftClient.appendAsLeader(KafkaRaftClient.java:1162)
at org.apache.kafka.raft.KafkaRaftClient.appendLeaderChangeMessage(KafkaRaftClient.java:450)
at org.apache.kafka.raft.KafkaRaftClient.onBecomeLeader(KafkaRaftClient.java:406)
at org.apache.kafka.raft.KafkaRaftClient.maybeTransitionToLeader(KafkaRaftClient.java:464)
at org.apache.kafka.raft.KafkaRaftClient.onBecomeCandidate(KafkaRaftClient.java:473)
at org.apache.kafka.raft.KafkaRaftClient.transitionToCandidate(KafkaRaftClient.java:493)
at org.apache.kafka.raft.KafkaRaftClient.initialize(KafkaRaftClient.java:375)
at org.apache.kafka.raft.RaftClientTestContext$Builder.build(RaftClientTestContext.java:231)
at org.apache.kafka.snapshot.SnapshotWriterTest.testWritingSnapshot(SnapshotWriterTest.java:43)
```
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r620232416
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
Yeah, This is not strictly required for correctness. Oh, I see, the check in 2280 is checking that the `epoch > current epoch`.
I mistakenly read it as `epoch != current epoch`. If we perform this check we are basically saying that the caller of `createSnapshot` needs to catch up to the current quorum epoch before it can generate a snapshot. Yes, I think epoch <= quorum epoch` is fine. Let me think about it and I'll update the Jira.
--
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 pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-852353685
> Are there any concerns with multi-threading here? I think the answer is "no". Any thoughts @feyman2016 or @jsancio?
Good point. There may be some concurrency issue with `createSnapshot`. How about moving the validation to `onSnapshotFrozen`?
This conversation also reminded me to create this issue: https://issues.apache.org/jira/browse/KAFKA-12873. It is slightly related to this point.
--
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] feyman2016 closed pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 closed pull request #10593:
URL: https://github.com/apache/kafka/pull/10593
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619793709
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -96,7 +96,7 @@ public BatchBuilder(
}
/**
- * Append a record to this patch. The caller must first verify there is room for the batch
+ * Append a record to this batch. The caller must first verify there is room for the batch
Review comment:
Side fix
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterTest.java
##########
@@ -38,7 +38,7 @@
@Test
public void testWritingSnapshot() throws IOException {
- OffsetAndEpoch id = new OffsetAndEpoch(10L, 3);
+ OffsetAndEpoch id = new OffsetAndEpoch(0L, 1);
Review comment:
The highwatermark here is 1, so we need to make the snapshotId's endOffset < 1.
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,51 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1. high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 2. high watermark must larger than the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong(), currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 3. the current leader epoch cache must larger than the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+ }
+
+ private void advanceHighWatermark(RaftClientTestContext context,
Review comment:
Extract the functionality to avoid duplicate
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
Review comment:
Conceptually, the `snapshotId.offset=highWatermarkOpt.get().offset` is ok, because the record at `snapshotId.offset` is not included in the snapshot, but I'm not sure if there are other restrictions because in the Jira description, it says: `The end offset and epoch of the snapshot is less than the high-watermark`, please kindly advice @jsancio
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
Thanks, previous I thought the quorum epoch is the leader epoch cache as a mistake~
Updated the PR, one thing I'm not sure about is that:
> The epoch of the snapshot is equal to the quorum epoch.
I think the snapshotId's epoch <= quorum epoch should be fine?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
Thanks, previous I thought the quorum epoch is the leader epoch cache as a mistake~
Updated the PR, in the jira, one thing I'm not sure about is that:
> 2. The epoch of the snapshot is equal to the quorum epoch.
I think the snapshotId's epoch <= quorum epoch should be fine?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
Review comment:
Thank you, I updated the PR as well
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r622425656
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,57 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
Review comment:
I would append a couple of batches after advancing the high-watermark. At this point the HWM equals the LEO.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,25 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset < snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
Review comment:
What do you mean by ", since the there should be a minimum size of records between the latest snapshot and the high-watermark when creating snapshot"?
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,57 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 2 the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+
+ // 3 the snapshotId should be validated against endOffsetForEpoch
+ OffsetAndEpoch endOffsetForEpoch = context.log.endOffsetForEpoch(currentEpoch);
Review comment:
Let's use `epoch` instead of `currentEpoch`. Since we are using `currentEpoch`, the `endOffsetForEpoch.offset` will equal the LEO. If you instead use `epoch` then the `endOffsetForEpoch.offset`. will be `4` which is less than the LEO (`5`).
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,57 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
Review comment:
This test is checking the `snapshotId` when the `KafkaRaftClient` is the leader. Let's have a similar test but when the `KafkaRaftClient` is a follower. Take a look at `KafkaRaftClient::testEmptyRecordSetInFetchResponse` for a simple example of how you can advance the high-watermark on the follower.
Note that the followers don't need to wait for the "high-watermark" to reach the current epoch to set (and advance) the high-watermark.
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterTest.java
##########
@@ -38,7 +38,7 @@
@Test
public void testWritingSnapshot() throws IOException {
- OffsetAndEpoch id = new OffsetAndEpoch(10L, 3);
+ OffsetAndEpoch id = new OffsetAndEpoch(1L, 1);
Review comment:
This applies to all of the tests in the suite.
Do you know why the `context.client.createSnapshot` doesn't throw an exception in these tests? Shouldn't the high-watermark be unknown (`Optional.empty`) in all of these cases?
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,25 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset < snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int quorumEpoch = quorum().epoch();
+ if (snapshotId.epoch > quorumEpoch) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the quorum epoch: " + quorumEpoch);
+ }
+ OffsetAndEpoch endOffsetAndEpoch = log.endOffsetForEpoch(snapshotId.epoch);
+ if (endOffsetAndEpoch.epoch != snapshotId.epoch || endOffsetAndEpoch.offset < snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId +
Review comment:
Let's use `InvalidArgumentException` instead of `KafkaException` for all of the exceptions thrown in this method.
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623566646
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,25 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset < snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
Review comment:
Based on https://cwiki.apache.org/confluence/display/KAFKA/KIP-630%3A+Kafka+Raft+Snapshot#KIP630:KafkaRaftSnapshot-WhentoSnapshot,
> 2. metadata.snapshot.min.new_records.size is the minimum size of the records in the replicated log between the latest snapshot and then high-watermark.
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-852954180
> > Are there any concerns with multi-threading here? I think the answer is "no". Any thoughts @feyman2016 or @jsancio?
>
> Good point. There may be some concurrency issue with `createSnapshot`. How about moving the validation to `onSnapshotFrozen`?
>
> This conversation also reminded me to create this issue: https://issues.apache.org/jira/browse/KAFKA-12873. It is slightly related to this point.
@jsancio Can you share more details about the possible concurrency scenario with `createSnapshot `? BTW, will moving the validation to `onSnapshotFrozen` imply that before creating the snapshot, there's no validation? I think maybe we can keep the validation here and add some additional check before `freeze()` which makes the snapshot visible?
--
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 #10593: KAFKA-10800 Enhance the test for validation when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
hachikuji merged pull request #10593:
URL: https://github.com/apache/kafka/pull/10593
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623897322
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterTest.java
##########
@@ -38,7 +38,7 @@
@Test
public void testWritingSnapshot() throws IOException {
- OffsetAndEpoch id = new OffsetAndEpoch(10L, 3);
+ OffsetAndEpoch id = new OffsetAndEpoch(1L, 1);
Review comment:
Oops, merged the latest trunk, and it failed because the igh-watermark be unknown `(Optional.empty)`, I need to update it
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619882084
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
Review comment:
You are correct. I think when I created the Jira I overlooked that both snapshot id's end offset and the high-watermark are exclusive values. Update the Jira's description.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
From the Jira:
> The end offset and epoch of the snapshot is valid based on the leader epoch cache.
How about also validating against the leader epoch cache? See https://github.com/apache/kafka/blob/trunk/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java#L124. This is important because both the snapshot and the leader epoch cache are used to validate offsets. See https://github.com/apache/kafka/blob/trunk/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java#L85. The term leader epoch cache comes the variable name `leaderEpochCache` used in `kafka.log.Log`.
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,51 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1. high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 2. high watermark must larger than the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong(), currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 3. the current leader epoch cache must larger than the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+ }
+
+ private void advanceHighWatermark(RaftClientTestContext context,
Review comment:
Thanks for clean up the code duplication.
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,51 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1. high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 2. high watermark must larger than the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong(), currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 3. the current leader epoch cache must larger than the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+ }
+
+ private void advanceHighWatermark(RaftClientTestContext context,
Review comment:
Thanks for cleaning up the code duplication.
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-826653099
@jsancio I updated the PR, and the failed tests should be unrelated, thanks!
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623569039
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,25 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset < snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int quorumEpoch = quorum().epoch();
+ if (snapshotId.epoch > quorumEpoch) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the quorum epoch: " + quorumEpoch);
+ }
+ OffsetAndEpoch endOffsetAndEpoch = log.endOffsetForEpoch(snapshotId.epoch);
+ if (endOffsetAndEpoch.epoch != snapshotId.epoch || endOffsetAndEpoch.offset < snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId +
Review comment:
Will fix, but I assume you want `IllegalArgumentException` instead of `InvalidArgumentException`?
--
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] feyman2016 edited a comment on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 edited a comment on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-826653099
@jsancio Thanks for the review! I updated the PR, and the failed tests should be unrelated, thanks!
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619793903
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,51 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1. high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 2. high watermark must larger than the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong(), currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 3. the current leader epoch cache must larger than the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+ }
+
+ private void advanceHighWatermark(RaftClientTestContext context,
Review comment:
Extract the functionality to avoid duplicate
--
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 #10593: KAFKA-10800 Enhance the test for validation when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r728399062
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1533,6 +1508,112 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotAsLeaderWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When leader creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ context.advanceLocalLeaderHighWatermarkToLogEndOffset();
+ // append some more records to make the LEO > high watermark
+ List<String> newRecords = Arrays.asList("d", "e", "f");
+ context.client.scheduleAppend(currentEpoch, newRecords);
+ context.time.sleep(context.appendLingerMs());
+ context.client.poll();
+ assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong() + newRecords.size());
+
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2 the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // 3 the snapshotId should be validated against endOffsetForEpoch
+ OffsetAndEpoch endOffsetForEpoch = context.log.endOffsetForEpoch(epoch);
+ assertEquals(epoch, endOffsetForEpoch.epoch);
+ OffsetAndEpoch invalidSnapshotId4 = new OffsetAndEpoch(endOffsetForEpoch.offset + 1, epoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId4.offset, invalidSnapshotId4.epoch));
+ }
+
+ @Test
+ public void testCreateSnapshotAsFollowerWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ int epoch = 5;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withElectedLeader(epoch, otherNodeId)
+ .build();
+ context.assertElectedLeader(epoch, otherNodeId);
+
+ // When follower creating snapshot:
+ // 1.1) high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(0, 0);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // Poll for our first fetch request
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ // The response does not advance the high watermark
+ List<String> records1 = Arrays.asList("a", "b", "c");
+ MemoryRecords batch1 = context.buildBatch(0L, 3, records1);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch1, 0L, Errors.NONE));
+ context.client.poll();
+
+ // 1.2) high watermark must larger than or equal to the snapshotId's endOffset
+ int currentEpoch = context.currentEpoch();
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2) the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // The high watermark advances to be larger than log.endOffsetForEpoch(3), to test the case 3
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 3L, 3);
+
+ List<String> records2 = Arrays.asList("d", "e", "f");
+ MemoryRecords batch2 = context.buildBatch(3L, 4, records2);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch2, 4L, Errors.NONE));
+ context.client.poll();
+ assertEquals(4L, context.client.highWatermark().getAsLong());
Review comment:
You are right @hachikuji . For line 1597 to be true, I think the test needs to do another round of fetch.
> // The high watermark advances to be larger than log.endOffsetForEpoch(3), to test the case 3
Line 1614 wants to fail because of an invalid offset and epoch based on the leader epoch cache. Not because it is greater than the high watermark.
```
assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId4.offset, invalidSnapshotId4.epoch));
```
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] feyman2016 commented on pull request #10593: KAFKA-10800 Enhance the test for validation when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-866954962
@jsancio Thanks!
@mumrah Hi, could you kindly help to review and merge? Thanks!
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-826304130
@jsancio Could you please help to review? Thanks!
Locally verified, all the failed tests should not be related. See failed tests in https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-10593/2/testReport/?cloudbees-analytics-link=scm-reporting%2Ftests%2Ffailed
--
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 pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-860980428
> Or may be we can make a seperate PR to handle the validation for onSnapshotFrozen?
I am okay with fixing this in a future PR. Do you want to go ahead and file an sub-task for https://issues.apache.org/jira/browse/KAFKA-10310 and link it here?
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619989241
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
Review comment:
Thank you, I updated the PR as well
--
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] mumrah commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
mumrah commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r643376055
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2276,6 +2277,24 @@ public void resign(int epoch) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
Review comment:
Can we make the comparisons with `snapshotId` consistently on the left-hand side?
E.g.,
```
snapshotId.offset > highWatermarkOpt.get().offset
```
Makes it a bit easier to parse this logic.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2276,6 +2277,24 @@ public void resign(int epoch) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset < snapshotId.offset) {
+ throw new IllegalArgumentException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
Review comment:
This might be better to split into two checks with different error messages. This would also let us avoid logging an `Optional`
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-850743673
@jsancio Thank you for the kindly review and help~
--
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 pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-858925958
> @jsancio Can you share more details about the possible concurrency scenario with createSnapshot ? BTW, will moving the validation to onSnapshotFrozen imply that before creating the snapshot, there's no validation? I think maybe we can keep the validation here and add some additional check before freeze() which makes the snapshot visible?
@feyman2016, I think it is reasonable to do both. Validate when `createSnapshot` is called and validate again in `onSnapshotFrozen`. In both cases this validation should be optional. Validate if it is created through `RaftClient.createSnapshot`. Don't validate if `KafkaRaftClient` creates the snapshot internally because of a `FetchResponse` from the leader.
I have been working on a PR related to this if you want to take a look: https://github.com/apache/kafka/pull/10786. It would be nice to get your PR merged before my 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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623971200
##########
File path: raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java
##########
@@ -168,12 +179,12 @@ Builder withMemoryPool(MemoryPool pool) {
return this;
}
- Builder withAppendLingerMs(int appendLingerMs) {
+ public Builder withAppendLingerMs(int appendLingerMs) {
Review comment:
Mark these methods as `public` to be reused in `SnapshotWriterTest`, in `SnapshotWriterTest`, it need a way to advance the high watermark of the snapshot validation won't be passed
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r641280932
##########
File path: raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java
##########
@@ -963,6 +952,20 @@ static void verifyLeaderChangeMessage(
new HashSet<>(leaderChangeMessage.grantingVoters()));
}
+ public void advanceLeaderHighWatermarkToEndOffset() throws InterruptedException {
+ Integer replicaId = null;
+ for (Integer voter: voters) {
+ if (voter != localId.getAsInt()) {
+ replicaId = voter;
+ break;
+ }
+ }
+ deliverRequest(fetchRequest(currentEpoch(), replicaId, log.endOffset().offset, currentEpoch(), 0));
+ pollUntilResponse();
+ assertSentFetchPartitionResponse(Errors.NONE, currentEpoch(), localId);
+ assertEquals(log.endOffset().offset, client.highWatermark().getAsLong());
+ }
Review comment:
Thank you, I copy-pasted yours.
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619882084
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
Review comment:
You are correct. I think when I created the Jira I overlooked that both snapshot id's end offset and the high-watermark are exclusive values. Update the Jira's description.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
+ }
+ int leaderEpoch = quorum().epoch();
+ if (snapshotId.epoch > leaderEpoch) {
+ throw new KafkaException("Trying to creating snapshot with snapshotId: " + snapshotId + " whose epoch is" +
+ " larger than the current leader epoch: " + leaderEpoch);
+ }
Review comment:
From the Jira:
> The end offset and epoch of the snapshot is valid based on the leader epoch cache.
How about also validating against the leader epoch cache? See https://github.com/apache/kafka/blob/trunk/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java#L124. This is important because both the snapshot and the leader epoch cache are used to validate offsets. See https://github.com/apache/kafka/blob/trunk/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java#L85. The term leader epoch cache comes the variable name `leaderEpochCache` used in `kafka.log.Log`.
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,51 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1. high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 2. high watermark must larger than the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong(), currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 3. the current leader epoch cache must larger than the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+ }
+
+ private void advanceHighWatermark(RaftClientTestContext context,
Review comment:
Thanks for clean up the code duplication.
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r643850801
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2276,6 +2277,24 @@ public void resign(int epoch) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
Review comment:
Make sense, fixed
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r654340330
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2255,9 +2255,9 @@ public void resign(int epoch) {
@Override
public Optional<SnapshotWriter<T>> createSnapshot(long committedOffset, int committedEpoch) {
- return log.createNewSnapshot(
- new OffsetAndEpoch(committedOffset + 1, committedEpoch)
- ).map(snapshot -> {
+ final OffsetAndEpoch snapshotId = new OffsetAndEpoch(committedOffset + 1, committedEpoch);
+ validateSnapshotId(snapshotId);
Review comment:
@jsancio Totally understood, I just confirmed the validation logic has been covered already as you mentioned, so just kept the test.
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2255,9 +2255,9 @@ public void resign(int epoch) {
@Override
public Optional<SnapshotWriter<T>> createSnapshot(long committedOffset, int committedEpoch) {
- return log.createNewSnapshot(
- new OffsetAndEpoch(committedOffset + 1, committedEpoch)
- ).map(snapshot -> {
+ final OffsetAndEpoch snapshotId = new OffsetAndEpoch(committedOffset + 1, committedEpoch);
+ validateSnapshotId(snapshotId);
Review comment:
@jsancio Totally understood, I just confirmed the validation logic has been covered already as you mentioned, so just kept the test and updated the PR name
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623897322
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterTest.java
##########
@@ -38,7 +38,7 @@
@Test
public void testWritingSnapshot() throws IOException {
- OffsetAndEpoch id = new OffsetAndEpoch(10L, 3);
+ OffsetAndEpoch id = new OffsetAndEpoch(1L, 1);
Review comment:
Oops, merged the latest trunk, and it failed because the high-watermark is unknown `(Optional.empty)`, I need to update it
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-826304130
@jsancio Could you please help to review? Thanks!
Locally verified, all the failed tests should not be related. See failed tests in https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-10593/2/testReport/?cloudbees-analytics-link=scm-reporting%2Ftests%2Ffailed
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623574263
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterTest.java
##########
@@ -38,7 +38,7 @@
@Test
public void testWritingSnapshot() throws IOException {
- OffsetAndEpoch id = new OffsetAndEpoch(10L, 3);
+ OffsetAndEpoch id = new OffsetAndEpoch(1L, 1);
Review comment:
I checked, it's because the client become leader and append one leaderChangeMessage, so the high-watermark is `LogOffsetMetadata(offset=1, metadata=Optional[MockOffsetMetadata(id=1)])` , detailed stacktrace as below:
```
at org.apache.kafka.raft.KafkaRaftClient.appendAsLeader(KafkaRaftClient.java:1162)
at org.apache.kafka.raft.KafkaRaftClient.appendLeaderChangeMessage(KafkaRaftClient.java:450)
at org.apache.kafka.raft.KafkaRaftClient.onBecomeLeader(KafkaRaftClient.java:406)
at org.apache.kafka.raft.KafkaRaftClient.maybeTransitionToLeader(KafkaRaftClient.java:464)
at org.apache.kafka.raft.KafkaRaftClient.onBecomeCandidate(KafkaRaftClient.java:473)
at org.apache.kafka.raft.KafkaRaftClient.transitionToCandidate(KafkaRaftClient.java:493)
at org.apache.kafka.raft.KafkaRaftClient.initialize(KafkaRaftClient.java:375)
at org.apache.kafka.raft.RaftClientTestContext$Builder.build(RaftClientTestContext.java:231)
at org.apache.kafka.snapshot.SnapshotWriterTest.testWritingSnapshot(SnapshotWriterTest.java:43)
at sun.reflect.NativeMethodAccessorImpl.invoke0(NativeMethodAccessorImpl.java:-1)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:688)
at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor$$Lambda$141.512404647.apply(Unknown Source:-1)
at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall$$Lambda$142.1063942252.apply(Unknown Source:-1)
at org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
at org.junit.jupiter.engine.execution.ExecutableInvoker$$Lambda$336.2109790380.apply(Unknown Source:-1)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$6(TestMethodTestDescriptor.java:210)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor$$Lambda$348.1944088123.execute(Unknown Source:-1)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:206)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:131)
at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:65)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:139)
at org.junit.platform.engine.support.hierarchical.NodeTestTask$$Lambda$284.547190487.execute(Unknown Source:-1)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
at org.junit.platform.engine.support.hierarchical.NodeTestTask$$Lambda$283.585275267.invoke(Unknown Source:-1)
at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
at org.junit.platform.engine.support.hierarchical.NodeTestTask$$Lambda$282.1799009183.execute(Unknown Source:-1)
at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService$$Lambda$288.130873454.accept(Unknown Source:-1)
```
--
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 pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-860908546
> I assume a KIP is needed.
We don't need a KIP. All of these API are internal APIs that are not accessible/publish to projects external to Apache Kafka.
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623706130
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,57 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
Review comment:
Make sense, fixed
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,57 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 2 the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+
+ // 3 the snapshotId should be validated against endOffsetForEpoch
+ OffsetAndEpoch endOffsetForEpoch = context.log.endOffsetForEpoch(currentEpoch);
Review comment:
Fixed
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-849753337
@jsancio Addressed the comments but just found that I would need to generalize `context.advanceLeaderHighWatermarkToEndOffset ()`, will update later~
--
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] feyman2016 edited a comment on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 edited a comment on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-832764031
@jsancio Hi, could you help to take a look? Thanks!
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623972347
##########
File path: raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java
##########
@@ -108,6 +108,17 @@
private final List<RaftResponse.Outbound> sentResponses = new ArrayList<>();
+ public static void advanceHighWatermark(RaftClientTestContext context,
Review comment:
This method will be called in both `KafkaRaftClientSnapshotTest` and `SnapshotWriterTest`, so mark it as public
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r623706510
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,57 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
Review comment:
Added the testcase `testCreateSnapshotAsFollowerWithInvalidSnapshotId`
--
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] feyman2016 removed a comment on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 removed a comment on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-830564003
@jsancio Hi, I addressed the comments and the failed tests should not be related
--
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] feyman2016 closed pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 closed pull request #10593:
URL: https://github.com/apache/kafka/pull/10593
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-850279646
Thanks for your quick reply, call for review again @jsancio :)
The failed test: `org.apache.kafka.connect.integration.BlockingConnectorTest.testBlockInSinkTaskStop` is not related
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Enhance the test for validation when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-968300161
@hachikuji @jsancio Just noticed this has been merged, thanks a lot for the review and also the commit!
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-849787584
> @jsancio Addressed the comments but just found that I would need to generalize context.advanceLeaderHighWatermarkToEndOffset (), will update later~
I needed something similar in a PR I am currently working and this is what I have if you want to adapt it to your PR:
```java
public void advanceLocalLeaderHighWatermarkToLogEndOffset() throws InterruptedException {
assertEquals(localId, currentLeader());
long localLogEndOffset = log.endOffset().offset;
Set<Integer> followers = voters.stream().filter(voter -> voter != localId.getAsInt()).collect(Collectors.toSet());
// Send a request from every follower
for (int follower : followers) {
deliverRequest(
fetchRequest(currentEpoch(), follower, localLogEndOffset, currentEpoch(), 0)
);
pollUntilResponse();
assertSentFetchPartitionResponse(Errors.NONE, currentEpoch(), localId);
}
pollUntil(() -> OptionalLong.of(localLogEndOffset).equals(client.highWatermark()));
}
```
--
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] feyman2016 edited a comment on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 edited a comment on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-860066550
> I have been working on a PR related to this if you want to take a look: #10786. It would be nice to get your PR merged before my PR.
@jsancio Thanks for the reply, now I understand the background. I think the optional validation for when `createSnapshot` is called has been covered in this PR, while the case for `onSnapshotFrozen` has not been covered yet. Looking at the code, `onSnapshotFrozen` is called by`FileRawSnapshotWriter#freeze` which may be called by `KafkaRaftClient ` or `SnapshotWriter`, if we want to do the optional validation, I assume a KIP is needed for tweaking `RawSnapshotWriter#freeze()` as sth like `RawSnapshotWriter#freeze(Origin origin)`, based on the origin, it can decide whether validation is needed. But since this is PR is preferable to be merged before #10786 , would it take too much time to go through the KIP and contain the change in this PR? Or may be we can make a seperate PR to handle the validation for `onSnapshotFrozen`?
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-830563990
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-852954993
@mumrah Thanks for the feedback, just addressed, but the potential concurrency issue needs more discussion~
--
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 #10593: KAFKA-10800 Enhance the test for validation when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r728392933
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1533,6 +1508,112 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotAsLeaderWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When leader creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ context.advanceLocalLeaderHighWatermarkToLogEndOffset();
+ // append some more records to make the LEO > high watermark
+ List<String> newRecords = Arrays.asList("d", "e", "f");
+ context.client.scheduleAppend(currentEpoch, newRecords);
+ context.time.sleep(context.appendLingerMs());
+ context.client.poll();
+ assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong() + newRecords.size());
+
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2 the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // 3 the snapshotId should be validated against endOffsetForEpoch
+ OffsetAndEpoch endOffsetForEpoch = context.log.endOffsetForEpoch(epoch);
+ assertEquals(epoch, endOffsetForEpoch.epoch);
+ OffsetAndEpoch invalidSnapshotId4 = new OffsetAndEpoch(endOffsetForEpoch.offset + 1, epoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId4.offset, invalidSnapshotId4.epoch));
+ }
+
+ @Test
+ public void testCreateSnapshotAsFollowerWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ int epoch = 5;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withElectedLeader(epoch, otherNodeId)
+ .build();
+ context.assertElectedLeader(epoch, otherNodeId);
+
+ // When follower creating snapshot:
+ // 1.1) high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(0, 0);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // Poll for our first fetch request
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ // The response does not advance the high watermark
+ List<String> records1 = Arrays.asList("a", "b", "c");
+ MemoryRecords batch1 = context.buildBatch(0L, 3, records1);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch1, 0L, Errors.NONE));
+ context.client.poll();
+
+ // 1.2) high watermark must larger than or equal to the snapshotId's endOffset
+ int currentEpoch = context.currentEpoch();
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2) the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // The high watermark advances to be larger than log.endOffsetForEpoch(3), to test the case 3
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 3L, 3);
+
+ List<String> records2 = Arrays.asList("d", "e", "f");
+ MemoryRecords batch2 = context.buildBatch(3L, 4, records2);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch2, 4L, Errors.NONE));
+ context.client.poll();
+ assertEquals(4L, context.client.highWatermark().getAsLong());
Review comment:
Seems like we should use `3L` instead. The leader would not have been able to advance the high watermark past the fetch offset of `3L`.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #10593: KAFKA-10800 Enhance the test for validation when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r727462578
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1533,6 +1508,112 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotAsLeaderWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When leader creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ context.advanceLocalLeaderHighWatermarkToLogEndOffset();
+ // append some more records to make the LEO > high watermark
+ List<String> newRecords = Arrays.asList("d", "e", "f");
+ context.client.scheduleAppend(currentEpoch, newRecords);
+ context.time.sleep(context.appendLingerMs());
+ context.client.poll();
+ assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong() + newRecords.size());
+
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2 the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // 3 the snapshotId should be validated against endOffsetForEpoch
+ OffsetAndEpoch endOffsetForEpoch = context.log.endOffsetForEpoch(epoch);
+ assertEquals(epoch, endOffsetForEpoch.epoch);
+ OffsetAndEpoch invalidSnapshotId4 = new OffsetAndEpoch(endOffsetForEpoch.offset + 1, epoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId4.offset, invalidSnapshotId4.epoch));
+ }
+
+ @Test
+ public void testCreateSnapshotAsFollowerWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ int epoch = 5;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withElectedLeader(epoch, otherNodeId)
+ .build();
+ context.assertElectedLeader(epoch, otherNodeId);
+
+ // When follower creating snapshot:
+ // 1.1) high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(0, 0);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // Poll for our first fetch request
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ // The response does not advance the high watermark
+ List<String> records1 = Arrays.asList("a", "b", "c");
+ MemoryRecords batch1 = context.buildBatch(0L, 3, records1);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch1, 0L, Errors.NONE));
+ context.client.poll();
+
+ // 1.2) high watermark must larger than or equal to the snapshotId's endOffset
+ int currentEpoch = context.currentEpoch();
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2) the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // The high watermark advances to be larger than log.endOffsetForEpoch(3), to test the case 3
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 3L, 3);
+
+ List<String> records2 = Arrays.asList("d", "e", "f");
+ MemoryRecords batch2 = context.buildBatch(3L, 4, records2);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch2, 4L, Errors.NONE));
+ context.client.poll();
+ assertEquals(4L, context.client.highWatermark().getAsLong());
Review comment:
This is minor but so we don't confuse future readers of this code, I think the watermark is suppose to be `6L` instead of `4L`. The high watermark should always be at batch boundaries.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jsancio commented on a change in pull request #10593: KAFKA-10800 Enhance the test for validation when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r727462578
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1533,6 +1508,112 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotAsLeaderWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When leader creating snapshot:
+ // 1.1 high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // 1.2 high watermark must larger than or equal to the snapshotId's endOffset
+ context.advanceLocalLeaderHighWatermarkToLogEndOffset();
+ // append some more records to make the LEO > high watermark
+ List<String> newRecords = Arrays.asList("d", "e", "f");
+ context.client.scheduleAppend(currentEpoch, newRecords);
+ context.time.sleep(context.appendLingerMs());
+ context.client.poll();
+ assertEquals(context.log.endOffset().offset, context.client.highWatermark().getAsLong() + newRecords.size());
+
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2 the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // 3 the snapshotId should be validated against endOffsetForEpoch
+ OffsetAndEpoch endOffsetForEpoch = context.log.endOffsetForEpoch(epoch);
+ assertEquals(epoch, endOffsetForEpoch.epoch);
+ OffsetAndEpoch invalidSnapshotId4 = new OffsetAndEpoch(endOffsetForEpoch.offset + 1, epoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId4.offset, invalidSnapshotId4.epoch));
+ }
+
+ @Test
+ public void testCreateSnapshotAsFollowerWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = 1;
+ int epoch = 5;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .withElectedLeader(epoch, otherNodeId)
+ .build();
+ context.assertElectedLeader(epoch, otherNodeId);
+
+ // When follower creating snapshot:
+ // 1.1) high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(0, 0);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId1.offset, invalidSnapshotId1.epoch));
+
+ // Poll for our first fetch request
+ context.pollUntilRequest();
+ RaftRequest.Outbound fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 0L, 0);
+
+ // The response does not advance the high watermark
+ List<String> records1 = Arrays.asList("a", "b", "c");
+ MemoryRecords batch1 = context.buildBatch(0L, 3, records1);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch1, 0L, Errors.NONE));
+ context.client.poll();
+
+ // 1.2) high watermark must larger than or equal to the snapshotId's endOffset
+ int currentEpoch = context.currentEpoch();
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() + 1, currentEpoch);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId2.offset, invalidSnapshotId2.epoch));
+
+ // 2) the quorum epoch must larger than or equal to the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 2, currentEpoch + 1);
+ assertThrows(IllegalArgumentException.class, () -> context.client.createSnapshot(invalidSnapshotId3.offset, invalidSnapshotId3.epoch));
+
+ // The high watermark advances to be larger than log.endOffsetForEpoch(3), to test the case 3
+ context.pollUntilRequest();
+ fetchRequest = context.assertSentFetchRequest();
+ assertTrue(voters.contains(fetchRequest.destinationId()));
+ context.assertFetchRequestData(fetchRequest, epoch, 3L, 3);
+
+ List<String> records2 = Arrays.asList("d", "e", "f");
+ MemoryRecords batch2 = context.buildBatch(3L, 4, records2);
+ context.deliverResponse(fetchRequest.correlationId, fetchRequest.destinationId(),
+ context.fetchResponse(epoch, otherNodeId, batch2, 4L, Errors.NONE));
+ context.client.poll();
+ assertEquals(4L, context.client.highWatermark().getAsLong());
Review comment:
This is minor but so we don't confuse future readers of this code, I think the watermark is suppose to be `6L` instead of `4L`. The high watermark should always be at batch boundaries.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619793709
##########
File path: raft/src/main/java/org/apache/kafka/raft/internals/BatchBuilder.java
##########
@@ -96,7 +96,7 @@ public BatchBuilder(
}
/**
- * Append a record to this patch. The caller must first verify there is room for the batch
+ * Append a record to this batch. The caller must first verify there is room for the batch
Review comment:
Side fix
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-860066550
> I have been working on a PR related to this if you want to take a look: #10786. It would be nice to get your PR merged before my PR.
Thanks for the reply, now I understand the background. I think the optional validation for when `createSnapshot` is called has been covered in this PR, while the case for `onSnapshotFrozen` has not been covered yet. Looking at the code, `onSnapshotFrozen` is called by`FileRawSnapshotWriter#freeze` which may be called by `KafkaRaftClient ` or `SnapshotWriter`, if we want to do the optional validation, I assume a KIP is needed for tweaking `RawSnapshotWriter#freeze()` as sth like `RawSnapshotWriter#freeze(Origin origin)`, based on the origin, it can decide whether validation is needed. But since this is PR is preferable to be merged before #10786 , would it take too much time to go through the KIP and contain the change in this PR? Or may be we can make a seperate PR to handle the validation for `onSnapshotFrozen`?
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r640723618
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterReaderTest.java
##########
@@ -135,4 +141,18 @@ public static void assertSnapshot(List<List<String>> batches, SnapshotReader<Str
assertEquals(expected, actual);
}
+
+ private RaftClientTestContext initContextAsLeaderAndAdvanceHighWatermark(OffsetAndEpoch snapshotId) throws Exception {
Review comment:
Cool, updated :)
--
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] feyman2016 commented on pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#issuecomment-832764031
@jsancio Hi, could you help to take a look?
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619795184
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,20 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset <= snapshotId.offset) {
Review comment:
Conceptually, the `snapshotId.offset=highWatermarkOpt.get().offset` is ok, because the record at `snapshotId.offset` is not included in the snapshot, but I'm not sure if there are other restrictions because in the Jira description, it says: `The end offset and epoch of the snapshot is less than the high-watermark`, please kindly advice @jsancio
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r637470950
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -2268,6 +2269,25 @@ private Long append(int epoch, List<T> records, boolean isAtomic) {
);
}
+ private void validateSnapshotId(OffsetAndEpoch snapshotId) {
+ Optional<LogOffsetMetadata> highWatermarkOpt = quorum().highWatermark();
+ if (!highWatermarkOpt.isPresent() || highWatermarkOpt.get().offset < snapshotId.offset) {
+ throw new KafkaException("Trying to creating snapshot with invalid snapshotId: " + snapshotId + " whose offset is larger than the high-watermark: " +
+ highWatermarkOpt + ". This may necessarily mean a bug in the caller, since the there should be a minimum " +
+ "size of records between the latest snapshot and the high-watermark when creating snapshot");
Review comment:
I see. I would remove that last sentence because I don\t think this check enforces that and I don't think that it should.
##########
File path: raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java
##########
@@ -108,6 +108,17 @@
private final List<RaftResponse.Outbound> sentResponses = new ArrayList<>();
+ public static void advanceHighWatermark(RaftClientTestContext context,
Review comment:
How about making it a object/instance method by removing the static keyword so that the user can use it as follow:
```
context.advanceLeaderHighWatermarkToEndOffset()
```
Note that I renamed the method to include leader and removed all of the parameters. I could be wrong but `RaftClientTestContext` should have all of the information it needs to implement this. Also note, that if you want to generalize this then majority of the nodes need to send a `FETCH` request to the leader for the leader to advance the high-watermark.
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterReaderTest.java
##########
@@ -135,4 +141,18 @@ public static void assertSnapshot(List<List<String>> batches, SnapshotReader<Str
assertEquals(expected, actual);
}
+
+ private RaftClientTestContext initContextAsLeaderAndAdvanceHighWatermark(OffsetAndEpoch snapshotId) throws Exception {
Review comment:
Okay but note that if the quorum is a single node quorum then the high-watermark should advance simply by `context.client.poll()`. You can simplify this if you want by making the quorum a single member quorum. See the test in `KafkaRaftClientTest.testLeaderAppendSingleMemberQuorum()`.
--
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] feyman2016 commented on a change in pull request #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
feyman2016 commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619793844
##########
File path: raft/src/test/java/org/apache/kafka/snapshot/SnapshotWriterTest.java
##########
@@ -38,7 +38,7 @@
@Test
public void testWritingSnapshot() throws IOException {
- OffsetAndEpoch id = new OffsetAndEpoch(10L, 3);
+ OffsetAndEpoch id = new OffsetAndEpoch(0L, 1);
Review comment:
The highwatermark here is 1, so we need to make the snapshotId's endOffset < 1.
--
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 #10593: KAFKA-10800 Validate the snapshot id when the state machine creates a snapshot
Posted by GitBox <gi...@apache.org>.
jsancio commented on a change in pull request #10593:
URL: https://github.com/apache/kafka/pull/10593#discussion_r619882204
##########
File path: raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -1335,6 +1313,51 @@ public void testFetchSnapshotRequestClusterIdValidation() throws Exception {
context.assertSentFetchSnapshotResponse(Errors.INCONSISTENT_CLUSTER_ID);
}
+ @Test
+ public void testCreateSnapshotWithInvalidSnapshotId() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+ int epoch = 2;
+
+ List<String> appendRecords = Arrays.asList("a", "b", "c");
+ OffsetAndEpoch invalidSnapshotId1 = new OffsetAndEpoch(3, epoch);
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters)
+ .appendToLog(epoch, appendRecords)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int currentEpoch = context.currentEpoch();
+
+ // When creating snapshot:
+ // 1. high watermark cannot be empty
+ assertEquals(OptionalLong.empty(), context.client.highWatermark());
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId1));
+
+ // 2. high watermark must larger than the snapshotId's endOffset
+ advanceHighWatermark(context, currentEpoch, currentEpoch, otherNodeId, localId);
+ assertNotEquals(OptionalLong.empty(), context.client.highWatermark());
+ OffsetAndEpoch invalidSnapshotId2 = new OffsetAndEpoch(context.client.highWatermark().getAsLong(), currentEpoch);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId2));
+
+ // 3. the current leader epoch cache must larger than the snapshotId's epoch
+ OffsetAndEpoch invalidSnapshotId3 = new OffsetAndEpoch(context.client.highWatermark().getAsLong() - 1, currentEpoch + 1);
+ assertThrows(KafkaException.class, () -> context.client.createSnapshot(invalidSnapshotId3));
+ }
+
+ private void advanceHighWatermark(RaftClientTestContext context,
Review comment:
Thanks for cleaning up the code duplication.
--
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