diff --git a/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaBrokerScript.java b/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaBrokerScript.java
index 75d26758..ff18e43e 100644
--- a/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaBrokerScript.java
+++ b/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaBrokerScript.java
@@ -30,14 +30,23 @@
 
 import java.io.IOException;
 import java.text.MessageFormat;
+import java.util.Properties;
 
 @Slf4j
 @AutoService(Script.class)
 public class KafkaBrokerScript extends AbstractServerScript {
 
+    @Override
+    public ShellResult add(Params params) {
+        Properties properties = new Properties();
+        properties.setProperty(PROPERTY_KEY_SKIP_LEVELS, "1");
+
+        return super.add(params, properties);
+    }
+
     @Override
     public ShellResult configure(Params params) {
-        return KafkaSetup.config(params);
+        return KafkaSetup.configure(params);
     }
 
     @Override
diff --git a/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaParams.java b/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaParams.java
index d1d2ea9a..ce126a59 100644
--- a/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaParams.java
+++ b/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaParams.java
@@ -81,6 +81,11 @@ public String kafkaLimits() {
         return (String) kafkaLimits.get("content");
     }
 
+    @Override
+    public String confDir() {
+        return serviceHome() + "/config";
+    }
+
     @Override
     public String getServiceName() {
         return "kafka";
diff --git a/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaSetup.java b/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaSetup.java
index d21a4072..f94049b1 100644
--- a/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaSetup.java
+++ b/bigtop-manager-stack/bigtop-manager-stack-bigtop/src/main/java/org/apache/bigtop/manager/stack/bigtop/v3_3_0/kafka/KafkaSetup.java
@@ -41,7 +41,8 @@
 @NoArgsConstructor(access = AccessLevel.PRIVATE)
 public class KafkaSetup {
 
-    public static ShellResult config(Params params) {
+    public static ShellResult configure(Params params) {
+        log.info("Configuring Kafka");
         KafkaParams kafkaParams = (KafkaParams) params;
 
         String confDir = kafkaParams.confDir();
@@ -52,7 +53,6 @@ public static ShellResult config(Params params) {
         LinuxFileUtils.createDirectories(kafkaParams.getKafkaLogDir(), kafkaUser, kafkaGroup, PERMISSION_755, true);
         LinuxFileUtils.createDirectories(kafkaParams.getKafkaPidDir(), kafkaUser, kafkaGroup, PERMISSION_755, true);
 
-        // server.properties
         List<String> zookeeperServerHosts = LocalSettings.hosts("zookeeper_server");
         Map<String, Object> paramMap = new HashMap<>();
         paramMap.put("zk_server_list", zookeeperServerHosts);
@@ -66,7 +66,6 @@ public static ShellResult config(Params params) {
                 kafkaParams.kafkaBroker(),
                 paramMap);
 
-        // kafka-env
         LinuxFileUtils.toFileByTemplate(
                 kafkaParams.getKafkaEnvContent(),
                 MessageFormat.format("{0}/kafka-env.sh", confDir),
@@ -75,7 +74,6 @@ public static ShellResult config(Params params) {
                 PERMISSION_644,
                 kafkaParams.getGlobalParamsMap());
 
-        // log4j
         LinuxFileUtils.toFileByTemplate(
                 kafkaParams.getKafkaLog4jContent(),
                 MessageFormat.format("{0}/log4j.properties", confDir),
@@ -84,7 +82,6 @@ public static ShellResult config(Params params) {
                 PERMISSION_644,
                 kafkaParams.getGlobalParamsMap());
 
-        // kafka.limits
         LinuxFileUtils.toFileByTemplate(
                 kafkaParams.kafkaLimits(),
                 MessageFormat.format("{0}/kafka.conf", KafkaParams.LIMITS_CONF_DIR),
@@ -93,6 +90,7 @@ public static ShellResult config(Params params) {
                 PERMISSION_644,
                 kafkaParams.getGlobalParamsMap());
 
-        return ShellResult.success("Kafka Server Configure success!");
+        log.info("Successfully configured Kafka");
+        return ShellResult.success();
     }
 }