Index: core/src/main/scala/kafka/log/Log.scala
===================================================================
--- core/src/main/scala/kafka/log/Log.scala	(revision 1393806)
+++ core/src/main/scala/kafka/log/Log.scala	(working copy)
@@ -25,9 +25,9 @@
 import java.text.NumberFormat
 import kafka.server.BrokerTopicStat
 import kafka.message.{ByteBufferMessageSet, MessageSet, InvalidMessageException, FileMessageSet}
-import kafka.common.{KafkaException, InvalidMessageSizeException, OffsetOutOfRangeException}
 import kafka.metrics.KafkaMetricsGroup
 import com.yammer.metrics.core.Gauge
+import kafka.common.{KafkaIOException, KafkaException, InvalidMessageSizeException, OffsetOutOfRangeException}
 
 object Log {
   val FileSuffix = ".kafka"
@@ -276,8 +276,7 @@
       }
       catch {
         case e: IOException =>
-          fatal("Halting due to unrecoverable I/O error while handling producer request", e)
-          Runtime.getRuntime.halt(1)
+          throw new KafkaIOException("IO exception in log append", e)
         case e2 => throw e2
       }
     }
Index: core/src/main/scala/kafka/common/KafkaIOException.scala
===================================================================
--- core/src/main/scala/kafka/common/KafkaIOException.scala	(revision 0)
+++ core/src/main/scala/kafka/common/KafkaIOException.scala	(revision 0)
@@ -0,0 +1,25 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+*/
+package kafka.common
+
+/**
+ * Kafka exception caused by real IOs.
+*/
+class KafkaIOException(message: String, t: Throwable) extends RuntimeException(message, t) {
+  def this(message: String) = this(message, null)
+  def this(t: Throwable) = this("", t)
+}
Index: core/src/main/scala/kafka/server/KafkaApis.scala
===================================================================
--- core/src/main/scala/kafka/server/KafkaApis.scala	(revision 1393806)
+++ core/src/main/scala/kafka/server/KafkaApis.scala	(working copy)
@@ -183,8 +183,8 @@
           BrokerTopicStat.getBrokerAllTopicStat.failedProduceRequestRate.mark()
           error("Error processing ProducerRequest on %s:%d".format(topic, partitionData.partition), e)
           e match {
-            case _: IOException =>
-              fatal("Halting due to unrecoverable I/O error while handling producer request: " + e.getMessage, e)
+            case _: KafkaIOException =>
+              fatal("Halting due to unrecoverable I/O error while handling producer request", e)
               // compiler requires scala.sys.exit (not System.exit).
               exit(1)
             case _ =>