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:
us...@infra.apache.org
