diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala
index f20c232..643941d 100644
--- a/core/src/main/scala/kafka/log/Log.scala
+++ b/core/src/main/scala/kafka/log/Log.scala
@@ -76,6 +76,9 @@ class Log(val dir: File,
   newGauge(name + "-" + "NumLogSegments",
            new Gauge[Int] { def value = numberOfSegments })
 
+  newGauge(name + "-" + "LogStartOffset",
+           new Gauge[Long] { def value = logStartOffset })
+
   newGauge(name + "-" + "LogEndOffset",
            new Gauge[Long] { def value = logEndOffset })
            
@@ -449,6 +452,11 @@ class Log(val dir: File,
    */
   def size: Long = logSegments.map(_.size).sum
 
+   /**
+   * The earliest message offset in the log
+   */
+  def logStartOffset: Long = logSegments.map(_.baseOffset).min
+
   /**
    *  The offset of the next message that will be appended to the log
    */