System Test Transient Failure on testcase_0122; patched by Sriram Sub…

…ramanian; reviewed by Jun Rao and Neha Narkhede; kafka-772
apache · Mar 5, 2013 · 0ee46e0 · 0ee46e0
1 parent 92ecebe
commit 0ee46e0
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/core/src/main/scala/kafka/server/AbstractFetcherThread.scala b/core/src/main/scala/kafka/server/AbstractFetcherThread.scala
@@ -114,7 +114,8 @@ abstract class AbstractFetcherThread(name: String, clientId: String, sourceBroke
           case(topicAndPartition, partitionData) =>
             val (topic, partitionId) = topicAndPartition.asTuple
             val currentOffset = partitionMap.get(topicAndPartition)
-            if (currentOffset.isDefined) {
+            // we append to the log if the current offset is defined and it is the same as the offset requested during fetch
+            if (currentOffset.isDefined && fetchRequest.requestInfo(topicAndPartition).offset == currentOffset.get) {
               partitionData.error match {
                 case ErrorMapping.NoError =>
                   val messages = partitionData.messages.asInstanceOf[ByteBufferMessageSet]