package kafka.server
import org.apache.kafka.common.KafkaException
object LogOffsetMetadata {
val UnknownOffsetMetadata = new LogOffsetMetadata(-1, 0, 0)
val UnknownSegBaseOffset = -1L
val UnknownFilePosition = -1
class OffsetOrdering extends Ordering[LogOffsetMetadata] {
override def compare(x: LogOffsetMetadata , y: LogOffsetMetadata ): Int = {
return x.offsetDiff(y).toInt
}
}
}
case class LogOffsetMetadata(messageOffset: Long,
segmentBaseOffset: Long = LogOffsetMetadata.UnknownSegBaseOffset,
relativePositionInSegment: Int = LogOffsetMetadata.UnknownFilePosition) {
def offsetOnOlderSegment(that: LogOffsetMetadata): Boolean = {
if (messageOffsetOnly())
throw new KafkaException("%s cannot compare its segment info with %s since it only has message offset info".format(this, that))
this.segmentBaseOffset < that.segmentBaseOffset
}
def offsetOnSameSegment(that: LogOffsetMetadata): Boolean = {
if (messageOffsetOnly())
throw new KafkaException("%s cannot compare its segment info with %s since it only has message offset info".format(this, that))
this.segmentBaseOffset == that.segmentBaseOffset
}
def precedes(that: LogOffsetMetadata): Boolean = this.messageOffset < that.messageOffset
def offsetDiff(that: LogOffsetMetadata): Long = {
this.messageOffset - that.messageOffset
}
def positionDiff(that: LogOffsetMetadata): Int = {
if(!offsetOnSameSegment(that))
throw new KafkaException("%s cannot compare its segment position with %s since they are not on the same segment".format(this, that))
if(messageOffsetOnly())
throw new KafkaException("%s cannot compare its segment position with %s since it only has message offset info".format(this, that))
this.relativePositionInSegment - that.relativePositionInSegment
}
def messageOffsetOnly(): Boolean = {
segmentBaseOffset == LogOffsetMetadata.UnknownSegBaseOffset && relativePositionInSegment == LogOffsetMetadata.UnknownFilePosition
}
override def toString = messageOffset.toString + " [" + segmentBaseOffset + " : " + relativePositionInSegment + "]"
}