diff -urN a/kafka/common/InvalidTopicException.scala b/kafka/common/InvalidTopicException.scala
--- a/kafka/common/InvalidTopicException.scala	1970-01-01 08:00:00.000000000 +0800
+++ b/kafka/common/InvalidTopicException.scala	2012-07-30 16:36:13.000000000 +0800
@@ -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
+
+/**
+ * Indicates that a topic failed by its checksum and is corrupt
+ */
+class InvalidTopicException(message: String) extends RuntimeException(message) {
+  def this() = this(null)
+}
diff -urN a/kafka/log/LogManager.scala b/kafka/log/LogManager.scala
--- a/kafka/log/LogManager.scala	2012-06-21 00:06:17.000000000 +0800
+++ b/kafka/log/LogManager.scala	2012-07-30 16:36:42.000000000 +0800
@@ -26,6 +26,7 @@
 import kafka.common.{InvalidTopicException, InvalidPartitionException}
 import kafka.api.OffsetRequest
 
+
 /**
  * The guy who creates and hands out logs
  */
@@ -144,6 +145,16 @@
    */
   private def createLog(topic: String, partition: Int): Log = {
     logCreationLock synchronized {
+      //check topic invalid
+      val topicByte = topic.getBytes()
+      for (i <- 0 until topicByte.length)
+      {
+         if(topicByte(i) == 0)
+         {
+           throw new InvalidTopicException("Invalid topic name") 
+         }
+      }
+      
       val d = new File(logDir, topic + "-" + partition)
       d.mkdirs()
       new Log(d, maxSize, flushInterval, false)