kafka console initial commit

src/main/scala/kafka/console/KafkaAclConsole.scala

@@ -0,0 +1,135 @@
+package kafka.console
+
+import java.util
+import java.util.concurrent.TimeUnit
+import java.util.{Collections, List}
+
+import com.xuxd.kafka.console.beans.AclEntry
+import com.xuxd.kafka.console.config.KafkaConfig
+import org.apache.kafka.common.acl._
+import org.apache.kafka.common.resource.{ResourcePattern, ResourcePatternFilter, ResourceType}
+
+import scala.jdk.CollectionConverters.SetHasAsJava
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-08-28 19:53:06
+ * */
+class KafkaAclConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig) with Logging {
+
+    def getAclList(): List[AclBinding] = {
+        withAdminClient(adminClient => adminClient.describeAcls(AclBindingFilter.ANY).values().get()).asInstanceOf[List[AclBinding]]
+    }
+
+    def addAcl(acls: List[AclBinding]): Boolean = {
+        withAdminClient(adminClient => {
+            try {
+                adminClient.createAcls(acls).all().get(3000, TimeUnit.MILLISECONDS)
+                true
+            } catch {
+                case e: Exception => log.error("addAcl error", e)
+                    false
+            }
+        }).asInstanceOf[Boolean]
+    }
+
+    // must param: entry.topic, entry.user
+    def addProducerAcl(entry: AclEntry): Boolean = {
+        // topic
+        // user
+        val param = entry.toAclBinding
+
+        val binding = new AclBinding(new ResourcePattern(ResourceType.TOPIC, param.pattern().name(), param.pattern().patternType()),
+            new AccessControlEntry(param.entry().principal(), param.entry().host(), param.entry().operation(), AclPermissionType.ALLOW))
+
+        addAcl(new util.ArrayList[AclBinding](getAclBindings(binding, Set(AclOperation.CREATE, AclOperation.DESCRIBE, AclOperation.WRITE)).asJava))
+    }
+
+    def addConsumerAcl(topicEntry: AclEntry, groupEntry: AclEntry): Boolean = {
+        // topic
+        // group
+        // user
+        val tp = topicEntry.toAclBinding
+        val gp = groupEntry.toAclBinding
+
+        val bindingTopic = new AclBinding(new ResourcePattern(ResourceType.TOPIC, tp.pattern().name(), tp.pattern().patternType()),
+            new AccessControlEntry(tp.entry().principal(), tp.entry().host(), tp.entry().operation(), AclPermissionType.ALLOW))
+
+        val bindingGroup = new AclBinding(new ResourcePattern(ResourceType.GROUP, gp.pattern().name(), gp.pattern().patternType()),
+            new AccessControlEntry(gp.entry().principal(), gp.entry().host(), gp.entry().operation(), AclPermissionType.ALLOW))
+
+        val acls: Set[AclBinding] = getAclBindings(bindingTopic, Set(AclOperation.READ)) ++ getAclBindings(bindingGroup, Set(AclOperation.READ))
+
+        addAcl(new util.ArrayList[AclBinding](acls.asJava))
+    }
+
+    def deleteAcl(entry: AclEntry, allResource: Boolean, allPrincipal: Boolean, allOperation: Boolean): Boolean = {
+        withAdminClient(adminClient => {
+            try {
+                val result = adminClient.deleteAcls(Collections.singleton(entry.toAclBindingFilter(allResource, allPrincipal, allOperation))).all().get(3000, TimeUnit.MILLISECONDS)
+                log.info("delete acl: {}", result)
+                true
+            } catch {
+                case e: Exception => log.error("addAcl error", e)
+                    false
+            }
+        }).asInstanceOf[Boolean]
+    }
+
+    def deleteAcl(filters: util.Collection[AclBindingFilter]): Boolean = {
+        withAdminClient(adminClient => {
+            try {
+                val result = adminClient.deleteAcls(filters).all().get(3000, TimeUnit.MILLISECONDS)
+                log.info("delete acl: {}", result)
+                true
+            } catch {
+                case e: Exception => log.error("deleteAcl error", e)
+                    false
+            }
+        }).asInstanceOf[Boolean]
+    }
+
+    def deleteUserAcl(entry: AclEntry): Boolean = {
+        val filter: AclBindingFilter = entry.toAclBindingFilter
+        val delFilter = new AclBindingFilter(new ResourcePatternFilter(ResourceType.ANY, ResourcePattern.WILDCARD_RESOURCE, filter.patternFilter().patternType()),
+            new AccessControlEntryFilter(filter.entryFilter().principal(), filter.entryFilter().host(), AclOperation.ANY, AclPermissionType.ANY))
+
+        deleteAcl(Collections.singleton(delFilter))
+    }
+
+    def deleteProducerAcl(entry: AclEntry): Boolean = {
+        val filter: AclBindingFilter = entry.toAclBindingFilter
+        val delFilter = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, filter.patternFilter().name(), filter.patternFilter().patternType()),
+            new AccessControlEntryFilter(filter.entryFilter().principal(), filter.entryFilter().host(), AclOperation.ANY, AclPermissionType.ANY))
+
+        deleteAcl(getAclFilters(delFilter, Set(AclOperation.CREATE, AclOperation.DESCRIBE, AclOperation.WRITE)).asJava)
+    }
+
+    def deleteConsumerAcl(topic: AclEntry, group: AclEntry): Boolean = {
+        val (topicFilter, groupFilter) = (topic.toAclBindingFilter(), group.toAclBindingFilter())
+        val delTopicFilter = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, topicFilter.patternFilter().name(), topicFilter.patternFilter().patternType()),
+            new AccessControlEntryFilter(topicFilter.entryFilter().principal(), topicFilter.entryFilter().host(), AclOperation.ANY, AclPermissionType.ANY))
+        val delGroupFilter = new AclBindingFilter(new ResourcePatternFilter(ResourceType.GROUP, groupFilter.patternFilter().name(), groupFilter.patternFilter().patternType()),
+            new AccessControlEntryFilter(groupFilter.entryFilter().principal(), groupFilter.entryFilter().host(), AclOperation.ANY, AclPermissionType.ANY))
+
+        val filters = getAclFilters(delTopicFilter, Set(AclOperation.READ)) ++ getAclFilters(delGroupFilter, Set(AclOperation.READ))
+        deleteAcl(filters.asJava)
+    }
+
+    private def getAclBindings(binding: AclBinding, ops: Set[AclOperation]): Set[AclBinding] = {
+        for {
+            op <- ops
+        } yield {
+            new AclBinding(new ResourcePattern(binding.pattern().resourceType(), binding.pattern().name(), binding.pattern().patternType()),
+                new AccessControlEntry(binding.entry().principal(), binding.entry().host(), op, binding.entry().permissionType()))
+        }
+    }
+
+    private def getAclFilters(filter: AclBindingFilter, ops: Set[AclOperation]): Set[AclBindingFilter] = {
+        ops.map(o => new AclBindingFilter(new ResourcePatternFilter(filter.patternFilter().resourceType(), filter.patternFilter().name(), filter.patternFilter().patternType()),
+            new AccessControlEntryFilter(filter.entryFilter().principal(), filter.entryFilter().host(), o, filter.entryFilter().permissionType())))
+    }
+
+}

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

