Index: core/src/main/scala/kafka/log/LogSegment.scala
===================================================================
--- core/src/main/scala/kafka/log/LogSegment.scala	(revision 1404727)
+++ core/src/main/scala/kafka/log/LogSegment.scala	(working copy)
@@ -20,7 +20,11 @@
                  val indexIntervalBytes: Int,
                  time: Time) extends Range with Logging {
   
-  var firstAppendTime: Option[Long] = None
+  var firstAppendTime: Option[Long] =
+    if (messageSet.sizeInBytes > 0)
+      Some(time.milliseconds)
+    else
+      None
   
   /* the number of bytes since we last added an entry in the offset index */
   var bytesSinceLastIndexEntry = 0
@@ -118,6 +122,8 @@
       return
     index.truncateTo(offset)  
     messageSet.truncateTo(mapping.position)
+    if (messageSet.sizeInBytes == 0)
+      firstAppendTime = None
   }
   
   /**
Index: core/src/main/scala/kafka/log/Log.scala
===================================================================
--- core/src/main/scala/kafka/log/Log.scala	(revision 1404727)
+++ core/src/main/scala/kafka/log/Log.scala	(working copy)
@@ -393,6 +393,7 @@
           // If the last segment to be deleted is empty and we roll the log, the new segment will have the same
           // file name. So simply reuse the last segment and reset the modified time.
           view(numToDelete - 1).messageSet.file.setLastModified(time.milliseconds)
+          view(numToDelete - 1).firstAppendTime = None
           numToDelete -=1
         }
       }
@@ -415,11 +416,14 @@
    */
   private def maybeRoll(segment: LogSegment): LogSegment = {
     if ((segment.messageSet.sizeInBytes > maxLogFileSize) ||
-       ((segment.firstAppendTime.isDefined) && (time.milliseconds - segment.firstAppendTime.get > rollIntervalMs)) ||
+       ((segment.messageSet.sizeInBytes > 0) && (time.milliseconds - segment.firstAppendTime.get > rollIntervalMs)) ||
        segment.index.isFull)
       roll()
-    else
+    else {
+      if (segment.messageSet.sizeInBytes == 0)
+        segment.firstAppendTime = None
       segment
+    }
   }
 
   /**