jsancio commented on a change in pull request #9816:
URL: https://github.com/apache/kafka/pull/9816#discussion_r564245683
##########
File path:
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientSnapshotTest.java
##########
@@ -43,7 +44,59 @@
final public class KafkaRaftClientSnapshotTest {
@Test
- public void testMissingFetchSnapshotRequest() throws Exception {
+ public void testFetchRequest() throws Exception {
+ int localId = 0;
+ int otherNodeId = localId + 1;
+ Set<Integer> voters = Utils.mkSet(localId, otherNodeId);
+
+ RaftClientTestContext context = new
RaftClientTestContext.Builder(localId, voters)
+ .withAppendLingerMs(1)
+ .build();
+
+ context.becomeLeader();
+ int epoch = context.currentEpoch();
+
+ String[] appendRecords = new String[] {"a", "b", "c"};
+ context.client.scheduleAppend(epoch, Arrays.asList(appendRecords));
+ context.time.sleep(context.appendLingerMs());
+ context.client.poll();
+
+ long localLogEndOffset = context.log.endOffset().offset;
+ assertTrue(
+ appendRecords.length <= localLogEndOffset,
+ String.format("Record length = %s, log end offset = %s",
appendRecords.length, localLogEndOffset)
+ );
+
+ // Advance the highWatermark
+ context.deliverRequest(context.fetchRequest(epoch, otherNodeId,
localLogEndOffset, epoch, 0));
+ context.pollUntilResponse();
+ context.assertSentFetchResponse(Errors.NONE, epoch,
OptionalInt.of(localId));
+ assertEquals(localLogEndOffset,
context.client.highWatermark().getAsLong());
+
+ OffsetAndEpoch snapshotId = new OffsetAndEpoch(localLogEndOffset,
epoch);
+ try (SnapshotWriter<String> snapshot =
context.client.createSnapshot(snapshotId)) {
+ snapshot.freeze();
+ }
+
+ context.client.poll();
+
+ assertEquals(snapshotId.offset, context.log.startOffset());
+
+ // Send Fetch request less than start offset
+ context.deliverRequest(context.fetchRequest(epoch, otherNodeId, 0,
epoch, 0));
+ context.pollUntilResponse();
+ FetchResponseData.FetchablePartitionResponse partitionResponse =
context.assertSentFetchResponse();
+ assertEquals(Errors.NONE,
Errors.forCode(partitionResponse.errorCode()));
+ assertEquals(epoch, partitionResponse.currentLeader().leaderEpoch());
+ assertEquals(localId, partitionResponse.currentLeader().leaderId());
+ assertEquals(snapshotId.epoch, partitionResponse.snapshotId().epoch());
+ assertEquals(snapshotId.offset,
partitionResponse.snapshotId().endOffset());
+ }
+
+ // TODO: Add a few more fetch request tests checking some of the error
conditions and edge conditions
Review comment:
Yep. Replaced the TODO with a few more tests.
----------------------------------------------------------------
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
