Index: core/src/main/scala/kafka/utils/Utils.scala
===================================================================
--- core/src/main/scala/kafka/utils/Utils.scala	(revision 1373685)
+++ core/src/main/scala/kafka/utils/Utils.scala	(working copy)
@@ -216,7 +216,7 @@
     if(props.containsKey(name))
       return getInt(props, name, -1)
     else
-      throw new KafkaException("Missing required property '" + name + "'")
+      throw new IllegalArgumentException("Missing required property '" + name + "'")
   }
   
   /**
@@ -239,7 +239,7 @@
    * @param name The property name
    * @param default The default value to use if the property is not found
    * @param range The range in which the value must fall (inclusive)
-   * @throws KafkaException If the value is not in the given range
+   * @throws IllegalArgumentException If the value is not in the given range
    * @return the integer value
    */
   def getIntInRange(props: Properties, name: String, default: Int, range: (Int, Int)): Int = {
@@ -249,7 +249,7 @@
       else
         default
     if(v < range._1 || v > range._2)
-      throw new KafkaException(name + " has value " + v + " which is not in the range " + range + ".")
+      throw new IllegalArgumentException(name + " has value " + v + " which is not in the range " + range + ".")
     else
       v
   }
@@ -261,7 +261,7 @@
       else
         default
     if(v < range._1 || v > range._2)
-      throw new KafkaException(name + " has value " + v + " which is not in the range " + range + ".")
+      throw new IllegalArgumentException(name + " has value " + v + " which is not in the range " + range + ".")
     else
       v
   }
@@ -269,21 +269,21 @@
   def getIntInRange(buffer: ByteBuffer, name: String, range: (Int, Int)): Int = {
     val value = buffer.getInt
     if(value < range._1 || value > range._2)
-      throw new KafkaException(name + " has value " + value + " which is not in the range " + range + ".")
+      throw new IllegalArgumentException(name + " has value " + value + " which is not in the range " + range + ".")
     else value
   }
 
   def getShortInRange(buffer: ByteBuffer, name: String, range: (Short, Short)): Short = {
     val value = buffer.getShort
     if(value < range._1 || value > range._2)
-      throw new KafkaException(name + " has value " + value + " which is not in the range " + range + ".")
+      throw new IllegalArgumentException(name + " has value " + value + " which is not in the range " + range + ".")
     else value
   }
 
   def getLongInRange(buffer: ByteBuffer, name: String, range: (Long, Long)): Long = {
     val value = buffer.getLong
     if(value < range._1 || value > range._2)
-      throw new KafkaException(name + " has value " + value + " which is not in the range " + range + ".")
+      throw new IllegalArgumentException(name + " has value " + value + " which is not in the range " + range + ".")
     else value
   }
 
@@ -294,7 +294,7 @@
     if(props.containsKey(name))
       return getLong(props, name, -1)
     else
-      throw new KafkaException("Missing required property '" + name + "'")
+      throw new IllegalArgumentException("Missing required property '" + name + "'")
   }
 
   /**
@@ -314,7 +314,7 @@
    * @param name The property name
    * @param default The default value to use if the property is not found
    * @param range The range in which the value must fall (inclusive)
-   * @throws KafkaException If the value is not in the given range
+   * @throws IllegalArgumentException If the value is not in the given range
    * @return the long value
    */
   def getLongInRange(props: Properties, name: String, default: Long, range: (Long, Long)): Long = {
@@ -324,7 +324,7 @@
       else
         default
     if(v < range._1 || v > range._2)
-      throw new KafkaException(name + " has value " + v + " which is not in the range " + range + ".")
+      throw new IllegalArgumentException(name + " has value " + v + " which is not in the range " + range + ".")
     else
       v
   }
@@ -344,7 +344,7 @@
     else if("false" == props.getProperty(name))
       false
     else
-      throw new KafkaException("Unacceptable value for property '" + name + "', boolean values must be either 'true' or 'false" )
+      throw new IllegalArgumentException("Unacceptable value for property '" + name + "', boolean values must be either 'true' or 'false" )
   }
   
   /**
@@ -364,7 +364,7 @@
     if(props.containsKey(name))
       props.getProperty(name)
     else
-      throw new KafkaException("Missing required property '" + name + "'")
+      throw new IllegalArgumentException("Missing required property '" + name + "'")
   }
 
   /**
@@ -378,13 +378,13 @@
       for(i <- 0 until propValues.length) {
         val prop = propValues(i).split("=")
         if(prop.length != 2)
-          throw new KafkaException("Illegal format of specifying properties '" + propValues(i) + "'")
+          throw new IllegalArgumentException("Illegal format of specifying properties '" + propValues(i) + "'")
         properties.put(prop(0), prop(1))
       }
       properties
     }
     else
-      throw new KafkaException("Missing required property '" + name + "'")
+      throw new IllegalArgumentException("Missing required property '" + name + "'")
   }
 
   /**
@@ -395,12 +395,12 @@
       val propString = props.getProperty(name)
       val propValues = propString.split(",")
       if(propValues.length < 1)
-        throw new KafkaException("Illegal format of specifying properties '" + propString + "'")
+        throw new IllegalArgumentException("Illegal format of specifying properties '" + propString + "'")
       val properties = new Properties
       for(i <- 0 until propValues.length) {
         val prop = propValues(i).split("=")
         if(prop.length != 2)
-          throw new KafkaException("Illegal format of specifying properties '" + propValues(i) + "'")
+          throw new IllegalArgumentException("Illegal format of specifying properties '" + propValues(i) + "'")
         properties.put(prop(0), prop(1))
       }
       properties
@@ -729,7 +729,7 @@
         val clazz = Class.forName(className)
         val clazzT = clazz.asInstanceOf[Class[T]]
         val constructors = clazzT.getConstructors
-        require(constructors.length == 1)
+        require(constructors.length == 1, "")
         constructors.head.newInstance().asInstanceOf[T]
     }
   }
@@ -816,6 +816,10 @@
     }
   }
 
+  def require(condition: Boolean, errmsg: String) {
+    if (!condition)
+      throw new IllegalArgumentException(errmsg)
+  }
   /**
    * Create a circular (looping) iterator over a collection.
    * @param coll An iterable over the underlying collection.
Index: core/src/main/scala/kafka/server/KafkaConfig.scala
===================================================================
--- core/src/main/scala/kafka/server/KafkaConfig.scala	(revision 1373685)
+++ core/src/main/scala/kafka/server/KafkaConfig.scala	(working copy)
@@ -36,7 +36,8 @@
 
   /* the broker id for this server */
   val brokerId: Int = Utils.getInt(props, "brokerid")
-  
+  Utils.require(brokerId >= 0, "Unacceptable value '" + brokerId + "'. The broker Id must be non-negative.")
+
   /* the SO_SNDBUFF buffer of the socket sever sockets */
   val socketSendBuffer: Int = Utils.getInt(props, "socket.send.buffer", 100*1024)