jsancio commented on a change in pull request #9816:
URL: https://github.com/apache/kafka/pull/9816#discussion_r557584847
##########
File path: core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
##########
@@ -29,15 +32,22 @@ import org.apache.kafka.snapshot.FileRawSnapshotReader
import org.apache.kafka.snapshot.FileRawSnapshotWriter
import org.apache.kafka.snapshot.RawSnapshotReader
import org.apache.kafka.snapshot.RawSnapshotWriter
+import org.apache.kafka.snapshot.Snapshots
import scala.compat.java8.OptionConverters._
-class KafkaMetadataLog(
+final class KafkaMetadataLog private (
log: Log,
+ snapshotIds: ConcurrentSkipListSet[raft.OffsetAndEpoch],
topicPartition: TopicPartition,
- maxFetchSizeInBytes: Int = 1024 * 1024
+ maxFetchSizeInBytes: Int
) extends ReplicatedLog {
+ private[this] var startSnapshotId = snapshotIds
Review comment:
Give me a sec to reply to this.
##########
File path: core/src/main/scala/kafka/raft/KafkaMetadataLog.scala
##########
@@ -79,13 +94,30 @@ class KafkaMetadataLog(
throw new IllegalArgumentException("Attempt to append an empty record
set")
val appendInfo = log.appendAsFollower(records.asInstanceOf[MemoryRecords])
+
+ if (appendInfo.rolled) {
+ log.deleteOldSegments()
+ }
+
new LogAppendInfo(appendInfo.firstOffset.getOrElse {
throw new KafkaException("Append failed unexpectedly")
}, appendInfo.lastOffset)
}
override def lastFetchedEpoch: Int = {
- log.latestEpoch.getOrElse(0)
+ log.latestEpoch.getOrElse {
+ latestSnapshotId.map { snapshotId =>
+ val logEndOffset = endOffset().offset
+ if (snapshotId.offset == logEndOffset) {
+ snapshotId.epoch
+ } else {
+ throw new KafkaException(
+ s"Log doesn't have a last fetch epoch and there is a snapshot
($snapshotId). " +
+ s"Expected the snapshot's end offset to match the logs end offset
($logEndOffset)"
+ )
+ }
+ } orElse(0)
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
