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 ||
diff --git a/core/src/test/scala/unit/kafka/producer/ProducerTest.scala b/core/src/test/scala/unit/kafka/producer/ProducerTest.scala
index b511d90..955709f 100644
--- a/core/src/test/scala/unit/kafka/producer/ProducerTest.scala
+++ b/core/src/test/scala/unit/kafka/producer/ProducerTest.scala
@@ -205,6 +205,7 @@ class ProducerTest extends JUnit3Suite with ZooKeeperTestHarness with Logging{
     props.put("partitioner.class", "kafka.utils.StaticPartitioner")
     props.put("request.timeout.ms", "2000")
     props.put("request.required.acks", "1")
+    props.put("message.send.max.retries", "1")
     props.put("metadata.broker.list", TestUtils.getBrokerListStrFromConfigs(Seq(config1, config2)))
 
     val topic = "new-topic"
@@ -217,8 +218,13 @@ class ProducerTest extends JUnit3Suite with ZooKeeperTestHarness with Logging{
     TestUtils.waitUntilLeaderIsElectedOrChanged(zkClient, topic, 2, 500)
     TestUtils.waitUntilLeaderIsElectedOrChanged(zkClient, topic, 3, 500)
 
-    val config = new ProducerConfig(props)
-    val producer = new Producer[String, String](config)
+    val prodConfig = new ProducerConfig(props)
+    val producer = new Producer[String, String](prodConfig)
+
+    props.put("request.required.acks", "0")
+    val prodConfig2 = new ProducerConfig(props)
+    val producer2 = new Producer[String, String](prodConfig2)
+
     try {
       // Available partition ids should be 0, 1, 2 and 3, all lead and hosted only
       // on broker 0
@@ -239,6 +245,15 @@ class ProducerTest extends JUnit3Suite with ZooKeeperTestHarness with Logging{
       case e => // success
     }
 
+    // try send again with ack=0, should fail also
+    try {
+      // These sends should fail since there are no available brokers
+      producer2.send(new KeyedMessage[String, String](topic, "test", "test1"))
+      fail("Should fail since no leader exists for the partition.")
+    } catch {
+      case e => // success
+    }
+
     // restart server 1
     server1.startup()
     TestUtils.waitUntilLeaderIsElectedOrChanged(zkClient, topic, 0, 500)
@@ -254,6 +269,7 @@ class ProducerTest extends JUnit3Suite with ZooKeeperTestHarness with Logging{
       case e: Exception => fail("Not expected", e)
     }
     producer.close
+    producer2.close
   }
 
   @Test