diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index 1437496..7e06223 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -84,7 +84,7 @@ object RequestChannel extends Logging {
     }
   }
   
-  case class Response(processor: Int, request: Request, responseSend: Send) {
+  case class Response(processor: Int, request: Request, responseSend: Send, closeSocket: Boolean = false) {
     request.responseCompleteTimeMs = SystemTime.milliseconds
 
     def this(request: Request, send: Send) =
diff --git a/core/src/main/scala/kafka/network/SocketServer.scala b/core/src/main/scala/kafka/network/SocketServer.scala
index d5bd143..c08554b 100644
--- a/core/src/main/scala/kafka/network/SocketServer.scala
+++ b/core/src/main/scala/kafka/network/SocketServer.scala
@@ -278,10 +278,15 @@ private[kafka] class Processor(val id: Int,
           // a null response send object indicates that there is no response to send to the client.
           // In this case, we just want to turn the interest ops to READ to be able to read more pipelined requests
           // that are sitting in the server's socket buffer
-          trace("Socket server received empty response to send, registering for read: " + curr)
-          key.interestOps(SelectionKey.OP_READ)
-          key.attach(null)
-          curr.request.updateRequestMetrics
+          if (curr.closeSocket) {
+            info("Closing socket for " + channelFor(key).socket.getInetAddress + " following the close-socket signal")
+            close(key)
+          } else {
+            trace("Socket server received empty response to send, registering for read: " + curr)
+            key.interestOps(SelectionKey.OP_READ)
+            key.attach(null)
+            curr.request.updateRequestMetrics
+          }
         } else {
           trace("Socket server received response to send, registering for write: " + curr)
           key.interestOps(SelectionKey.OP_WRITE)
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index 208e3ef..557eb1c 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -171,8 +171,14 @@ class KafkaApis(val requestChannel: RequestChannel,
         m => replicaManager.getReplicationFactorForPartition(m.topic, m.partition) != 1)
     if(produceRequest.requiredAcks == 0) {
       // send a fake producer response if producer request.required.acks = 0. This mimics the behavior of a 0.7 producer
-      // and is tuned for very high throughput
-      requestChannel.sendResponse(new RequestChannel.Response(request.processor, request, null))
+      // and is tuned for very high throughput; however, if there is any exception in handling the request the response will
+      // indicate the socket server to close the socket
+      if (numPartitionsInError != 0) {
+        debug("Sending the close socket signal due to error handling produce request [%s] with Ack=0".format(produceRequest.data.toString))
+        requestChannel.sendResponse(new RequestChannel.Response(request.processor, request, null, true))
+      } else {
+        requestChannel.sendResponse(new RequestChannel.Response(request.processor, request, null))
+      }
     } else if (produceRequest.requiredAcks == 1 ||
         produceRequest.numPartitions <= 0 ||
         allPartitionHaveReplicationFactorOne ||