@@ -0,0 +1,56 @@
+package kafka.console
+
+import java.util
+import java.util.Set
+import java.util.concurrent.TimeUnit
+
+import com.xuxd.kafka.console.config.KafkaConfig
+import org.apache.kafka.clients.admin.{ScramCredentialInfo, ScramMechanism, UserScramCredentialDeletion, UserScramCredentialUpsertion}
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-08-28 11:29:48
+ * */
+class KafkaConfigConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig) with Logging {
+
+    private val defaultIterations = 4096
+
+    def getUserList(): Set[String] = {
+        withAdminClient({
+            adminClient => adminClient.describeUserScramCredentials().all().get().keySet()
+        }).asInstanceOf[Set[String]]
+    }
+
+    def addOrUpdateUser(name: String, pass: String): Boolean = {
+        withAdminClient(adminClient => {
+            try {
+                adminClient.alterUserScramCredentials(util.Arrays.asList(
+                    new UserScramCredentialUpsertion(name,
+                        new ScramCredentialInfo(ScramMechanism.fromMechanismName(config.getSaslMechanism), defaultIterations), pass)))
+                    .all().get(3000, TimeUnit.MILLISECONDS)
+                true
+            } catch {
+                case ex: Exception => log.error("addOrUpdateUser error", ex)
+                    false
+            }
+
+        }).asInstanceOf[Boolean]
+    }
+
+    def deleteUser(name: String): Boolean = {
+        withAdminClient(adminClient => {
+            try {
+                adminClient.alterUserScramCredentials(util.Arrays.asList(
+                    new UserScramCredentialDeletion(name, ScramMechanism.fromMechanismName(config.getSaslMechanism))))
+                    .all().get(3000, TimeUnit.MILLISECONDS)
+                true
+            } catch {
+                case ex: Exception => log.error("deleteUser error", ex)
+                    false
+            }
+
+        }).asInstanceOf[Boolean]
+    }
+}

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

@@ -0,0 +1,38 @@
+package kafka.console
+
+import java.util.Properties
+
+import com.xuxd.kafka.console.config.KafkaConfig
+import org.apache.kafka.clients.CommonClientConfigs
+import org.apache.kafka.clients.admin.{Admin, AdminClientConfig}
+import org.apache.kafka.common.config.SaslConfigs
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-08-28 11:56:48
+ * */
+class KafkaConsole(config: KafkaConfig) {
+
+    protected def withAdminClient(f: Admin => Any): Any = {
+
+        val admin = createAdminClient()
+        try {
+            f(admin)
+        } finally {
+            admin.close()
+        }
+    }
+
+    private def createAdminClient(): Admin = {
+        val props: Properties = new Properties();
+        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, config.getBootstrapServer)
+        props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, config.getRequestTimeoutMs())
+        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, config.getSecurityProtocol())
+        props.put(SaslConfigs.SASL_MECHANISM, config.getSaslMechanism())
+        props.put(SaslConfigs.SASL_JAAS_CONFIG, config.getSaslJaasConfig())
+
+        Admin.create(props)
+    }
+}

src/main/scala/kafka/console/Logging.scala

@@ -0,0 +1,14 @@
+package kafka.console
+
+import org.slf4j.{Logger, LoggerFactory}
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-08-28 11:30:14
+ * */
+trait Logging {
+
+    protected lazy val log : Logger = LoggerFactory.getLogger(this.getClass)
+}