kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject git commit: Incorrect index in the log of a follower; patched by Jun Rao; reviewed by Neha Narkhede and Jay Kreps; kafka-804
Date Mon, 18 Mar 2013 22:53:16 GMT
Updated Branches:
  refs/heads/0.8 d40d255bc -> 46ebdc16e


Incorrect index in the log of a follower; patched by Jun Rao; reviewed by Neha Narkhede and
Jay Kreps; kafka-804


Project: http://git-wip-us.apache.org/repos/asf/kafka/repo
Commit: http://git-wip-us.apache.org/repos/asf/kafka/commit/46ebdc16
Tree: http://git-wip-us.apache.org/repos/asf/kafka/tree/46ebdc16
Diff: http://git-wip-us.apache.org/repos/asf/kafka/diff/46ebdc16

Branch: refs/heads/0.8
Commit: 46ebdc16e99c4bb9236a6c25c47ff2c97fdc0f53
Parents: d40d255
Author: Jun Rao <junrao@gmail.com>
Authored: Mon Mar 18 15:53:03 2013 -0700
Committer: Jun Rao <junrao@gmail.com>
Committed: Mon Mar 18 15:53:03 2013 -0700

----------------------------------------------------------------------
 core/src/main/scala/kafka/log/Log.scala          |   30 ++++++++--------
 core/src/test/scala/unit/kafka/log/LogTest.scala |    8 ++--
 2 files changed, 19 insertions(+), 19 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kafka/blob/46ebdc16/core/src/main/scala/kafka/log/Log.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/kafka/log/Log.scala b/core/src/main/scala/kafka/log/Log.scala
index f6ee475..7d71451 100644
--- a/core/src/main/scala/kafka/log/Log.scala
+++ b/core/src/main/scala/kafka/log/Log.scala
@@ -270,33 +270,32 @@ private[kafka] class Log(val dir: File,
 
       try {
         // they are valid, insert them in the log
-        val offsetsAndNumAppendedMessages = lock synchronized {
+        val offsets = lock synchronized {
           val firstOffset = nextOffset.get
 
           // maybe roll the log if this segment is full
           val segment = maybeRoll(segments.view.last)
           
           // assign offsets to the messageset
-          val offsets = 
+          val lastOffset =
             if(assignOffsets) {
               val offsetCounter = new AtomicLong(nextOffset.get)
-              val firstOffset = offsetCounter.get
               try {
                 validMessages = validMessages.assignOffsets(offsetCounter, messageSetInfo.codec)
               } catch {
                 case e: IOException => throw new KafkaException("Error in validating messages
while appending to log '%s'".format(name), e)
               }
-              val lastOffset = offsetCounter.get - 1
-              val numMessages = lastOffset - firstOffset + 1
+              val assignedLastOffset = offsetCounter.get - 1
+              val numMessages = assignedLastOffset - firstOffset + 1
               BrokerTopicStats.getBrokerTopicStats(topicName).messagesInRate.mark(numMessages)
               BrokerTopicStats.getBrokerAllTopicsStats.messagesInRate.mark(numMessages)
-              (firstOffset, lastOffset)
+              assignedLastOffset
             } else {
               require(messageSetInfo.offsetsMonotonic, "Out of order offsets found in " +
messages)
               require(messageSetInfo.firstOffset >= nextOffset.get, 
                       "Attempt to append a message set beginning with offset %d to a log
with log end offset %d."
                       .format(messageSetInfo.firstOffset, nextOffset.get))
-              (messageSetInfo.firstOffset, messageSetInfo.lastOffset)
+              messageSetInfo.lastOffset
             }
 
           // Check if the message sizes are valid. This check is done after assigning offsets
to ensure the comparison
@@ -308,23 +307,24 @@ private[kafka] class Log(val dir: File,
           }
 
           // now append to the log
-          trace("Appending message set to %s offset: %d nextOffset: %d messageSet: %s"
-                .format(this.name, offsets._1, nextOffset.get(), validMessages))
-          segment.append(offsets._1, validMessages)
+          segment.append(firstOffset, validMessages)
           
           // advance the log end offset
-          nextOffset.set(offsets._2 + 1)
-          val numAppendedMessages = (nextOffset.get - firstOffset).toInt
+          nextOffset.set(lastOffset + 1)
+
+          trace("Appended message set to log %s with first offset: %d, next offset: %d, and
messages: %s"
+                  .format(this.name, firstOffset, nextOffset.get(), validMessages))
 
           // return the offset at which the messages were appended
-          (offsets._1, offsets._2, numAppendedMessages)
+          (firstOffset, lastOffset)
         }
         
         // maybe flush the log and index
-        maybeFlush(offsetsAndNumAppendedMessages._3)
+        val numAppendedMessages = (offsets._2 - offsets._1 + 1).toInt
+        maybeFlush(numAppendedMessages)
         
         // return the first and last offset
-        (offsetsAndNumAppendedMessages._1, offsetsAndNumAppendedMessages._2)
+        offsets
       } catch {
         case e: IOException => throw new KafkaStorageException("I/O exception in append
to log '%s'".format(name), e)
       }

http://git-wip-us.apache.org/repos/asf/kafka/blob/46ebdc16/core/src/test/scala/unit/kafka/log/LogTest.scala
----------------------------------------------------------------------
diff --git a/core/src/test/scala/unit/kafka/log/LogTest.scala b/core/src/test/scala/unit/kafka/log/LogTest.scala
index 786ae03..4ed88e8 100644
--- a/core/src/test/scala/unit/kafka/log/LogTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogTest.scala
@@ -379,15 +379,15 @@ class LogTest extends JUnitSuite {
                         needsRecovery = true)
       val messages = List("one", "two", "three", "four", "five", "six")
       val ms = new ByteBufferMessageSet(compressionCodec = codec, 
-                                        offsetCounter = new AtomicLong(5), 
+                                        offsetCounter = new AtomicLong(0),
                                         messages = messages.map(s => new Message(s.getBytes)):_*)
-      val firstOffset = ms.shallowIterator.toList.head.offset
-      val lastOffset = ms.shallowIterator.toList.last.offset
+      val firstOffset = ms.toList.head.offset
+      val lastOffset = ms.toList.last.offset
       val (first, last) = log.append(ms, assignOffsets = false)
       assertEquals(last + 1, log.logEndOffset)
       assertEquals(firstOffset, first)
       assertEquals(lastOffset, last)
-      assertTrue(log.read(5, 64*1024).size > 0)
+      assertTrue(log.read(0, 64*1024).size > 0)
       log.delete()
     }
   }


Mime
View raw message