KAFKA-3427: broker can return incorrect version of fetch response whe…

…n the broker hits an unknown exception Author: Jun Rao <[email protected]> Reviewers: Ismael Juma, Becket Qin Closes apache#1101 from junrao/kafka-3427
LioRoger · Mar 20, 2016 · 6553679 · 6553679
1 parent b6c6291
commit 6553679
Showing 1 changed file with 2 additions and 1 deletion.
diff --git a/core/src/main/scala/kafka/api/FetchRequest.scala b/core/src/main/scala/kafka/api/FetchRequest.scala
@@ -150,7 +150,8 @@ case class FetchRequest(versionId: Short = FetchRequest.CurrentVersion,
       case (topicAndPartition, data) =>
         (topicAndPartition, FetchResponsePartitionData(Errors.forException(e).code, -1, MessageSet.Empty))
     }
-    val errorResponse = FetchResponse(correlationId, fetchResponsePartitionData)
+    val fetchRequest = request.requestObj.asInstanceOf[FetchRequest]
+    val errorResponse = FetchResponse(correlationId, fetchResponsePartitionData, fetchRequest.versionId)
     // Magic value does not matter here because the message set is empty
     requestChannel.sendResponse(new RequestChannel.Response(request, new FetchResponseSend(request.connectionId, errorResponse)))
   }