jsancio commented on code in PR #17500:
URL: https://github.com/apache/kafka/pull/17500#discussion_r1806635797
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3426,6 +3428,12 @@ public Optional<SnapshotWriter<T>> createSnapshot(
throw new IllegalStateException("Cannot create snapshot before the
replica has been initialized");
}
+ AbstractIterator<? extends RecordBatch> batchIterator =
log.read(snapshotId.offset(), Isolation.COMMITTED).records.batchIterator();
+ if (batchIterator.hasNext() && batchIterator.peek().baseOffset() !=
snapshotId.offset()) {
+ logger.error("Cannot create snapshot at offset {} because it is
not batch aligned", snapshotId.offset());
Review Comment:
This is not an error in the KRaft layer. This is an error in the metadata
layer. You can log this at info since it should be rare. In the metadata layer
we can log an error for this case.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3426,6 +3428,12 @@ public Optional<SnapshotWriter<T>> createSnapshot(
throw new IllegalStateException("Cannot create snapshot before the
replica has been initialized");
}
+ AbstractIterator<? extends RecordBatch> batchIterator =
log.read(snapshotId.offset(), Isolation.COMMITTED).records.batchIterator();
+ if (batchIterator.hasNext() && batchIterator.peek().baseOffset() !=
snapshotId.offset()) {
+ logger.error("Cannot create snapshot at offset {} because it is
not batch aligned", snapshotId.offset());
+ return Optional.empty();
Review Comment:
Instead of returning a `Optional.empty()`. This method can throw an
`IllegalArgumentException` since this is a programming error in the metadata
layer.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java:
##########
@@ -1916,7 +1916,8 @@ public void
testCreateSnapshotAsLeaderWithInvalidSnapshotId(boolean withKip853Rp
// When leader creating snapshot:
// 1.1 high watermark cannot be empty
assertEquals(OptionalLong.empty(), context.client.highWatermark());
- assertThrows(IllegalArgumentException.class, () ->
context.client.createSnapshot(invalidSnapshotId1, 0));
+ IllegalArgumentException exception =
assertThrows(IllegalArgumentException.class, () ->
context.client.createSnapshot(invalidSnapshotId1, 0));
+ assertEquals("Cannot create a snapshot with an id
(OffsetAndEpoch(offset=4, epoch=2)) greater than the high-watermark (0)",
exception.getMessage());
Review Comment:
We use the following indentation in the raft module:
```java
IllegalArgumentException exception = assertThrows(
IllegalArgumentException.class,
() -> context.client.createSnapshot(invalidSnapshotId1, 0)
);
assertEquals(
String.format(
"Cannot create a snapshot with an id (%s) greater than the
high-watermark (%s)",
...
),
exception.getMessage()
);
```
This comment applies to a few places in this file.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
