Index: core/src/test/scala/unit/kafka/log/LogManagerTest.scala
===================================================================
--- core/src/test/scala/unit/kafka/log/LogManagerTest.scala	(revision 1384137)
+++ core/src/test/scala/unit/kafka/log/LogManagerTest.scala	(working copy)
@@ -179,33 +179,4 @@
     assertTrue("The last flush time has to be within defaultflushInterval of current time ",
                      (System.currentTimeMillis - log.getLastFlushedTime) < 100)
   }
-
-  @Test
-  def testConfigurablePartitions() {
-    val props = TestUtils.createBrokerConfig(0, -1)
-    logManager.shutdown()
-    config = new KafkaConfig(props) {
-                   override val logFileSize = 256
-                   override val topicPartitionsMap = Utils.getTopicPartitions("testPartition:2")
-                   override val flushInterval = 100
-                 }
-    logManager = new LogManager(config, scheduler, time, maxRollInterval, veryLargeLogFlushInterval, maxLogAge, false)
-    logManager.startup
-
-    for(i <- 0 until 1) {
-      val log = logManager.getOrCreateLog(name, i)
-      for(i <- 0 until 250) {
-        var set = TestUtils.singleMessageSet("test".getBytes())
-        log.append(set)
-      }
-    }
-
-    try
-    {
-      val log = logManager.getOrCreateLog(name, 2)
-      assertTrue("Should not come here", log != null)
-    } catch {
-       case _ =>
-    }
-  }
 }
Index: core/src/main/scala/kafka/server/KafkaConfig.scala
===================================================================
--- core/src/main/scala/kafka/server/KafkaConfig.scala	(revision 1384137)
+++ core/src/main/scala/kafka/server/KafkaConfig.scala	(working copy)
@@ -112,9 +112,6 @@
   /* the maximum time in ms that a message in any topic is kept in memory before flushed to disk */
   val defaultFlushIntervalMs = props.getInt("log.default.flush.interval.ms", flushSchedulerThreadRate)
 
-   /* the number of partitions for selected topics, e.g., topic1:8,topic2:16 */
-  val topicPartitionsMap = Utils.getTopicPartitions(props.getString("topic.partition.count.map", ""))
-
   /* enable auto creation of topic on the server */
   val autoCreateTopics = props.getBoolean("auto.create.topics", true)
 
Index: core/src/main/scala/kafka/log/LogManager.scala
===================================================================
--- core/src/main/scala/kafka/log/LogManager.scala	(revision 1384137)
+++ core/src/main/scala/kafka/log/LogManager.scala	(working copy)
@@ -23,7 +23,7 @@
 import kafka.server.KafkaConfig
 import kafka.api.OffsetRequest
 import kafka.log.Log._
-import kafka.common.{KafkaException, UnknownTopicOrPartitionException}
+import kafka.common.KafkaException
 
 /**
  * The guy who creates and hands out logs
@@ -95,12 +95,6 @@
    * Create a log for the given topic and the given partition
    */
   private def createLog(topic: String, partition: Int): Log = {
-    if (partition < 0 || partition >= config.topicPartitionsMap.getOrElse(topic, numPartitions)) {
-      val error = "Wrong partition %d, valid partitions (0, %d)."
-              .format(partition, (config.topicPartitionsMap.getOrElse(topic, numPartitions) - 1))
-      warn(error)
-      throw new UnknownTopicOrPartitionException(error)
-    }
     logCreationLock synchronized {
       val d = new File(logDir, topic + "-" + partition)
       d.mkdirs()