rohitrmd commented on a change in pull request #10276:
URL: https://github.com/apache/kafka/pull/10276#discussion_r593859421
##########
File path: core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala
##########
@@ -340,6 +340,124 @@ final class KafkaMetadataLogTest {
batchBuilder.build()
}
+ @Test
+ def testValidateEpochGreaterThanLastKnownEpoch(): Unit = {
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ val numberOfRecords = 1
+ val epoch = 1
+
+ append(log, numberOfRecords, epoch)
+
+ val resultOffsetAndEpoch = log.validateOffsetAndEpoch(numberOfRecords,
epoch + 1)
+ assertEquals(ValidOffsetAndEpoch.Type.DIVERGING,
resultOffsetAndEpoch.`type`())
+ assertEquals(new OffsetAndEpoch(log.endOffset.offset, epoch),
resultOffsetAndEpoch.offsetAndEpoch())
+ }
+
+ @Test
+ def testValidateEpochLessThanOldestSnapshotEpoch(): Unit = {
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ val numberOfRecords = 10
+ val epoch = 1
+
+ append(log, numberOfRecords, epoch)
+ log.updateHighWatermark(new LogOffsetMetadata(numberOfRecords))
+
+ val snapshotId = new OffsetAndEpoch(numberOfRecords, epoch)
+ TestUtils.resource(log.createSnapshot(snapshotId)) { snapshot =>
+ snapshot.freeze()
+ }
+ assertTrue(log.deleteBeforeSnapshot(snapshotId))
+
+ val resultOffsetAndEpoch = log.validateOffsetAndEpoch(numberOfRecords,
epoch - 1)
+ assertEquals(ValidOffsetAndEpoch.Type.SNAPSHOT,
resultOffsetAndEpoch.`type`())
+ assertEquals(new OffsetAndEpoch(numberOfRecords, epoch),
resultOffsetAndEpoch.offsetAndEpoch())
+ }
+
+ @Test
+ def testValidateOffsetLessThanOldestSnapshotOffset(): Unit = {
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ val offset = 2
+ val epoch = 1
+
+ append(log, offset, epoch)
+ log.updateHighWatermark(new LogOffsetMetadata(offset))
+
+ val snapshotId = new OffsetAndEpoch(offset, epoch)
+ TestUtils.resource(log.createSnapshot(snapshotId)) { snapshot =>
+ snapshot.freeze()
+ }
+ assertTrue(log.deleteBeforeSnapshot(snapshotId))
+
+ val resultOffsetAndEpoch = log.validateOffsetAndEpoch(offset - 1, epoch)
+ assertEquals(ValidOffsetAndEpoch.Type.SNAPSHOT,
resultOffsetAndEpoch.`type`())
+ assertEquals(new OffsetAndEpoch(offset, epoch),
resultOffsetAndEpoch.offsetAndEpoch())
+ }
+
+ @Test
+ def testValidateOffsetEqualToOldestSnapshotOffset(): Unit = {
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ val offset = 2
+ val epoch = 1
+
+ append(log, offset, epoch)
+ log.updateHighWatermark(new LogOffsetMetadata(offset))
+
+ val snapshotId = new OffsetAndEpoch(offset, epoch)
+ TestUtils.resource(log.createSnapshot(snapshotId)) { snapshot =>
+ snapshot.freeze()
+ }
+ assertTrue(log.deleteBeforeSnapshot(snapshotId))
+
+ val resultOffsetAndEpoch = log.validateOffsetAndEpoch(offset, epoch)
+ assertEquals(ValidOffsetAndEpoch.Type.VALID, resultOffsetAndEpoch.`type`())
+ assertEquals(new OffsetAndEpoch(offset, epoch),
resultOffsetAndEpoch.offsetAndEpoch())
+ }
+
+ @Test
+ def testValidateEpochUnknown(): Unit = {
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ val numberOfRecords = 1
+ val epoch = 1
+
+ append(log, numberOfRecords, epoch)
+
+ val resultOffsetAndEpoch = log.validateOffsetAndEpoch(numberOfRecords,
epoch + 10)
+ assertEquals(ValidOffsetAndEpoch.Type.DIVERGING,
resultOffsetAndEpoch.`type`())
+ assertEquals(new OffsetAndEpoch(log.endOffset.offset, epoch),
resultOffsetAndEpoch.offsetAndEpoch())
+ }
+
+ @Test
+ def testValidateOffsetGreatThanEndOffset(): Unit = {
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ val numberOfRecords = 1
+ val epoch = 1
+
+ append(log, numberOfRecords, epoch)
+
+ val resultOffsetAndEpoch = log.validateOffsetAndEpoch(numberOfRecords + 1,
epoch)
+ assertEquals(ValidOffsetAndEpoch.Type.DIVERGING,
resultOffsetAndEpoch.`type`())
+ assertEquals(new OffsetAndEpoch(log.endOffset.offset, epoch),
resultOffsetAndEpoch.offsetAndEpoch())
+ }
+
+ @Test
+ def testValidateValidEpochAndOffset(): Unit = {
+ val log = buildMetadataLog(tempDir, mockTime)
+
+ val numberOfRecords = 1
+ val epoch = 1
+
+ append(log, numberOfRecords, epoch)
+
+ val resultOffsetAndEpoch = log.validateOffsetAndEpoch(numberOfRecords,
epoch)
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
