user detail

src/main/java/com/xuxd/kafka/console/service/AclService.java

@@ -40,4 +40,6 @@ public interface AclService {
 
     ResponseData getOperationList();
 
+    ResponseData getUserDetail(String username);
+
 }

src/main/java/com/xuxd/kafka/console/service/impl/AclServiceImpl.java

@@ -181,6 +181,18 @@ public class AclServiceImpl implements AclService, SmartInitializingSingleton {
         return ResponseData.create().data(operations).success();
     }
 
+    @Override public ResponseData getUserDetail(String username) {
+        Map<String, Object> param = new HashMap<>();
+        param.put("username", username);
+        List<KafkaUserDO> dos = kafkaUserMapper.selectByMap(param);
+        if (dos.isEmpty()) {
+            return ResponseData.create().data(new CounterList<>(dos)).success("Retrieved the user info is null.");
+        }
+        // check for consistency.
+
+        return null;
+    }
+
     @Override public void afterSingletonsInstantiated() {
         if (kafkaConfig.isAdminCreate()) {
             log.info("Start create admin user, username: {}, password: {}", kafkaConfig.getAdminUsername(), kafkaConfig.getAdminPassword());

src/main/scala/kafka/console/KafkaConfigConsole.scala

@@ -7,10 +7,8 @@ import java.util.{Properties, Set}
 import com.xuxd.kafka.console.config.KafkaConfig
 import kafka.server.ConfigType
 import kafka.utils.Implicits.PropertiesOps
-import kafka.zk.{AdminZkClient, KafkaZkClient}
 import org.apache.kafka.clients.admin._
 import org.apache.kafka.common.security.scram.internals.{ScramCredentialUtils, ScramFormatter}
-import org.apache.kafka.common.utils.Time
 
 /**
  * kafka-console-ui.
@@ -51,27 +49,24 @@ class KafkaConfigConsole(config: KafkaConfig) extends KafkaConsole(config: Kafka
     }
 
     def addOrUpdateUserWithZK(name: String, pass: String): Boolean = {
+        withZKClient(adminZkClient => {
+            try {
+                val credential = new ScramFormatter(org.apache.kafka.common.security.scram.internals.ScramMechanism.forMechanismName(config.getSaslMechanism))
+                    .generateCredential(pass, defaultIterations)
+                val credentialStr = ScramCredentialUtils.credentialToString(credential)
 
-        val zkClient = KafkaZkClient(config.getZookeeperAddr, false, 30000, 30000, Int.MaxValue, Time.SYSTEM)
-        val adminZkClient = new AdminZkClient(zkClient)
-        try {
-            val credential = new ScramFormatter(org.apache.kafka.common.security.scram.internals.ScramMechanism.forMechanismName(config.getSaslMechanism))
-                .generateCredential(pass, defaultIterations)
-            val credentialStr = ScramCredentialUtils.credentialToString(credential)
+                val userConfig: Properties = new Properties()
+                userConfig.put(config.getSaslMechanism, credentialStr)
 
-            val userConfig: Properties = new Properties()
-            userConfig.put(config.getSaslMechanism, credentialStr)
-
-            val configs = adminZkClient.fetchEntityConfig(ConfigType.User, name)
-            userConfig ++= configs
-            adminZkClient.changeConfigs(ConfigType.User, name, userConfig)
-            true
-        } catch {
-            case e: Exception => log.error("addOrUpdateAdminWithZK error.", e)
-                false
-        } finally {
-            zkClient.close()
-        }
+                val configs = adminZkClient.fetchEntityConfig(ConfigType.User, name)
+                userConfig ++= configs
+                adminZkClient.changeConfigs(ConfigType.User, name, userConfig)
+                true
+            } catch {
+                case e: Exception => log.error("addOrUpdateAdminWithZK error.", e)
+                    false
+            }
+        }).asInstanceOf[Boolean]
     }
 
     def deleteUser(name: String): (Boolean, String) = {

src/main/scala/kafka/console/KafkaConsole.scala

@@ -3,9 +3,11 @@ package kafka.console
 import java.util.Properties
 
 import com.xuxd.kafka.console.config.KafkaConfig
+import kafka.zk.{AdminZkClient, KafkaZkClient}
 import org.apache.kafka.clients.CommonClientConfigs
 import org.apache.kafka.clients.admin.{Admin, AdminClientConfig}
 import org.apache.kafka.common.config.SaslConfigs
+import org.apache.kafka.common.utils.Time
 
 /**
  * kafka-console-ui.
@@ -25,6 +27,16 @@ class KafkaConsole(config: KafkaConfig) {
         }
     }
 
+    protected def withZKClient(f: AdminZkClient => Any): Any = {
+        val zkClient = KafkaZkClient(config.getZookeeperAddr, false, 30000, 30000, Int.MaxValue, Time.SYSTEM)
+        val adminZkClient = new AdminZkClient(zkClient)
+        try {
+            f(adminZkClient)
+        } finally {
+            zkClient.close()
+        }
+    }
+
     private def createAdminClient(): Admin = {
         val props: Properties = new Properties();
         props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServer)