jiazhizhong/kafka-console-ui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

topic search

Browse Source
This commit is contained in:
许晓东
2021-09-09 19:53:02 +08:00
parent f4413107d7
commit 966f22da44
8 changed files with 103 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
src/main/java/com/xuxd/kafka/console/beans/enums/TopicType.java Normal file
View File

@@ -0,0 +1,11 @@
package com.xuxd.kafka.console.beans.enums;
/**
* kafka-console-ui.
*
* @author xuxd
* @date 2021-09-09 17:42:38
**/
public enum TopicType {
ALL, SYSTEM, NORMAL;
}

6
src/main/java/com/xuxd/kafka/console/controller/TopicController.java
View File

@@ -1,9 +1,11 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.beans.enums.TopicType;
import com.xuxd.kafka.console.service.TopicService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
/**
@@ -20,7 +22,7 @@ public class TopicController {
private TopicService topicService;
@GetMapping("/list")
public Object getTopicList() {
return topicService.getTopicList();
public Object getTopicList(@RequestParam(required = false) String topic, @RequestParam String type) {
return topicService.getTopicList(topic, TopicType.valueOf(type.toUpperCase()));
}
}

3
src/main/java/com/xuxd/kafka/console/service/TopicService.java
View File

@@ -1,6 +1,7 @@
package com.xuxd.kafka.console.service;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.enums.TopicType;
/**
* kafka-console-ui.
@@ -12,6 +13,6 @@ public interface TopicService {
ResponseData getTopicNameList();
ResponseData getTopicList();
ResponseData getTopicList(String topic, TopicType type);
}

36
src/main/java/com/xuxd/kafka/console/service/impl/TopicServiceImpl.java
View File

@@ -1,12 +1,17 @@
package com.xuxd.kafka.console.service.impl;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.enums.TopicType;
import com.xuxd.kafka.console.beans.vo.TopicDescriptionVO;
import com.xuxd.kafka.console.service.TopicService;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import kafka.console.TopicConsole;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.admin.TopicDescription;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@@ -25,11 +30,36 @@ public class TopicServiceImpl implements TopicService {
private TopicConsole topicConsole;
@Override public ResponseData getTopicNameList() {
return ResponseData.create().data(topicConsole.getTopicNameList()).success();
return ResponseData.create().data(topicConsole.getTopicNameList(true)).success();
}
@Override public ResponseData getTopicList() {
List<TopicDescription> topicDescriptions = topicConsole.getTopicList(topicConsole.getTopicNameList());
@Override public ResponseData getTopicList(String topic, TopicType type) {
Set<String> topicSet = new HashSet<>();
switch (type) {
case SYSTEM:
Set<String> internalTopicSet = topicConsole.getInternalTopicNameList();
if (StringUtils.isEmpty(topic)) {
topicSet.addAll(internalTopicSet);
} else {
if (internalTopicSet.contains(topic)) {
topicSet.add(topic);
} else {
return ResponseData.create().data(Collections.emptyList()).success();
}
}
break;
case NORMAL:
Set<String> internalTopicS = topicConsole.getInternalTopicNameList();
if (internalTopicS.contains(topic)) {
return ResponseData.create().data(Collections.emptyList()).success();
}
topicSet.addAll(StringUtils.isEmpty(topic) ? topicConsole.getTopicNameList(false) : Collections.singleton(topic));
break;
default:
topicSet.addAll(StringUtils.isEmpty(topic) ? topicConsole.getTopicNameList(true) : Collections.singleton(topic));
break;
}
List<TopicDescription> topicDescriptions = topicConsole.getTopicList(topicSet);
topicDescriptions.sort(Comparator.comparing(TopicDescription::name));
return ResponseData.create().data(topicDescriptions.stream().map(d -> TopicDescriptionVO.from(d))).success();

7
src/main/resources/application.yml
View File

@@ -7,7 +7,7 @@ server:
kafka:
config:
# kafka broker地址，多个以逗号分隔
# bootstrap-server: 'localhost:9092'
bootstrap-server: 'localhost:9092'
request-timeout-ms: 60000
# 服务端是否启用acl，如果不启用，下面的几项都忽略即可
enable-acl: true
@@ -17,7 +17,7 @@ kafka:
# 超级管理员用户名，在broker上已经配置为超级管理员
admin-username: admin
# 超级管理员密码
admin-password: admin!QAZ
admin-password: admin
# 启动自动创建配置的超级管理员用户
admin-create: true
# broker连接的zk地址
@@ -48,6 +48,3 @@ logging:
cron:
# clear-dirty-user: 0 * * * * ?
clear-dirty-user: 0 0 1 * * ?
#spring.jpa.database-platform=org.hibernate.dialect.H2Dialect

25
src/main/scala/kafka/console/TopicConsole.scala
View File

@@ -7,6 +7,8 @@ import java.util.{Collections, List, Set}
import com.xuxd.kafka.console.config.KafkaConfig
import org.apache.kafka.clients.admin.{ListTopicsOptions, TopicDescription}
import scala.jdk.CollectionConverters.{CollectionHasAsScala, SetHasAsJava}
/**
* kafka-console-ui.
*
@@ -15,8 +17,13 @@ import org.apache.kafka.clients.admin.{ListTopicsOptions, TopicDescription}
* */
class TopicConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig) with Logging {
def getTopicNameList(): Set[String] = {
withAdminClientAndCatchError(admin => admin.listTopics(new ListTopicsOptions().listInternal(true)).names()
/**
* get all topic name set.
*
* @return all topic name set.
*/
def getTopicNameList(internal: Boolean = true): Set[String] = {
withAdminClientAndCatchError(admin => admin.listTopics(new ListTopicsOptions().listInternal(internal)).names()
.get(3000, TimeUnit.MILLISECONDS),
e => {
log.error("listTopics error.", e)
@@ -24,6 +31,20 @@ class TopicConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig
}).asInstanceOf[Set[String]]
}
/**
* get all internal topic name set.
*
* @return internal topic name set.
*/
def getInternalTopicNameList(): Set[String] = {
withAdminClientAndCatchError(admin => admin.listTopics(new ListTopicsOptions().listInternal(true)).listings()
.get(3000, TimeUnit.MILLISECONDS).asScala.filter(_.isInternal).map(_.name()).toSet[String].asJava,
e => {
log.error("listInternalTopics error.", e)
Collections.emptySet()
}).asInstanceOf[Set[String]]
}
def getTopicList(topics: Set[String]): List[TopicDescription] = {
if (topics == null || topics.isEmpty) {
Collections.emptyList()