diff --git a/lib/utils.js b/lib/utils.js
index 0df1a19a..cf1edc1b 100644
--- a/lib/utils.js
+++ b/lib/utils.js
@@ -1,6 +1,7 @@
 var assert = require('assert');
 var InvalidConfigError = require('./errors/InvalidConfigError');
 var legalChars = new RegExp('^[a-zA-Z0-9._-]*$');
+const allowedTopicLength = 249;
 
 function validateConfig (property, value) {
   if (!legalChars.test(value)) {
@@ -8,6 +9,26 @@ function validateConfig (property, value) {
   }
 }
 
+function validateTopicNames (topics) {
+  // Rewriting same validations done by Apache Kafka team for topics
+  // iterating over topics
+  topics.some(function (topic) {
+    if (topic.length <= 0) {
+      throw new InvalidConfigError('topic name is illegal, cannot be empty');
+    }
+    if (topic === '.' || topic === '..') {
+      throw new InvalidConfigError('topic name cannot be . or ..');
+    }
+    if (topic.length > allowedTopicLength) {
+      throw new InvalidConfigError(`topic name is illegal, cannot be longer than ${allowedTopicLength} characters`);
+    }
+    if (!legalChars.test(topic)) {
+      throw new InvalidConfigError(`topic name ${topic} is illegal, contains a character other than ASCII alphanumerics .,_ and -`);
+    }
+  });
+  return true;
+}
+
 function validateTopics (topics) {
   if (topics.some(function (topic) {
     if ('partition' in topic) {
@@ -79,5 +100,6 @@ module.exports = {
   validateConfig: validateConfig,
   validateTopics: validateTopics,
   groupPartitionsByTopic: groupPartitionsByTopic,
-  createTopicPartitionList: createTopicPartitionList
+  createTopicPartitionList: createTopicPartitionList,
+  validateTopicNames: validateTopicNames
 };