kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nehanarkh...@apache.org
Subject svn commit: r1396713 [2/2] - in /incubator/kafka/branches/0.8: bin/ config/ core/src/main/scala/kafka/admin/ core/src/main/scala/kafka/api/ core/src/main/scala/kafka/cluster/ core/src/main/scala/kafka/common/ core/src/main/scala/kafka/consumer/ core/sr...
Date Wed, 10 Oct 2012 18:03:14 GMT
Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala Wed Oct 10 18:03:07 2012
@@ -27,9 +27,15 @@ import org.I0Itec.zkclient.IZkChildListe
 /**
  * This class represents the state machine for replicas. It defines the states that a replica can be in, and
  * transitions to move the replica to another legal state. The different states that a replica can be in are -
- * 1. OnlineReplica     : Once a replica is started, it is in this state. Valid previous state are OnlineReplica or
- *                        OfflineReplica
- * 2. OfflineReplica    : If a replica dies, it moves to this state. Valid previous state is OnlineReplica
+ * 1. NewReplica        : The controller can create new replicas during partition reassignment. In this state, a
+ *                        replica can only get become follower state change request.  Valid previous
+ *                        state is NonExistentReplica
+ * 2. OnlineReplica     : Once a replica is started and part of the assigned replicas for its partition, it is in this
+ *                        state. In this state, it can get either become leader or become follower state change requests.
+ *                        Valid previous state are NewReplica, OnlineReplica or OfflineReplica
+ * 3. OfflineReplica    : If a replica dies, it moves to this state. This happens when the broker hosting the replica
+ *                        is down. Valid previous state are NewReplica, OnlineReplica
+ * 4. NonExistentReplica: If a replica is deleted, it is moved to this state. Valid previous state is OfflineReplica
  */
 class ReplicaStateMachine(controller: KafkaController) extends Logging {
   this.logIdent = "[Replica state machine on Controller " + controller.config.brokerId + "]: "
@@ -49,7 +55,8 @@ class ReplicaStateMachine(controller: Ka
     // initialize replica state
     initializeReplicaState()
     // move all Online replicas to Online
-    handleStateChanges(controllerContext.liveBrokerIds.toSeq, OnlineReplica)
+    handleStateChanges(ZkUtils.getAllReplicasOnBroker(zkClient, controllerContext.allTopics.toSeq,
+      controllerContext.liveBrokerIds.toSeq), OnlineReplica)
     info("Started replica state machine with initial state -> " + replicaState.toString())
   }
 
@@ -72,20 +79,11 @@ class ReplicaStateMachine(controller: Ka
    * @param targetState  The state that the replicas should be moved to
    * The controller's allLeaders cache should have been updated before this
    */
-  def handleStateChanges(brokerIds: Seq[Int], targetState: ReplicaState) {
-    info("Invoking state change to %s for brokers %s".format(targetState, brokerIds.mkString(",")))
+  def handleStateChanges(replicas: Seq[PartitionAndReplica], targetState: ReplicaState) {
+    info("Invoking state change to %s for replicas %s".format(targetState, replicas.mkString(",")))
     try {
       brokerRequestBatch.newBatch()
-      brokerIds.foreach { brokerId =>
-        // read all the partitions and their assigned replicas into a map organized by
-        // { replica id -> partition 1, partition 2...
-        val partitionsAssignedToThisBroker = getPartitionsAssignedToBroker(controllerContext.allTopics.toSeq, brokerId)
-        partitionsAssignedToThisBroker.foreach { topicAndPartition =>
-          handleStateChange(topicAndPartition._1, topicAndPartition._2, brokerId, targetState)
-        }
-        if(partitionsAssignedToThisBroker.size == 0)
-          info("No state transitions triggered since no partitions are assigned to brokers %s".format(brokerIds.mkString(",")))
-      }
+      replicas.foreach(r => handleStateChange(r.topic, r.partition, r.replica, targetState))
       brokerRequestBatch.sendRequestsToBrokers()
     }catch {
       case e => error("Error while moving some replicas to %s state".format(targetState), e)
@@ -100,28 +98,62 @@ class ReplicaStateMachine(controller: Ka
    * @param replicaId   The replica for which the state transition is invoked
    * @param targetState The end state that the replica should be moved to
    */
-  private def handleStateChange(topic: String, partition: Int, replicaId: Int, targetState: ReplicaState) {
+  def handleStateChange(topic: String, partition: Int, replicaId: Int, targetState: ReplicaState) {
     try {
+      replicaState.getOrElseUpdate((topic, partition, replicaId), NonExistentReplica)
       targetState match {
-        case OnlineReplica =>
-          assertValidPreviousStates(topic, partition, replicaId, List(OnlineReplica, OfflineReplica), targetState)
-          // check if the leader for this partition is alive or even exists
-          // NOTE: technically, we could get the leader from the allLeaders cache, but we need to read zookeeper
-          // for the ISR anyways
+        case NewReplica =>
+          assertValidPreviousStates(topic, partition, replicaId, List(NonExistentReplica), targetState)
+          // start replica as a follower to the current leader for its partition
           val leaderAndIsrOpt = ZkUtils.getLeaderAndIsrForPartition(zkClient, topic, partition)
           leaderAndIsrOpt match {
             case Some(leaderAndIsr) =>
-              controllerContext.liveBrokerIds.contains(leaderAndIsr.leader) match {
-                case true => // leader is alive
-                  brokerRequestBatch.addRequestForBrokers(List(replicaId), topic, partition, leaderAndIsr)
-                  replicaState.put((topic, partition, replicaId), OnlineReplica)
-                  info("Replica %d for partition [%s, %d] state changed to OnlineReplica".format(replicaId, topic, partition))
-                case false => // ignore partitions whose leader is not alive
+              if(leaderAndIsr.leader == replicaId)
+                throw new StateChangeFailedException("Replica %d for partition [%s, %d] cannot be moved to NewReplica"
+                  .format(replicaId, topic, partition) + "state as it is being requested to become leader")
+              brokerRequestBatch.addLeaderAndIsrRequestForBrokers(List(replicaId), topic, partition, leaderAndIsr)
+            case None => // new leader request will be sent to this replica when one gets elected
+          }
+          replicaState.put((topic, partition, replicaId), NewReplica)
+          info("Replica %d for partition [%s, %d] state changed to NewReplica".format(replicaId, topic, partition))
+        case NonExistentReplica =>
+          assertValidPreviousStates(topic, partition, replicaId, List(OfflineReplica), targetState)
+          // send stop replica command
+          brokerRequestBatch.addStopReplicaRequestForBrokers(List(replicaId), topic, partition)
+          // remove this replica from the assigned replicas list for its partition
+          val currentAssignedReplicas = controllerContext.partitionReplicaAssignment((topic, partition))
+          controllerContext.partitionReplicaAssignment.put((topic, partition),
+            currentAssignedReplicas.filterNot(_ == replicaId))
+          info("Replica %d for partition [%s, %d] state changed to NonExistentReplica".format(replicaId, topic, partition))
+          replicaState.remove((topic, partition, replicaId))
+        case OnlineReplica =>
+          assertValidPreviousStates(topic, partition, replicaId, List(NewReplica, OnlineReplica, OfflineReplica), targetState)
+          replicaState((topic, partition, replicaId)) match {
+            case NewReplica =>
+              // add this replica to the assigned replicas list for its partition
+              val currentAssignedReplicas = controllerContext.partitionReplicaAssignment((topic, partition))
+              controllerContext.partitionReplicaAssignment.put((topic, partition), currentAssignedReplicas :+ replicaId)
+              info("Replica %d for partition [%s, %d] state changed to OnlineReplica".format(replicaId, topic, partition))
+            case _ =>
+              // check if the leader for this partition is alive or even exists
+              // NOTE: technically, we could get the leader from the allLeaders cache, but we need to read zookeeper
+              // for the ISR anyways
+              val leaderAndIsrOpt = ZkUtils.getLeaderAndIsrForPartition(zkClient, topic, partition)
+              leaderAndIsrOpt match {
+                case Some(leaderAndIsr) =>
+                  controllerContext.liveBrokerIds.contains(leaderAndIsr.leader) match {
+                    case true => // leader is alive
+                      brokerRequestBatch.addLeaderAndIsrRequestForBrokers(List(replicaId), topic, partition, leaderAndIsr)
+                      replicaState.put((topic, partition, replicaId), OnlineReplica)
+                      info("Replica %d for partition [%s, %d] state changed to OnlineReplica".format(replicaId, topic, partition))
+                    case false => // ignore partitions whose leader is not alive
+                  }
+                case None => // ignore partitions who don't have a leader yet
               }
-            case None => // ignore partitions who don't have a leader yet
           }
+          replicaState.put((topic, partition, replicaId), OnlineReplica)
         case OfflineReplica =>
-          assertValidPreviousStates(topic, partition, replicaId, List(OnlineReplica), targetState)
+          assertValidPreviousStates(topic, partition, replicaId, List(NewReplica, OnlineReplica), targetState)
           // As an optimization, the controller removes dead replicas from the ISR
           var zookeeperPathUpdateSucceeded: Boolean = false
           var newLeaderAndIsr: LeaderAndIsr = null
@@ -144,7 +176,7 @@ class ReplicaStateMachine(controller: Ka
             }
           }
           // send the shrunk ISR state change request only to the leader
-          brokerRequestBatch.addRequestForBrokers(List(newLeaderAndIsr.leader), topic, partition, newLeaderAndIsr)
+          brokerRequestBatch.addLeaderAndIsrRequestForBrokers(List(newLeaderAndIsr.leader), topic, partition, newLeaderAndIsr)
           // update the local leader and isr cache
           controllerContext.allLeaders.put((topic, partition), newLeaderAndIsr.leader)
           replicaState.put((topic, partition, replicaId), OfflineReplica)
@@ -226,7 +258,9 @@ class ReplicaStateMachine(controller: Ka
 }
 
 sealed trait ReplicaState { def state: Byte }
-case object OnlineReplica extends ReplicaState { val state: Byte = 1 }
-case object OfflineReplica extends ReplicaState { val state: Byte = 2 }
+case object NewReplica extends ReplicaState { val state: Byte = 1 }
+case object OnlineReplica extends ReplicaState { val state: Byte = 2 }
+case object OfflineReplica extends ReplicaState { val state: Byte = 3 }
+case object NonExistentReplica extends ReplicaState { val state: Byte = 4 }
 
 

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/FileMessageSet.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/FileMessageSet.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/FileMessageSet.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/FileMessageSet.scala Wed Oct 10 18:03:07 2012
@@ -31,35 +31,45 @@ import kafka.metrics.{KafkaTimer, KafkaM
 /**
  * An on-disk message set. The set can be opened either mutably or immutably. Mutation attempts
  * will fail on an immutable message set. An optional limit and start position can be applied to the message set
- * which will control the position in the file at which the set begins.
+ * which will control the position in the file at which the set begins
  */
 @nonthreadsafe
 class FileMessageSet private[kafka](val file: File,
                                     private[log] val channel: FileChannel,
-                                    private[log] val start: Long = 0L,
-                                    private[log] val limit: Long = Long.MaxValue) extends MessageSet with Logging {
+                                    private[log] val start: Long, // the starting position in the file
+                                    private[log] val limit: Long, // the length (may be less than the file length)
+                                    val mutable: Boolean) extends MessageSet with Logging {
   
-  /* the size of the message set in bytes */
-  private val _size = new AtomicLong(scala.math.min(channel.size(), limit) - start)
-    
-  /* set the file position to the last byte in the file */
-  channel.position(channel.size)
+  private val setSize = new AtomicLong()
+
+  if(mutable) {
+    if(limit < Long.MaxValue || start > 0)
+      throw new KafkaException("Attempt to open a mutable message set with a view or offset, which is not allowed.")
+
+    setSize.set(channel.size())
+    channel.position(channel.size)
+  } else {
+    setSize.set(scala.math.min(channel.size(), limit) - start)
+  }
   
   /**
    * Create a file message set with no limit or offset
    */
-  def this(file: File, channel: FileChannel) = this(file, channel, 0, Long.MaxValue)
+  def this(file: File, channel: FileChannel, mutable: Boolean) = 
+    this(file, channel, 0, Long.MaxValue, mutable)
   
   /**
    * Create a file message set with no limit or offset
    */
-  def this(file: File) = this(file, Utils.openChannel(file, mutable = true))
+  def this(file: File, mutable: Boolean) = 
+    this(file, Utils.openChannel(file, mutable), mutable)
   
   /**
    * Return a message set which is a view into this set starting from the given position and with the given size limit.
    */
   def read(position: Long, size: Long): FileMessageSet = {
-    new FileMessageSet(file, channel, this.start + position, scala.math.min(this.start + position + size, sizeInBytes()))
+    new FileMessageSet(file, channel, this.start + position, scala.math.min(this.start + position + size, sizeInBytes()),
+      false)
   }
   
   /**
@@ -69,7 +79,7 @@ class FileMessageSet private[kafka](val 
   private[log] def searchFor(targetOffset: Long, startingPosition: Int): OffsetPosition = {
     var position = startingPosition
     val buffer = ByteBuffer.allocate(12)
-    val size = _size.get()
+    val size = setSize.get()
     while(position + 12 < size) {
       buffer.rewind()
       channel.read(buffer, position)
@@ -128,22 +138,29 @@ class FileMessageSet private[kafka](val 
   /**
    * The number of bytes taken up by this file set
    */
-  def sizeInBytes(): Long = _size.get()
+  def sizeInBytes(): Long = setSize.get()
+
+  def checkMutable(): Unit = {
+    if(!mutable)
+      throw new KafkaException("Attempt to invoke mutation on immutable message set.")
+  }
   
   /**
    * Append this message to the message set
    */
   def append(messages: MessageSet): Unit = {
+    checkMutable()
     var written = 0L
     while(written < messages.sizeInBytes)
       written += messages.writeTo(channel, 0, messages.sizeInBytes)
-    _size.getAndAdd(written)
+    setSize.getAndAdd(written)
   }
  
   /**
    * Commit all written data to the physical disk
    */
   def flush() = {
+    checkMutable()
     LogFlushStats.logFlushTimer.time {
       channel.force(true)
     }
@@ -153,7 +170,8 @@ class FileMessageSet private[kafka](val 
    * Close this message set
    */
   def close() {
-    flush()
+    if(mutable)
+      flush()
     channel.close()
   }
   
@@ -170,12 +188,13 @@ class FileMessageSet private[kafka](val 
    * given size falls on a valid byte offset.
    */
   def truncateTo(targetSize: Long) = {
+    checkMutable()
     if(targetSize > sizeInBytes())
       throw new KafkaException("Attempt to truncate log segment to %d bytes failed since the current ".format(targetSize) +
         " size of this log segment is only %d bytes".format(sizeInBytes()))
     channel.truncate(targetSize)
     channel.position(targetSize)
-    _size.set(targetSize)
+    setSize.set(targetSize)
   }
   
 }

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/Log.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/Log.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/Log.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/Log.scala Wed Oct 10 18:03:07 2012
@@ -101,8 +101,8 @@ object Log {
 private[kafka] class Log(val dir: File, 
                          val maxLogFileSize: Long, 
                          val maxMessageSize: Int, 
-                         val flushInterval: Int = Int.MaxValue,
-                         val rollIntervalMs: Long = Long.MaxValue, 
+                         val flushInterval: Int,
+                         val rollIntervalMs: Long, 
                          val needsRecovery: Boolean, 
                          val maxIndexSize: Int = (10*1024*1024),
                          val indexIntervalBytes: Int = 4096,
@@ -128,10 +128,10 @@ private[kafka] class Log(val dir: File, 
   private var nextOffset: AtomicLong = new AtomicLong(segments.view.last.nextOffset())
 
   newGauge(name + "-" + "NumLogSegments",
-           new Gauge[Int] { def getValue = numberOfSegments })
+           new Gauge[Int] { def value() = numberOfSegments })
 
   newGauge(name + "-" + "LogEndOffset",
-           new Gauge[Long] { def getValue = logEndOffset })
+           new Gauge[Long] { def value() = logEndOffset })
 
   /* The name of this log */
   def name  = dir.getName()
@@ -151,7 +151,8 @@ private[kafka] class Log(val dir: File, 
         if(!Log.indexFilename(dir, start).exists)
           throw new IllegalStateException("Found log file with no corresponding index file.")
         logSegments.add(new LogSegment(dir = dir, 
-                                       startOffset = start,
+                                       startOffset = start, 
+                                       mutable = false, 
                                        indexIntervalBytes = indexIntervalBytes, 
                                        maxIndexSize = maxIndexSize))
       }
@@ -160,7 +161,8 @@ private[kafka] class Log(val dir: File, 
     if(logSegments.size == 0) {
       // no existing segments, create a new mutable segment
       logSegments.add(new LogSegment(dir = dir, 
-                                     startOffset = 0,
+                                     startOffset = 0, 
+                                     mutable = true, 
                                      indexIntervalBytes = indexIntervalBytes, 
                                      maxIndexSize = maxIndexSize))
     } else {
@@ -174,9 +176,17 @@ private[kafka] class Log(val dir: File, 
         }
       })
 
-      // run recovery on the last segment if necessary
+      //make the final section mutable and run recovery on it if necessary
+      val last = logSegments.remove(logSegments.size - 1)
+      last.close()
+      val mutableSegment = new LogSegment(dir = dir, 
+                                          startOffset = last.start, 
+                                          mutable = true, 
+                                          indexIntervalBytes = indexIntervalBytes, 
+                                          maxIndexSize = maxIndexSize)
       if(needsRecovery)
-        recoverSegment(logSegments.get(logSegments.size - 1))
+        recoverSegment(mutableSegment)
+      logSegments.add(mutableSegment)
     }
     new SegmentList(logSegments.toArray(new Array[LogSegment](logSegments.size)))
   }
@@ -396,11 +406,12 @@ private[kafka] class Log(val dir: File, 
     }
     debug("Rolling log '" + name + "' to " + logFile.getName + " and " + indexFile.getName)
     segments.view.lastOption match {
-      case Some(segment) => segment.index.trimToSize()
+      case Some(segment) => segment.index.makeReadOnly()
       case None => 
     }
     val segment = new LogSegment(dir, 
                                  startOffset = newOffset,
+                                 mutable = true, 
                                  indexIntervalBytes = indexIntervalBytes, 
                                  maxIndexSize = maxIndexSize)
     segments.append(segment)
@@ -535,7 +546,8 @@ private[kafka] class Log(val dir: File, 
       val deletedSegments = segments.trunc(segments.view.size)
       debug("Truncate and start log '" + name + "' to " + newOffset)
       segments.append(new LogSegment(dir, 
-                                     newOffset,
+                                     newOffset, 
+                                     mutable = true, 
                                      indexIntervalBytes = indexIntervalBytes, 
                                      maxIndexSize = maxIndexSize))
       deleteSegments(deletedSegments)

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/LogSegment.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/LogSegment.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/LogSegment.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/LogSegment.scala Wed Oct 10 18:03:07 2012
@@ -28,9 +28,9 @@ class LogSegment(val messageSet: FileMes
   
   @volatile var deleted = false
   
-  def this(dir: File, startOffset: Long, indexIntervalBytes: Int, maxIndexSize: Int) = 
-    this(new FileMessageSet(file = Log.logFilename(dir, startOffset)), 
-         new OffsetIndex(file = Log.indexFilename(dir, startOffset), baseOffset = startOffset, maxIndexSize = maxIndexSize),
+  def this(dir: File, startOffset: Long, mutable: Boolean, indexIntervalBytes: Int, maxIndexSize: Int) = 
+    this(new FileMessageSet(file = Log.logFilename(dir, startOffset), mutable = mutable), 
+         new OffsetIndex(file = Log.indexFilename(dir, startOffset), baseOffset = startOffset, mutable = mutable, maxIndexSize = maxIndexSize),
          startOffset,
          indexIntervalBytes,
          SystemTime)

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/OffsetIndex.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/OffsetIndex.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/OffsetIndex.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/log/OffsetIndex.scala Wed Oct 10 18:03:07 2012
@@ -49,36 +49,39 @@ import kafka.utils._
  * All external APIs translate from relative offsets to full offsets, so users of this class do not interact with the internal 
  * storage format.
  */
-class OffsetIndex(val file: File, val baseOffset: Long, maxIndexSize: Int = -1) extends Logging {
-  
+class OffsetIndex(val file: File, val baseOffset: Long, var mutable: Boolean, maxIndexSize: Int = -1) extends Logging {
+
   /* the memory mapping */
   private var mmap: MappedByteBuffer = 
     {
       val newlyCreated = file.createNewFile()
       val raf = new RandomAccessFile(file, "rw")
       try {
-        /* pre-allocate the file if necessary */
-        if(newlyCreated) {
+        if(mutable) {
+          /* if mutable create and memory map a new sparse file */
           if(maxIndexSize < 8)
             throw new IllegalArgumentException("Invalid max index size: " + maxIndexSize)
-          raf.setLength(roundToExactMultiple(maxIndexSize, 8))
-        }
           
-        val len = raf.length()  
-        if(len < 0 || len % 8 != 0)
-          throw new IllegalStateException("Index file " + file.getName + " is corrupt, found " + len + 
-                                          " bytes which is not positive or not a multiple of 8.")
+          /* pre-allocate the file if necessary */
+          if(newlyCreated)
+            raf.setLength(roundToExactMultiple(maxIndexSize, 8))
+          val idx = raf.getChannel.map(FileChannel.MapMode.READ_WRITE, 0, raf.length())
           
-        /* memory-map the file */
-        val idx = raf.getChannel.map(FileChannel.MapMode.READ_WRITE, 0, len)
-          
-        /* set the position in the index for the next entry */
-        if(newlyCreated)
-          idx.position(0)
-        else
-          // if this is a pre-existing index, assume it is all valid and set position to last entry
-          idx.position(roundToExactMultiple(idx.limit, 8))
-        idx
+          /* set the position in the index for the next entry */
+          if(newlyCreated)
+            idx.position(0)
+          else
+            // if this is a pre-existing index, assume it is all valid and set position to last entry
+            idx.position(roundToExactMultiple(idx.limit, 8))
+          idx
+        } else {
+          /* if not mutable, just mmap what they gave us */
+          val len = raf.length()
+          if(len < 0 || len % 8 != 0)
+            throw new IllegalStateException("Index file " + file.getName + " is corrupt, found " + len + 
+                                            " bytes which is not positive or not a multiple of 8.")
+          raf.getChannel.map(FileChannel.MapMode.READ_ONLY, 0, len)
+        }
       } finally {
         Utils.swallow(raf.close())
       }
@@ -88,7 +91,7 @@ class OffsetIndex(val file: File, val ba
   val maxEntries = mmap.limit / 8
   
   /* the number of entries in the index */
-  private var size = new AtomicInteger(mmap.position / 8)
+  private var size = if(mutable) new AtomicInteger(mmap.position / 8) else new AtomicInteger(mmap.limit / 8)
   
   /* the last offset in the index */
   var lastOffset = readLastOffset()
@@ -112,6 +115,8 @@ class OffsetIndex(val file: File, val ba
    * the pair (baseOffset, 0) is returned.
    */
   def lookup(targetOffset: Long): OffsetPosition = {
+    if(entries == 0)
+      return OffsetPosition(baseOffset, 0)
     val idx = mmap.duplicate
     val slot = indexSlotFor(idx, targetOffset)
     if(slot == -1)
@@ -123,20 +128,16 @@ class OffsetIndex(val file: File, val ba
   /**
    * Find the slot in which the largest offset less than or equal to the given
    * target offset is stored.
-   * Return -1 if the least entry in the index is larger than the target offset or the index is empty
+   * Return -1 if the least entry in the index is larger than the target offset 
    */
   private def indexSlotFor(idx: ByteBuffer, targetOffset: Long): Int = {
     // we only store the difference from the baseoffset so calculate that
     val relativeOffset = targetOffset - baseOffset
     
-    // check if the index is empty
-    if(entries == 0)
-      return -1
-    
     // check if the target offset is smaller than the least offset
     if(logical(idx, 0) > relativeOffset)
       return -1
-      
+    
     // binary search for the entry
     var lo = 0
     var hi = entries-1
@@ -174,6 +175,8 @@ class OffsetIndex(val file: File, val ba
    */
   def append(logicalOffset: Long, position: Int) {
     this synchronized {
+      if(!mutable)
+        throw new IllegalStateException("Attempt to append to an immutable offset index " + file.getName)
       if(isFull)
         throw new IllegalStateException("Attempt to append to a full index (size = " + size + ").")
       if(size.get > 0 && logicalOffset <= lastOffset)
@@ -224,17 +227,17 @@ class OffsetIndex(val file: File, val ba
   }
   
   /**
-   * Trim this segment to fit just the valid entries, deleting all trailing unwritten bytes from
-   * the file.
+   * Make this segment read-only, flush any unsaved changes, and truncate any excess bytes
    */
-  def trimToSize() {
+  def makeReadOnly() {
     this synchronized {
+      mutable = false
       flush()
       val raf = new RandomAccessFile(file, "rws")
       try {
         val newLength = entries * 8
         raf.setLength(newLength)
-        this.mmap = raf.getChannel().map(FileChannel.MapMode.READ_WRITE, 0, newLength)
+        this.mmap = raf.getChannel().map(FileChannel.MapMode.READ_ONLY, 0, newLength)
       } finally {
         Utils.swallow(raf.close())
       }
@@ -262,7 +265,8 @@ class OffsetIndex(val file: File, val ba
   
   /** Close the index */
   def close() {
-    trimToSize()
+    if(mutable)
+      makeReadOnly()
   }
   
   /**

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/metrics/KafkaCSVMetricsReporter.scala Wed Oct 10 18:03:07 2012
@@ -50,10 +50,9 @@ private class KafkaCSVMetricsReporter ex
         if (!csvDir.exists())
           csvDir.mkdirs()
         underlying = new CsvReporter(Metrics.defaultRegistry(), csvDir)
-        if (props.getBoolean("kafka.csv.metrics.reporter.enabled", default = false)) {
-          initialized = true
+        if (props.getBoolean("kafka.csv.metrics.reporter.enabled", false))
           startReporter(metricsConfig.pollingIntervalSecs)
-        }
+        initialized = true
       }
     }
   }

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/network/RequestChannel.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/network/RequestChannel.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/network/RequestChannel.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/network/RequestChannel.scala Wed Oct 10 18:03:07 2012
@@ -92,7 +92,7 @@ class RequestChannel(val numProcessors: 
   newGauge(
     "RequestQueueSize",
     new Gauge[Int] {
-      def getValue = requestQueue.size
+      def value() = requestQueue.size
     }
   )
 

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/ProducerPool.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/ProducerPool.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/ProducerPool.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/ProducerPool.scala Wed Oct 10 18:03:07 2012
@@ -21,7 +21,7 @@ import kafka.cluster.Broker
 import java.util.Properties
 import collection.mutable.HashMap
 import java.lang.Object
-import kafka.utils.{Utils, Logging}
+import kafka.utils.Logging
 import kafka.api.TopicMetadata
 import kafka.common.UnavailableProducerException
 

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/producer/async/ProducerSendThread.scala Wed Oct 10 18:03:07 2012
@@ -36,7 +36,7 @@ class ProducerSendThread[K,V](val thread
   newGauge(
     "ProducerQueueSize-" + getId,
     new Gauge[Int] {
-      def getValue = queue.size
+      def value() = queue.size
     }
   )
 

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherManager.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherManager.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherManager.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherManager.scala Wed Oct 10 18:03:07 2012
@@ -23,7 +23,7 @@ import kafka.cluster.Broker
 
 abstract class AbstractFetcherManager(protected val name: String, numFetchers: Int = 1) extends Logging {
     // map of (source brokerid, fetcher Id per source broker) => fetcher
-  private val fetcherThreadMap = new mutable.HashMap[(Broker, Int), AbstractFetcherThread]
+  private val fetcherThreadMap = new mutable.HashMap[(Int, Int), AbstractFetcherThread]
   private val mapLock = new Object
   this.logIdent = "[" + name + "], "
 
@@ -37,7 +37,7 @@ abstract class AbstractFetcherManager(pr
   def addFetcher(topic: String, partitionId: Int, initialOffset: Long, sourceBroker: Broker) {
     mapLock synchronized {
       var fetcherThread: AbstractFetcherThread = null
-      val key = (sourceBroker, getFetcherId(topic, partitionId))
+      val key = (sourceBroker.id, getFetcherId(topic, partitionId))
       fetcherThreadMap.get(key) match {
         case Some(f) => fetcherThread = f
         case None =>
@@ -64,6 +64,15 @@ abstract class AbstractFetcherManager(pr
     }
   }
 
+  def fetcherSourceBroker(topic: String, partitionId: Int): Option[Int] = {
+    mapLock synchronized {
+      for ( ((sourceBrokerId, _), fetcher) <- fetcherThreadMap)
+        if (fetcher.hasPartition(topic, partitionId))
+          return Some(sourceBrokerId)
+    }
+    None
+  }
+
   def closeAllFetchers() {
     mapLock synchronized {
       for ( (_, fetcher) <- fetcherThreadMap) {
@@ -72,4 +81,4 @@ abstract class AbstractFetcherManager(pr
       fetcherThreadMap.clear()
     }
   }
-}
+}
\ No newline at end of file

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherThread.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherThread.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/AbstractFetcherThread.scala Wed Oct 10 18:03:07 2012
@@ -119,7 +119,7 @@ abstract class  AbstractFetcherThread(na
                   warn("current offset %d for topic %s partition %d out of range; reset offset to %d"
                     .format(currentOffset.get, topic, partitionId, newOffset))
                 case _ =>
-                  error("error for %s %d to broker %d".format(topic, partitionId, sourceBroker.id),
+                  error("error for %s %d to broker %s ".format(topic, partitionId, sourceBroker.host),
                     ErrorMapping.exceptionFor(partitionData.error))
                   partitionsWithError += topicAndPartition
                   fetchMap.remove(topicAndPartition)
@@ -165,7 +165,7 @@ class FetcherLagMetrics(name: (String, I
   newGauge(
     name._1 + "-" + name._2 + "-ConsumerLag",
     new Gauge[Long] {
-      def getValue = lagVal.get
+      def value() = lagVal.get
     }
   )
 

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaApis.scala Wed Oct 10 18:03:07 2012
@@ -17,7 +17,6 @@
 
 package kafka.server
 
-import java.io.IOException
 import kafka.admin.{CreateTopicCommand, AdminUtils}
 import kafka.api._
 import kafka.message._
@@ -76,7 +75,6 @@ class KafkaApis(val requestChannel: Requ
     requestChannel.sendResponse(new Response(request, new BoundedByteBufferSend(leaderAndISRResponse)))
   }
 
-
   def handleStopReplicaRequest(request: RequestChannel.Request){
     val stopReplicaRequest = request.requestObj.asInstanceOf[StopReplicaRequest]
     if(requestLogger.isTraceEnabled)
@@ -85,7 +83,7 @@ class KafkaApis(val requestChannel: Requ
 
     val responseMap = new HashMap[(String, Int), Short]
 
-    for((topic, partitionId) <- stopReplicaRequest.stopReplicaSet){
+    for((topic, partitionId) <- stopReplicaRequest.partitions){
       val errorCode = replicaManager.stopReplica(topic, partitionId)
       responseMap.put((topic, partitionId), errorCode)
     }

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaConfig.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaConfig.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaConfig.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/KafkaConfig.scala Wed Oct 10 18:03:07 2012
@@ -21,7 +21,7 @@ import java.util.Properties
 import kafka.message.Message
 import kafka.consumer.ConsumerConfig
 import java.net.InetAddress
-import kafka.utils.{Topic, Utils, VerifiableProperties, ZKConfig}
+import kafka.utils.{Utils, VerifiableProperties, ZKConfig}
 
 /**
  * Configuration settings for the kafka server

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaFetcherManager.scala Wed Oct 10 18:03:07 2012
@@ -20,7 +20,8 @@ package kafka.server
 import kafka.cluster.Broker
 
 class ReplicaFetcherManager(private val brokerConfig: KafkaConfig, private val replicaMgr: ReplicaManager)
-        extends AbstractFetcherManager("ReplicaFetcherManager on broker " + brokerConfig.brokerId + ", ", brokerConfig.numReplicaFetchers) {
+        extends AbstractFetcherManager("ReplicaFetcherManager on broker " + brokerConfig.brokerId + ", ",
+          brokerConfig.numReplicaFetchers) {
 
   override def createFetcherThread(fetcherId: Int, sourceBroker: Broker): AbstractFetcherThread = {
     new ReplicaFetcherThread("ReplicaFetcherThread-%d-%d-on-broker-%d".format(sourceBroker.id, fetcherId, brokerConfig.brokerId), sourceBroker, brokerConfig, replicaMgr)

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaManager.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaManager.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaManager.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/ReplicaManager.scala Wed Oct 10 18:03:07 2012
@@ -47,13 +47,13 @@ class ReplicaManager(val config: KafkaCo
   newGauge(
     "LeaderCount",
     new Gauge[Int] {
-      def getValue = leaderPartitions.size
+      def value() = leaderPartitions.size
     }
   )
   newGauge(
     "UnderReplicatedPartitions",
     new Gauge[Int] {
-      def getValue = {
+      def value() = {
         leaderPartitionsLock synchronized {
           leaderPartitions.count(_.isUnderReplicated)
         }
@@ -159,19 +159,6 @@ class ReplicaManager(val config: KafkaCo
       }
       responseMap.put(partitionInfo, errorCode)
     }
-
-    /**
-     *  If IsInit flag is on, this means that the controller wants to treat topics not in the request
-     *  as deleted.
-     *  TODO: Handle this properly as part of KAFKA-330
-     */
-//    if(leaderAndISRRequest.isInit == LeaderAndIsrRequest.IsInit){
-//      startHighWaterMarksCheckPointThread
-//      val partitionsToRemove = allPartitions.filter(p => !leaderAndISRRequest.leaderAndISRInfos.contains(p._1)).map(entry => entry._1)
-//      info("Init flag is set in leaderAndISR request, partitions to remove: %s".format(partitionsToRemove))
-//      partitionsToRemove.foreach(p => stopReplica(p._1, p._2))
-//    }
-
     responseMap
   }
 

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/RequestPurgatory.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/RequestPurgatory.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/RequestPurgatory.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/server/RequestPurgatory.scala Wed Oct 10 18:03:07 2012
@@ -69,7 +69,7 @@ abstract class RequestPurgatory[T <: Del
   newGauge(
     "NumDelayedRequests",
     new Gauge[Int] {
-      def getValue = expiredRequestReaper.unsatisfied.get()
+      def value() = expiredRequestReaper.unsatisfied.get()
     }
   )
 

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/ConsumerOffsetChecker.scala Wed Oct 10 18:03:07 2012
@@ -22,7 +22,6 @@ import joptsimple._
 import org.I0Itec.zkclient.ZkClient
 import kafka.utils.{ZkUtils, ZKStringSerializer, Logging}
 import kafka.consumer.SimpleConsumer
-import collection.mutable.Map
 import kafka.api.{PartitionOffsetRequestInfo, OffsetRequest}
 import kafka.common.TopicAndPartition
 import scala.collection._
@@ -34,9 +33,8 @@ object ConsumerOffsetChecker extends Log
 
   private val BidPidPattern = """(\d+)-(\d+)""".r
 
-  private val BrokerIpPattern = """.*:([^:]+):(\d+$)""".r
+  private val BrokerIpPattern = """.*:(\d+\.\d+\.\d+\.\d+):(\d+$)""".r
   // e.g., 127.0.0.1-1315436360737:127.0.0.1:9092
-  // e.g., host.domain.com-1315436360737:host.domain.com:9092
 
   private def getConsumer(zkClient: ZkClient, bid: String): Option[SimpleConsumer] = {
     val brokerInfo = ZkUtils.readDataMaybeNull(zkClient, "/brokers/ids/%s".format(bid))._1

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/DumpLogSegments.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/DumpLogSegments.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/DumpLogSegments.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/DumpLogSegments.scala Wed Oct 10 18:03:07 2012
@@ -43,7 +43,7 @@ object DumpLogSegments {
   /* print out the contents of the index */
   def dumpIndex(file: File) {
     val startOffset = file.getName().split("\\.")(0).toLong
-    val index = new OffsetIndex(file = file, baseOffset = startOffset)
+    val index = new OffsetIndex(file = file, baseOffset = startOffset, mutable = false)
     for(i <- 0 until index.entries) {
       val entry = index.entry(i)
       // since it is a sparse file, in the event of a crash there may be many zero entries, stop if we see one
@@ -57,7 +57,7 @@ object DumpLogSegments {
   def dumpLog(file: File, printContents: Boolean) {
     val startOffset = file.getName().split("\\.")(0).toLong
     println("Starting offset: " + startOffset)
-    val messageSet = new FileMessageSet(file)
+    val messageSet = new FileMessageSet(file, false)
     var validBytes = 0L
     for(messageAndOffset <- messageSet) {
       val msg = messageAndOffset.message

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/Utils.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/Utils.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/Utils.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/Utils.scala Wed Oct 10 18:03:07 2012
@@ -19,6 +19,7 @@ package kafka.utils
 
 import java.io._
 import java.nio._
+import charset.Charset
 import java.nio.channels._
 import java.lang.management._
 import java.util.zip.CRC32
@@ -638,7 +639,7 @@ object Utils extends Logging {
     builder.toString
   }
 
-  def mapToJson[T <: Any](map: Map[String, List[String]]): String = {
+  def mapToJson[T <: Any](map: Map[String, Seq[String]]): String = {
     val builder = new StringBuilder
     builder.append("{ ")
     var numElements = 0
@@ -715,6 +716,18 @@ object Utils extends Logging {
       for (forever <- Stream.continually(1); t <- coll) yield t
     stream.iterator
   }
+
+  def readFileIntoString(path: String): String = {
+    val stream = new FileInputStream(new File(path))
+    try {
+      val fc = stream.getChannel()
+      val bb = fc.map(FileChannel.MapMode.READ_ONLY, 0, fc.size())
+      Charset.defaultCharset().decode(bb).toString()
+    }
+    finally {
+      stream.close()
+    }
+  }
 }
 
 /**

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/ZkUtils.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/ZkUtils.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/ZkUtils.scala (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/utils/ZkUtils.scala Wed Oct 10 18:03:07 2012
@@ -27,12 +27,15 @@ import kafka.api.LeaderAndIsr
 import org.apache.zookeeper.data.Stat
 import java.util.concurrent.locks.{ReentrantLock, Condition}
 import kafka.common.{KafkaException, NoEpochForPartitionException}
+import kafka.controller.{PartitionAndReplica, ReassignedPartitionsContext}
+import kafka.admin._
 
 object ZkUtils extends Logging {
   val ConsumersPath = "/consumers"
   val BrokerIdsPath = "/brokers/ids"
   val BrokerTopicsPath = "/brokers/topics"
   val ControllerPath = "/controller"
+  val ReassignPartitionsPath = "/admin/reassign_partitions"
 
   def getTopicPath(topic: String): String ={
     BrokerTopicsPath + "/" + topic
@@ -76,20 +79,23 @@ object ZkUtils extends Logging {
     val leaderAndIsrOpt = leaderAndIsrInfo._1
     val stat = leaderAndIsrInfo._2
     leaderAndIsrOpt match {
-      case Some(leaderAndIsrStr) =>
-        SyncJSON.parseFull(leaderAndIsrStr) match {
-          case Some(m) =>
-            val leader = m.asInstanceOf[Map[String, String]].get("leader").get.toInt
-            val epoch = m.asInstanceOf[Map[String, String]].get("leaderEpoch").get.toInt
-            val isrString = m.asInstanceOf[Map[String, String]].get("ISR").get
-            val isr = Utils.getCSVList(isrString).map(r => r.toInt)
-            val zkPathVersion = stat.getVersion
-            debug("Leader %d, Epoch %d, Isr %s, Zk path version %d for topic %s and partition %d".format(leader, epoch,
-              isr.toString(), zkPathVersion, topic, partition))
-            Some(LeaderAndIsr(leader, epoch, isr.toList, zkPathVersion))
-          case None => None
-        }
-      case None => None // TODO: Handle if leader and isr info is not available in zookeeper
+      case Some(leaderAndIsrStr) => parseLeaderAndIsr(leaderAndIsrStr, topic, partition, stat)
+      case None => None
+    }
+  }
+
+  def parseLeaderAndIsr(leaderAndIsrStr: String, topic: String, partition: Int, stat: Stat): Option[LeaderAndIsr] = {
+    SyncJSON.parseFull(leaderAndIsrStr) match {
+      case Some(m) =>
+        val leader = m.asInstanceOf[Map[String, String]].get("leader").get.toInt
+        val epoch = m.asInstanceOf[Map[String, String]].get("leaderEpoch").get.toInt
+        val isrString = m.asInstanceOf[Map[String, String]].get("ISR").get
+        val isr = Utils.getCSVList(isrString).map(r => r.toInt)
+        val zkPathVersion = stat.getVersion
+        debug("Leader %d, Epoch %d, Isr %s, Zk path version %d for topic %s and partition %d".format(leader, epoch,
+          isr.toString(), zkPathVersion, topic, partition))
+        Some(LeaderAndIsr(leader, epoch, isr.toList, zkPathVersion))
+      case None => None
     }
   }
 
@@ -295,11 +301,13 @@ object ZkUtils extends Logging {
   def conditionalUpdatePersistentPath(client: ZkClient, path: String, data: String, expectVersion: Int): (Boolean, Int) = {
     try {
       val stat = client.writeData(path, data, expectVersion)
-      debug("Conditional update to the zookeeper path %s with expected version %d succeeded and returned the new version: %d".format(path, expectVersion, stat.getVersion))
+      info("Conditional update of zkPath %s with value %s and expected version %d succeeded, returning the new version: %d"
+        .format(path, data, expectVersion, stat.getVersion))
       (true, stat.getVersion)
     } catch {
       case e: Exception =>
-        debug("Conditional update to the zookeeper path %s with expected version %d failed".format(path, expectVersion), e)
+        error("Conditional update of zkPath %s with data %s and expected version %d failed".format(path, data,
+          expectVersion), e)
         (false, -1)
     }
   }
@@ -511,6 +519,66 @@ object ZkUtils extends Logging {
     }.flatten[(String, Int)].toSeq
   }
 
+  def getPartitionsBeingReassigned(zkClient: ZkClient): Map[(String, Int), ReassignedPartitionsContext] = {
+    // read the partitions and their new replica list
+    val jsonPartitionMapOpt = readDataMaybeNull(zkClient, ReassignPartitionsPath)._1
+    jsonPartitionMapOpt match {
+      case Some(jsonPartitionMap) =>
+        val reassignedPartitions = parsePartitionReassignmentData(jsonPartitionMap)
+        reassignedPartitions.map { p =>
+          val newReplicas = p._2
+          (p._1 -> new ReassignedPartitionsContext(newReplicas))
+        }
+      case None => Map.empty[(String, Int), ReassignedPartitionsContext]
+    }
+  }
+
+  def parsePartitionReassignmentData(jsonData: String):Map[(String, Int), Seq[Int]] = {
+    SyncJSON.parseFull(jsonData) match {
+      case Some(m) =>
+        val replicaMap = m.asInstanceOf[Map[String, Seq[String]]]
+        replicaMap.map { reassignedPartitions =>
+          val topic = reassignedPartitions._1.split(",").head
+          val partition = reassignedPartitions._1.split(",").last.toInt
+          val newReplicas = reassignedPartitions._2.map(_.toInt)
+          (topic, partition) -> newReplicas
+        }
+      case None => Map.empty[(String, Int), Seq[Int]]
+    }
+  }
+
+  def updatePartitionReassignmentData(zkClient: ZkClient, partitionsToBeReassigned: Map[(String, Int), Seq[Int]]) {
+    val zkPath = ZkUtils.ReassignPartitionsPath
+    partitionsToBeReassigned.size match {
+      case 0 => // need to delete the /admin/reassign_partitions path
+        deletePath(zkClient, zkPath)
+        info("No more partitions need to be reassigned. Deleting zk path %s".format(zkPath))
+      case _ =>
+        val jsonData = Utils.mapToJson(partitionsToBeReassigned.map(p => ("%s,%s".format(p._1._1, p._1._2)) -> p._2.map(_.toString)))
+        try {
+          updatePersistentPath(zkClient, zkPath, jsonData)
+          info("Updated partition reassignment path with %s".format(jsonData))
+        }catch {
+          case nne: ZkNoNodeException =>
+            ZkUtils.createPersistentPath(zkClient, zkPath, jsonData)
+            debug("Created path %s with %s for partition reassignment".format(zkPath, jsonData))
+          case e2 => throw new AdministrationException(e2.toString)
+        }
+    }
+  }
+
+  def getAllReplicasOnBroker(zkClient: ZkClient, topics: Seq[String], brokerIds: Seq[Int]): Seq[PartitionAndReplica] = {
+    brokerIds.map { brokerId =>
+      // read all the partitions and their assigned replicas into a map organized by
+      // { replica id -> partition 1, partition 2...
+      val partitionsAssignedToThisBroker = getPartitionsAssignedToBroker(zkClient, topics, brokerId)
+      if(partitionsAssignedToThisBroker.size == 0)
+        info("No state transitions triggered since no partitions are assigned to brokers %s".format(brokerIds.mkString(",")))
+      partitionsAssignedToThisBroker.map(p => new PartitionAndReplica(p._1, p._2, brokerId))
+    }.flatten
+  }
+
+
   def deletePartition(zkClient : ZkClient, brokerId: Int, topic: String) {
     val brokerIdPath = BrokerIdsPath + "/" + brokerId
     zkClient.delete(brokerIdPath)

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/admin/AdminTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/admin/AdminTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/admin/AdminTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/admin/AdminTest.scala Wed Oct 10 18:03:07 2012
@@ -20,8 +20,10 @@ import junit.framework.Assert._
 import org.junit.Test
 import org.scalatest.junit.JUnit3Suite
 import kafka.zk.ZooKeeperTestHarness
+import kafka.server.{KafkaServer, KafkaConfig}
+import collection.mutable.ListBuffer
 import kafka.common.ErrorMapping
-import kafka.utils.TestUtils
+import kafka.utils.{Utils, ZkUtils, TestUtils}
 
 class AdminTest extends JUnit3Suite with ZooKeeperTestHarness {
 
@@ -205,4 +207,182 @@ class AdminTest extends JUnit3Suite with
         }
     }
   }
+
+  @Test
+  def testPartitionReassignmentWithLeaderInNewReplicas() {
+    val expectedReplicaAssignment = Map(0  -> List("0", "1", "2"))
+    val topic = "test"
+    // create brokers
+    val servers = TestUtils.createBrokerConfigs(4).map(b => TestUtils.createServer(new KafkaConfig(b)))
+    // create the topic
+    AdminUtils.createTopicPartitionAssignmentPathInZK(topic, expectedReplicaAssignment, zkClient)
+    // reassign partition 0
+    val newReplicas = Seq(0, 2, 3)
+    val partitionToBeReassigned = 0
+    val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, Map((topic, partitionToBeReassigned) -> newReplicas))
+    assertTrue("Partition reassignment attempt failed for [test, 0]", reassignPartitionsCommand.reassignPartitions())
+    // wait until reassignment is completed
+    TestUtils.waitUntilTrue(() => {
+      val partitionsBeingReassigned = ZkUtils.getPartitionsBeingReassigned(zkClient).mapValues(_.newReplicas);
+      CheckReassignmentStatus.checkIfPartitionReassignmentSucceeded(zkClient, topic, partitionToBeReassigned, newReplicas,
+      Map((topic, partitionToBeReassigned) -> newReplicas), partitionsBeingReassigned) == ReassignmentCompleted;
+    }, 1000)
+    val assignedReplicas = ZkUtils.getReplicasForPartition(zkClient, topic, partitionToBeReassigned)
+    assertEquals("Partition should have been reassigned to 0, 2, 3", newReplicas, assignedReplicas)
+    servers.foreach(_.shutdown())
+  }
+
+  @Test
+  def testPartitionReassignmentWithLeaderNotInNewReplicas() {
+    val expectedReplicaAssignment = Map(0  -> List("0", "1", "2"))
+    val topic = "test"
+    // create brokers
+    val servers = TestUtils.createBrokerConfigs(4).map(b => TestUtils.createServer(new KafkaConfig(b)))
+    // create the topic
+    AdminUtils.createTopicPartitionAssignmentPathInZK(topic, expectedReplicaAssignment, zkClient)
+    // reassign partition 0
+    val newReplicas = Seq(1, 2, 3)
+    val partitionToBeReassigned = 0
+    val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, Map((topic, partitionToBeReassigned) -> newReplicas))
+    assertTrue("Partition reassignment failed for test, 0", reassignPartitionsCommand.reassignPartitions())
+    // wait until reassignment is completed
+    TestUtils.waitUntilTrue(() => {
+      val partitionsBeingReassigned = ZkUtils.getPartitionsBeingReassigned(zkClient).mapValues(_.newReplicas);
+      CheckReassignmentStatus.checkIfPartitionReassignmentSucceeded(zkClient, topic, partitionToBeReassigned, newReplicas,
+      Map((topic, partitionToBeReassigned) -> newReplicas), partitionsBeingReassigned) == ReassignmentCompleted;
+    }, 1000)
+    val assignedReplicas = ZkUtils.getReplicasForPartition(zkClient, topic, partitionToBeReassigned)
+    assertEquals("Partition should have been reassigned to 0, 2, 3", newReplicas, assignedReplicas)
+    // leader should be 2
+    servers.foreach(_.shutdown())
+  }
+
+  @Test
+  def testPartitionReassignmentNonOverlappingReplicas() {
+    val expectedReplicaAssignment = Map(0  -> List("0", "1"))
+    val topic = "test"
+    // create brokers
+    val servers = TestUtils.createBrokerConfigs(4).map(b => TestUtils.createServer(new KafkaConfig(b)))
+    // create the topic
+    AdminUtils.createTopicPartitionAssignmentPathInZK(topic, expectedReplicaAssignment, zkClient)
+    // reassign partition 0
+    val newReplicas = Seq(2, 3)
+    val partitionToBeReassigned = 0
+    val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, Map((topic, partitionToBeReassigned) -> newReplicas))
+    assertTrue("Partition reassignment failed for test, 0", reassignPartitionsCommand.reassignPartitions())
+    // wait until reassignment is completed
+    TestUtils.waitUntilTrue(() => {
+      val partitionsBeingReassigned = ZkUtils.getPartitionsBeingReassigned(zkClient).mapValues(_.newReplicas);
+      CheckReassignmentStatus.checkIfPartitionReassignmentSucceeded(zkClient, topic, partitionToBeReassigned, newReplicas,
+      Map((topic, partitionToBeReassigned) -> newReplicas), partitionsBeingReassigned) == ReassignmentCompleted;
+    }, 1000)
+    val assignedReplicas = ZkUtils.getReplicasForPartition(zkClient, topic, partitionToBeReassigned)
+    assertEquals("Partition should have been reassigned to 2, 3", newReplicas, assignedReplicas)
+    // leader should be 2
+    servers.foreach(_.shutdown())
+  }
+
+  @Test
+  def testReassigningNonExistingPartition() {
+    val topic = "test"
+    // create brokers
+    val servers = TestUtils.createBrokerConfigs(4).map(b => TestUtils.createServer(new KafkaConfig(b)))
+    // reassign partition 0
+    val newReplicas = Seq(2, 3)
+    val partitionToBeReassigned = 0
+    val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, Map((topic, partitionToBeReassigned) -> newReplicas))
+    assertTrue("Partition reassignment failed for test, 0", reassignPartitionsCommand.reassignPartitions())
+    val reassignedPartitions = ZkUtils.getPartitionsBeingReassigned(zkClient)
+    assertFalse("Partition should not be reassigned", reassignedPartitions.contains((topic, partitionToBeReassigned)))
+    // leader should be 2
+    servers.foreach(_.shutdown())
+  }
+
+  @Test
+  def testResumePartitionReassignmentThatWasCompleted() {
+    val expectedReplicaAssignment = Map(0  -> List("0", "1"))
+    val topic = "test"
+    // create the topic
+    AdminUtils.createTopicPartitionAssignmentPathInZK(topic, expectedReplicaAssignment, zkClient)
+    // put the partition in the reassigned path as well
+    // reassign partition 0
+    val newReplicas = Seq(0, 1)
+    val partitionToBeReassigned = 0
+    val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, Map((topic, partitionToBeReassigned) -> newReplicas))
+    reassignPartitionsCommand.reassignPartitions
+    // create brokers
+    val servers = TestUtils.createBrokerConfigs(2).map(b => TestUtils.createServer(new KafkaConfig(b)))
+    TestUtils.waitUntilTrue(checkIfReassignPartitionPathExists, 1000)
+    val assignedReplicas = ZkUtils.getReplicasForPartition(zkClient, topic, partitionToBeReassigned)
+    assertEquals("Partition should have been reassigned to 0, 1", newReplicas, assignedReplicas)
+    servers.foreach(_.shutdown())
+  }
+
+  @Test
+  def testResumePartitionReassignmentAfterLeaderWasMoved() {
+    var expectedReplicaAssignment = Map(0  -> List(1, 0, 2, 3))
+    val leaderForPartitionMap = Map(0 -> 2)
+    val topic = "test"
+    val serverConfigs = TestUtils.createBrokerConfigs(4).map(b => new KafkaConfig(b))
+    val servers = new ListBuffer[KafkaServer]
+    // create the topic
+    AdminUtils.createTopicPartitionAssignmentPathInZK(topic,
+      expectedReplicaAssignment.map(r => (r._1) -> r._2.map(_.toString)), zkClient)
+    // bring up just brokers 0 and 1
+    servers.append(TestUtils.createServer(serverConfigs(0)))
+    servers.append(TestUtils.createServer(serverConfigs(1)))
+    val newReplicas = Seq(2, 3)
+    val partitionToBeReassigned = 0
+    val reassignPartitionsCommand = new ReassignPartitionsCommand(zkClient, Map((topic, partitionToBeReassigned) -> newReplicas))
+    reassignPartitionsCommand.reassignPartitions
+    // this partition reassignment request should be ignored since replicas 2 and 3 are not online
+    // and the admin path should be deleted as well
+    TestUtils.waitUntilTrue(checkIfReassignmentPathIsDeleted, 1000)
+    var assignedReplicas = ZkUtils.getReplicasForPartition(zkClient, topic, partitionToBeReassigned)
+    assertEquals("Partition should not be reassigned to 2, 3 yet", expectedReplicaAssignment(0), assignedReplicas)
+    // create brokers
+    servers.append(TestUtils.createServer(serverConfigs(2)))
+    servers.append(TestUtils.createServer(serverConfigs(3)))
+    // wait until new replicas catch up with leader
+    TestUtils.waitUntilTrue(checkIfNewReplicasInIsr, 2000)
+    // change the assigned replicas to 0 and 1
+    updateAssignedReplicasForPartition("test", 0, List(0, 1))
+    // reissue the partition reassignment
+    reassignPartitionsCommand.reassignPartitions
+    // create leaders for the partition to be reassigned
+    TestUtils.makeLeaderForPartition(zkClient, topic, leaderForPartitionMap)
+    // bounce controller
+    servers.head.shutdown()
+    servers.head.startup()
+    TestUtils.waitUntilTrue(checkIfReassignPartitionPathExists, 1500)
+    assignedReplicas = ZkUtils.getReplicasForPartition(zkClient, topic, partitionToBeReassigned)
+    assertEquals("Partition should have been reassigned to 2, 3", newReplicas, assignedReplicas)
+    servers.foreach(_.shutdown())
+  }
+
+  private def checkIfReassignPartitionPathExists(): Boolean = {
+    ZkUtils.pathExists(zkClient, ZkUtils.ReassignPartitionsPath)
+  }
+
+  private def checkIfReassignmentPathIsDeleted(): Boolean = {
+    !ZkUtils.pathExists(zkClient, ZkUtils.ReassignPartitionsPath)
+  }
+
+  private def checkIfNewReplicasInIsr(): Boolean = {
+    val leaderAndIsrOpt = ZkUtils.getLeaderAndIsrForPartition(zkClient, "test", 0)
+    leaderAndIsrOpt match {
+      case Some(leaderAndIsr) =>
+        if(leaderAndIsr.isr.contains(2) && leaderAndIsr.isr.contains(3))
+          true
+        else
+          false
+      case None => false
+    }
+  }
+
+  private def updateAssignedReplicasForPartition(topic: String, partition: Int, newAssignedReplicas: Seq[Int]) {
+    val zkPath = ZkUtils.getTopicPath(topic)
+    val jsonPartitionMap = Utils.mapToJson(Map(partition.toString -> newAssignedReplicas.map(_.toString)))
+    ZkUtils.updatePersistentPath(zkClient, zkPath, jsonPartitionMap)
+  }
 }

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/api/RequestResponseSerializationTest.scala Wed Oct 10 18:03:07 2012
@@ -95,7 +95,7 @@ object SerializationTestUtils{
   }
 
   def createTestStopReplicaRequest() : StopReplicaRequest = {
-    new StopReplicaRequest(Set((topic1, 0), (topic2, 0)))
+    new StopReplicaRequest(collection.immutable.Set((topic1, 0), (topic2, 0)))
   }
 
   def createTestStopReplicaResponse() : StopReplicaResponse = {

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/FileMessageSetTest.scala Wed Oct 10 18:03:07 2012
@@ -29,7 +29,7 @@ class FileMessageSetTest extends BaseMes
   val messageSet = createMessageSet(messages)
   
   def createMessageSet(messages: Seq[Message]): FileMessageSet = {
-    val set = new FileMessageSet(tempFile())
+    val set = new FileMessageSet(tempFile(), true)
     set.append(new ByteBufferMessageSet(NoCompressionCodec, messages: _*))
     set.flush()
     set

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogOffsetTest.scala Wed Oct 10 18:03:07 2012
@@ -30,7 +30,7 @@ import org.scalatest.junit.JUnit3Suite
 import kafka.admin.CreateTopicCommand
 import kafka.api.{PartitionOffsetRequestInfo, FetchRequestBuilder, OffsetRequest}
 import kafka.utils.TestUtils._
-import kafka.common.{ErrorMapping, TopicAndPartition, UnknownTopicOrPartitionException}
+import kafka.common.{ErrorMapping, TopicAndPartition}
 
 object LogOffsetTest {
   val random = new Random()  

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogSegmentTest.scala Wed Oct 10 18:03:07 2012
@@ -15,10 +15,10 @@ class LogSegmentTest extends JUnit3Suite
   
   def createSegment(offset: Long): LogSegment = {
     val msFile = TestUtils.tempFile()
-    val ms = new FileMessageSet(msFile)
+    val ms = new FileMessageSet(msFile, true)
     val idxFile = TestUtils.tempFile()
     idxFile.delete()
-    val idx = new OffsetIndex(idxFile, offset, 100)
+    val idx = new OffsetIndex(idxFile, offset, true, 100)
     val seg = new LogSegment(ms, idx, offset, 10, SystemTime)
     segments += seg
     seg

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/LogTest.scala Wed Oct 10 18:03:07 2012
@@ -342,33 +342,6 @@ class LogTest extends JUnitSuite {
     assertEquals("Should change log size", log.size, 0)
   }
 
-  @Test
-  def testReopenThenTruncate() {
-    val set = TestUtils.singleMessageSet("test".getBytes())
-
-    // create a log
-    var log = new Log(logDir, 
-                      maxLogFileSize = set.sizeInBytes * 5, 
-                      maxMessageSize = config.maxMessageSize, 
-                      maxIndexSize = 1000, 
-                      indexIntervalBytes = 10000, 
-                      needsRecovery = true)
-    
-    // add enough messages to roll over several segments then close and re-open and attempt to truncate
-    for(i <- 0 until 100)
-      log.append(set)
-    log.close()
-    log = new Log(logDir, 
-                  maxLogFileSize = set.sizeInBytes * 5, 
-                  maxMessageSize = config.maxMessageSize, 
-                  maxIndexSize = 1000, 
-                  indexIntervalBytes = 10000, 
-                  needsRecovery = true)
-    log.truncateTo(3)
-    assertEquals("All but one segment should be deleted.", 1, log.numberOfSegments)
-    assertEquals("Log end offset should be 3.", 3, log.logEndOffset)
-  }
-  
   def assertContains(ranges: Array[Range], offset: Long) = {
     Log.findRange(ranges, offset) match {
       case Some(range) => 

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/OffsetIndexTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/OffsetIndexTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/OffsetIndexTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/log/OffsetIndexTest.scala Wed Oct 10 18:03:07 2012
@@ -33,7 +33,7 @@ class OffsetIndexTest extends JUnitSuite
   
   @Before
   def setup() {
-    this.idx = new OffsetIndex(file = nonExistantTempFile(), baseOffset = 45L, maxIndexSize = 30 * 8)
+    this.idx = new OffsetIndex(file = nonExistantTempFile(), baseOffset = 45L, mutable = true, maxIndexSize = 30 * 8)
   }
   
   @After
@@ -41,7 +41,7 @@ class OffsetIndexTest extends JUnitSuite
     if(this.idx != null)
       this.idx.file.delete()
   }
-  
+
   @Test
   def randomLookupTest() {
     assertEquals("Not present value should return physical offset 0.", OffsetPosition(idx.baseOffset, 0), idx.lookup(92L))
@@ -88,6 +88,25 @@ class OffsetIndexTest extends JUnitSuite
     }
     assertWriteFails("Append should fail on a full index", idx, idx.maxEntries + 1, classOf[IllegalStateException])
   }
+
+  
+  @Test
+  def testReadOnly() {
+    /* add some random values */
+    val vals = List((49, 1), (52, 2), (55, 3))
+    for((logical, physical) <- vals)
+      idx.append(logical, physical)
+    
+    idx.makeReadOnly()
+    
+    assertEquals("File length should just contain added entries.", vals.size * 8L, idx.file.length())
+    assertEquals("Last offset field should be initialized", vals.last._1, idx.lastOffset)
+    
+    for((logical, physical) <- vals)
+    	assertEquals("Should still be able to find everything.", OffsetPosition(logical, physical), idx.lookup(logical))
+    	
+    assertWriteFails("Append should fail on read-only index", idx, 60, classOf[IllegalStateException])
+  }
   
   @Test(expected = classOf[IllegalArgumentException])
   def appendOutOfOrder() {
@@ -96,13 +115,13 @@ class OffsetIndexTest extends JUnitSuite
   }
   
   @Test
-  def testReopen() {
+  def reopenAsReadonly() {
     val first = OffsetPosition(51, 0)
     val sec = OffsetPosition(52, 1)
     idx.append(first.offset, first.position)
     idx.append(sec.offset, sec.position)
     idx.close()
-    val idxRo = new OffsetIndex(file = idx.file, baseOffset = idx.baseOffset)
+    val idxRo = new OffsetIndex(file = idx.file, baseOffset = idx.baseOffset, mutable = false)
     assertEquals(first, idxRo.lookup(first.offset))
     assertEquals(sec, idxRo.lookup(sec.offset))
     assertWriteFails("Append should fail on read-only index", idxRo, 53, classOf[IllegalStateException])
@@ -110,8 +129,7 @@ class OffsetIndexTest extends JUnitSuite
   
   @Test
   def truncate() {
-	val idx = new OffsetIndex(file = nonExistantTempFile(), baseOffset = 0L, maxIndexSize = 10 * 8)
-	idx.truncate()
+	val idx = new OffsetIndex(file = nonExistantTempFile(), baseOffset = 0L, mutable = true, maxIndexSize = 10 * 8)
     for(i <- 1 until 10)
       idx.append(i, i)
       
@@ -137,6 +155,13 @@ class OffsetIndexTest extends JUnitSuite
       case e: Exception => assertEquals("Got an unexpected exception.", klass, e.getClass)
     }
   }
+  
+  def makeIndex(baseOffset: Long, mutable: Boolean, vals: Seq[(Long, Int)]): OffsetIndex = {
+    val idx = new OffsetIndex(file = nonExistantTempFile(), baseOffset = baseOffset, mutable = mutable, maxIndexSize = 2 * vals.size * 8)
+    for ((logical, physical) <- vals)
+      idx.append(logical, physical)
+    idx
+  }
 
   def monotonicSeq(base: Int, len: Int): Seq[Int] = {
     val rand = new Random(1L)

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/BaseMessageSetTestCases.scala Wed Oct 10 18:03:07 2012
@@ -67,7 +67,7 @@ trait BaseMessageSetTestCases extends JU
       val channel = new RandomAccessFile(file, "rw").getChannel()
       val written = set.writeTo(channel, 0, 1024)
       assertEquals("Expect to write the number of bytes in the set.", set.sizeInBytes, written)
-      val newSet = new FileMessageSet(file, channel)
+      val newSet = new FileMessageSet(file, channel, false)
       checkEquals(set.iterator, newSet.iterator)
     }
   }

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/MessageCompressionTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/MessageCompressionTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/MessageCompressionTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/message/MessageCompressionTest.scala Wed Oct 10 18:03:07 2012
@@ -60,7 +60,6 @@ class MessageCompressionTest extends JUn
       true
     } catch {
       case e: UnsatisfiedLinkError => false
-      case e: org.xerial.snappy.SnappyError => false
     }
   }
 }

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/metrics/KafkaTimerTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/metrics/KafkaTimerTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/metrics/KafkaTimerTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/metrics/KafkaTimerTest.scala Wed Oct 10 18:03:07 2012
@@ -35,20 +35,20 @@ class KafkaTimerTest extends JUnit3Suite
     timer.time {
       clock.addMillis(1000)
     }
-    assertEquals(1, metric.getCount())
-    assertTrue((metric.getMax() - 1000).abs <= Double.Epsilon)
-    assertTrue((metric.getMin() - 1000).abs <= Double.Epsilon)
+    assertEquals(1, metric.count())
+    assertTrue((metric.max() - 1000).abs <= Double.Epsilon)
+    assertTrue((metric.min() - 1000).abs <= Double.Epsilon)
   }
 
   private class ManualClock extends Clock {
 
     private var ticksInNanos = 0L
 
-    override def getTick() = {
+    override def tick() = {
       ticksInNanos
     }
 
-    override def getTime() = {
+    override def time() = {
       TimeUnit.NANOSECONDS.toMillis(ticksInNanos)
     }
 

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/LeaderElectionTest.scala Wed Oct 10 18:03:07 2012
@@ -70,20 +70,22 @@ class LeaderElectionTest extends JUnit3S
     // kill the server hosting the preferred replica
     servers.last.shutdown()
     // check if leader moves to the other server
-    val leader2 = waitUntilLeaderIsElectedOrChanged(zkClient, topic, partitionId, 1500, if(leader1.get == 0) None else leader1)
+    val leader2 = waitUntilLeaderIsElectedOrChanged(zkClient, topic, partitionId, 1500,
+      if(leader1.get == 0) None else leader1)
     val leaderEpoch2 = ZkUtils.getEpochForPartition(zkClient, topic, partitionId)
     debug("Leader is elected to be: %s".format(leader1.getOrElse(-1)))
     debug("leader Epoc: " + leaderEpoch2)
     assertEquals("Leader must move to broker 0", 0, leader2.getOrElse(-1))
     if(leader1.get == leader2.get)
-      assertEquals("Second epoch value should be " + leaderEpoch1, leaderEpoch1, leaderEpoch2)
+      assertEquals("Second epoch value should be " + leaderEpoch1+1, leaderEpoch1+1, leaderEpoch2)
     else
       assertEquals("Second epoch value should be %d".format(leaderEpoch1+1) , leaderEpoch1+1, leaderEpoch2)
 
     servers.last.startup()
     servers.head.shutdown()
     Thread.sleep(zookeeper.tickTime)
-    val leader3 = waitUntilLeaderIsElectedOrChanged(zkClient, topic, partitionId, 1500, if(leader2.get == 1) None else leader2)
+    val leader3 = waitUntilLeaderIsElectedOrChanged(zkClient, topic, partitionId, 1500,
+      if(leader2.get == 1) None else leader2)
     val leaderEpoch3 = ZkUtils.getEpochForPartition(zkClient, topic, partitionId)
     debug("leader Epoc: " + leaderEpoch3)
     debug("Leader is elected to be: %s".format(leader3.getOrElse(-1)))

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/RequestPurgatoryTest.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/RequestPurgatoryTest.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/RequestPurgatoryTest.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/server/RequestPurgatoryTest.scala Wed Oct 10 18:03:07 2012
@@ -13,7 +13,7 @@
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
- */
+*/
 
 package kafka.server
 
@@ -69,9 +69,10 @@ class RequestPurgatoryTest extends JUnit
     purgatory.watch(r2)
     purgatory.awaitExpiration(r1)
     val elapsed = System.currentTimeMillis - start
+    println("Start = %d, Elapsed = %d".format(start, elapsed))
     assertTrue("r1 expired", purgatory.expired.contains(r1))
     assertTrue("r2 hasn't expired", !purgatory.expired.contains(r2))
-    assertTrue("Time for expiration %d should at least %d".format(elapsed, expiration), elapsed >= expiration)
+    assertTrue("Time for expiration was about 20ms", (elapsed - expiration).abs < 10L)
   }
   
   class MockRequestPurgatory extends RequestPurgatory[DelayedRequest, ProducerRequest] {

Modified: incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/utils/TestUtils.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/utils/TestUtils.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/utils/TestUtils.scala (original)
+++ incubator/kafka/branches/0.8/core/src/test/scala/unit/kafka/utils/TestUtils.scala Wed Oct 10 18:03:07 2012
@@ -389,7 +389,8 @@ object TestUtils extends Logging {
             newLeaderAndISR.leaderEpoch += 1
             newLeaderAndISR.zkVersion += 1
           }
-          ZkUtils.updatePersistentPath(zkClient, ZkUtils.getTopicPartitionLeaderAndIsrPath( topic, partition), newLeaderAndISR.toString)
+          ZkUtils.updatePersistentPath(zkClient, ZkUtils.getTopicPartitionLeaderAndIsrPath(topic, partition),
+            newLeaderAndISR.toString)
         } catch {
           case oe => error("Error while electing leader for topic %s partition %d".format(topic, partition), oe)
         }

Modified: incubator/kafka/branches/0.8/project/build/KafkaProject.scala
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/project/build/KafkaProject.scala?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/project/build/KafkaProject.scala (original)
+++ incubator/kafka/branches/0.8/project/build/KafkaProject.scala Wed Oct 10 18:03:07 2012
@@ -66,42 +66,17 @@ class KafkaProject(info: ProjectInfo) ex
         <scope>compile</scope>
       </dependency>
 
-    def metricsDeps =
-      <dependencies>
-        <dependency>
-          <groupId>com.yammer.metrics</groupId>
-          <artifactId>metrics-core</artifactId>
-          <version>3.0.0-10ccc80c</version>
-          <scope>compile</scope>
-        </dependency>
-        <dependency>
-          <groupId>com.yammer.metrics</groupId>
-          <artifactId>metrics-annotations</artifactId>
-          <version>3.0.0-10ccc80c</version>
-          <scope>compile</scope>
-        </dependency>
-      </dependencies>
-
     object ZkClientDepAdder extends RuleTransformer(new RewriteRule() {
       override def transform(node: Node): Seq[Node] = node match {
         case Elem(prefix, "dependencies", attribs, scope, deps @ _*) => {
-          Elem(prefix, "dependencies", attribs, scope, deps ++ zkClientDep:_*)
-        }
-        case other => other
-      }
-    })
-
-    object MetricsDepAdder extends RuleTransformer(new RewriteRule() {
-      override def transform(node: Node): Seq[Node] = node match {
-        case Elem(prefix, "dependencies", attribs, scope, deps @ _*) => {
-          Elem(prefix, "dependencies", attribs, scope, deps ++ metricsDeps:_*)
+          Elem(prefix, "dependencies", attribs, scope, deps ++ zkClientDep :_*)
         }
         case other => other
       }
     })
 
     override def pomPostProcess(pom: Node): Node = {
-      MetricsDepAdder(ZkClientDepAdder(pom))
+      ZkClientDepAdder(pom)
     }
 
     override def artifactID = "kafka"
@@ -276,6 +251,7 @@ class KafkaProject(info: ProjectInfo) ex
   trait CoreDependencies {
     val log4j = "log4j" % "log4j" % "1.2.15"
     val jopt = "net.sf.jopt-simple" % "jopt-simple" % "3.2"
+    val metricsCore = "com.yammer.metrics" % "metrics-core" % "latest.release"
     val slf4jSimple = "org.slf4j" % "slf4j-simple" % "latest.release"
   }
   

Modified: incubator/kafka/branches/0.8/system_test/testcase_to_run.json
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/system_test/testcase_to_run.json?rev=1396713&r1=1396712&r2=1396713&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/system_test/testcase_to_run.json (original)
+++ incubator/kafka/branches/0.8/system_test/testcase_to_run.json Wed Oct 10 18:03:07 2012
@@ -1,6 +1,5 @@
 {
-    "ReplicaBasicTest"  : [
-        "testcase_0001",
+    "ReplicaBasicTest"   : [
         "testcase_1"
     ]
 }



Mime
View raw message