完善基本功能和readme

README.MD

@@ -1 +1,74 @@
 # kafka map
+
+给编程插上翅膀，给kafka安装上导航。
+
+## 快速了解
+
+`kafka map`是使用`Java8`和`React`给`Apache Kafka`开发的一款web ui工具。
+通过这款工具可以很方便直观的查看Broker、 topics、 partitions、consumers等信息，以及查看Topic中的消息。
+
+目前支持的功能有：
+
+- 多集群管理
+- topic监控 —— 查看分区数量、副本数量、存储大小、offset
+- topic管理 —— 创建、删除、扩容
+- broker查看 —— Partitions as Leader、	Partitions as Follower
+- consumer管理（查看、删除）
+- 重置offset
+- 消费消息 —— 支持String和json方式展示
+
+## 截图
+
+[截图](docs/screenshot.md)
+
+## 协议与条款
+
+如您需要在企业网络中使用 `kafka-map` ，建议先征求 IT 管理员的同意。下载、使用或分发 `kafka-map` 前，您必须同意 [协议](./LICENSE) 条款与限制。本项目不提供任何担保，亦不承担任何责任。
+
+## 依赖环境
+
+- Java8 或更高版本
+- Apache Kafka 1.1.0 或更高版本
+
+
+## 快速安装
+
+下载安装包并解压到你喜欢的目录
+
+### 前台运行
+进入kafka-map文件夹执行
+```shell
+java -jar kafka-map.jar 
+```
+
+### 系统服务方式运行
+
+在 `/etc/systemd/system/` 下创建 `kafka-map.service` 并写入以下内容
+
+> 根据自身实际情况修改 WorkingDirectory 和 ExecStart下面的程序所在目录及java
+```shell
+[Unit]
+Description=kafka map service
+After=network.target
+
+[Service]
+WorkingDirectory=/usr/local/kafka-map
+ExecStart=/usr/bin/java -jar /usr/local/kafka-map/kafka-map.jar
+Restart=on-failure
+
+[Install]
+WantedBy=multi-user.target
+```
+
+重载系统服务&&设置开机启动&&启动服务&&查看状态
+
+```shell
+systemctl daemon-reload
+systemctl enable kafka-map
+systemctl start kafka-map
+systemctl status kafka-map
+```
+
+### 使用
+
+接下来使用浏览器打开服务器的 `8080` 端口即可访问。

build.sh

@@ -0,0 +1,5 @@
+cd web && npm run build
+mv build ../src/main/resources/static
+source ~/.bash_profile
+cd .. && mvn clean package -Dmaven.test.skip=true
+rm -rf src/main/resources/static

docs/screenshot.md

@@ -0,0 +1,28 @@
+
+### 集群管理
+
+![集群管理](../screenshot/cluster.png)
+
+### 主题管理
+
+![主题管理](../screenshot/topic.png)
+
+### 消费组
+
+![主题管理](../screenshot/consumer.png)
+
+### topic详情——分区
+
+![topic详情——分区](../screenshot/topic-info-partition.png)
+
+### topic详情——消费组
+
+![topic详情——消费组](../screenshot/topic-info-consumer.png)
+
+### topic详情——消费组重置offset
+
+![topic详情——消费组重置offset](../screenshot/topic-info-consumer-reset-offset.png)
+
+### 消费消息
+
+![消费消息](../screenshot/topic-data.png)

src/main/java/cn/typesafe/km/controller/handle/GlobalExceptionHandler.java

@@ -0,0 +1,35 @@
+package cn.typesafe.km.controller.handle;
+
+import org.apache.kafka.common.errors.TopicExistsException;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.ControllerAdvice;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.ResponseStatus;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@ControllerAdvice
+public class GlobalExceptionHandler {
+
+    @ResponseStatus(value = HttpStatus.BAD_REQUEST)
+    @ResponseBody
+    @ExceptionHandler(value = TopicExistsException.class)
+    public Map<String, Object> exceptionHandler(TopicExistsException e) {
+        String message = e.getMessage();
+        Map<String, Object> data = new HashMap<>();
+        data.put("message", message);
+        return data;
+    }
+
+    @ResponseStatus(value = HttpStatus.BAD_REQUEST)
+    @ResponseBody
+    @ExceptionHandler(value = IllegalArgumentException.class)
+    public Map<String, Object> illegalArgumentException(IllegalArgumentException e) {
+        String message = e.getMessage();
+        Map<String, Object> data = new HashMap<>();
+        data.put("message", message);
+        return data;
+    }
+}

src/main/java/cn/typesafe/km/service/ClusterService.java

@@ -4,6 +4,7 @@
 import cn.typesafe.km.entity.Cluster;
 import cn.typesafe.km.repository.ClusterRepository;
 import cn.typesafe.km.util.ID;
+import cn.typesafe.km.util.Networks;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AdminClientConfig;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -43,6 +44,7 @@ private AdminClient createAdminClient(String servers) {
         Properties properties = new Properties();
         properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
         properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "5000");
+        properties.put(AdminClientConfig.RETRIES_CONFIG, "0");
         return AdminClient.create(properties);
     }
 
@@ -88,6 +90,19 @@ public AdminClient getAdminClient(String id) {
     @Transactional(rollbackFor = Exception.class)
     public void create(Cluster cluster) throws ExecutionException, InterruptedException {
         String uuid = ID.uuid();
+        for (String server : cluster.getServers().split(",")) {
+            String[] split = server.split(":");
+            String host = split[0];
+            boolean hostReachable = Networks.isHostReachable(host);
+            if (!hostReachable) {
+                throw new IllegalArgumentException("Host " + host + " is not reachable.");
+            }
+            int port = Integer.parseInt(split[1]);
+            boolean hostConnected = Networks.isHostConnected(host, port);
+            if (!hostConnected) {
+                throw new IllegalArgumentException("server " + server + " can't connected.");
+            }
+        }
 
         AdminClient adminClient = getAdminClient(uuid, cluster.getServers());
         String clusterId = adminClient.describeCluster().clusterId().get();

src/main/java/cn/typesafe/km/util/Networks.java

@@ -0,0 +1,29 @@
+package cn.typesafe.km.util;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+
+public class Networks {
+
+    public static boolean isHostReachable(String host) {
+        try {
+            return InetAddress.getByName(host).isReachable(1000);
+        } catch (IOException ignored) {
+
+        }
+        return false;
+    }
+
+    public static boolean isHostConnected(String host, int port) {
+        try (Socket socket = new Socket()) {
+            socket.connect(new InetSocketAddress(host, port), 3000);
+            InetAddress localAddress = socket.getLocalAddress();
+            String hostName = localAddress.getHostName();
+            return true;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+}

src/main/resources/application.yml

@@ -5,4 +5,6 @@ spring:
   jpa:
     hibernate:
       ddl-auto: update
-    show-sql: true
+    show-sql: true
+server:
+  port: 8080

web/src/common/request.js

@@ -19,7 +19,7 @@ const handleError = (error) => {
     }
     if (error.response !== undefined) {
         let data = error.response.data;
-        message.error(`status: ${data.status}, error: ${data.error}, message: ${data.message}`, 10);
+        message.error(`${data.message}`, 10);
         return false;
     }
     return true;

web/src/components/Broker.js

@@ -14,8 +14,10 @@ class Broker extends Component {
     componentDidMount() {
         let urlParams = new URLSearchParams(this.props.location.search);
         let clusterId = urlParams.get('clusterId');
+        let clusterName = urlParams.get('clusterName');
         this.setState({
             clusterId: clusterId,
+            clusterName: clusterName
         })
         this.loadItems(clusterId);
     }
@@ -74,8 +76,8 @@ class Broker extends Component {
                         onBack={() => {
                             this.props.history.goBack();
                         }}
-                        title={'Broker 管理'}
-                        subTitle={this.state.clusterName}
+                        title={this.state.clusterName}
+                        subTitle={'Broker 管理'}
                     />
                 </div>
 

web/src/components/Cluster.js

@@ -136,6 +136,7 @@ class Cluster extends Component {
                     modalVisible: false
                 });
                 await this.loadTableData(this.state.queryParams);
+                return true;
             }
         } finally {
             this.setState({
@@ -200,7 +201,7 @@ class Cluster extends Component {
             dataIndex: 'topicCount',
             key: 'topicCount',
             render: (topicCount, record, index) => {
-                return <Link to={`/topic?clusterId=${record['id']}&clusterName=${record['name']}`}>
+                return <Link to={`/topic?clusterId=${record['id']}&clusterName=${record['name']}&brokerCount=${record['brokerCount']}`}>
                     {topicCount}
                 </Link>
             }

web/src/components/ClusterModal.js

@@ -26,8 +26,10 @@ const ClusterModal = ({title, handleOk, handleCancel, confirmLoading, model}) =>
                 form
                     .validateFields()
                     .then(values => {
-                        form.resetFields();
-                        handleOk(values);
+                        let success = handleOk(values);
+                        if(success === true){
+                            form.resetFields();
+                        }
                     })
                     .catch(info => {
 

web/src/components/ConsumerGroup.js

@@ -185,8 +185,8 @@ class ConsumerGroup extends Component {
                         onBack={() => {
                             this.props.history.goBack();
                         }}
-                        title={'消费组管理'}
-                        subTitle={this.state.clusterName}
+                        subTitle={'消费组管理'}
+                        title={this.state.clusterName}
                     />
                 </div>
 
@@ -200,7 +200,7 @@ class ConsumerGroup extends Component {
                                 <Space>
                                     <Search
                                         ref={this.inputRefOfName}
-                                        placeholder="GroupID"
+                                        placeholder="group id"
                                         allowClear
                                         onSearch={this.handleSearchByName}
                                     />

web/src/components/ConsumerGroupInfo.js

@@ -46,7 +46,7 @@ class ConsumerGroupInfo extends Component {
                             this.props.history.goBack();
                         }}
                         title={this.state.groupId}
-                        subTitle="详细信息"
+                        subTitle="消费组详情"
                     >
                         <Row>
                             <Space size='large'>

web/src/components/Topic.js

@@ -51,6 +51,7 @@ class Topic extends Component {
         modalVisible: false,
         clusterId: undefined,
         clusterName: undefined,
+        brokerCount: 1,
         selectedRow: {},
         createPartitionConfirmLoading: false
     }
@@ -59,9 +60,11 @@ class Topic extends Component {
         let urlParams = new URLSearchParams(this.props.location.search);
         let clusterId = urlParams.get('clusterId');
         let clusterName = urlParams.get('clusterName');
+        let brokerCount = urlParams.get('brokerCount');
         this.setState({
             clusterId: clusterId,
-            clusterName: clusterName
+            clusterName: clusterName,
+            brokerCount: brokerCount
         })
         let query = {
             ...this.state.queryParams,
@@ -128,6 +131,7 @@ class Topic extends Component {
                     modalVisible: false
                 });
                 await this.loadTableData(this.state.queryParams);
+                return true;
             }
         } finally {
             this.setState({
@@ -282,8 +286,8 @@ class Topic extends Component {
                         onBack={() => {
                             this.props.history.goBack();
                         }}
-                        title={'主题管理'}
-                        subTitle={this.state.clusterName}
+                        subTitle={'主题管理'}
+                        title={this.state.clusterName}
                     />
                 </div>
 
@@ -384,6 +388,7 @@ class Topic extends Component {
                                 handleCancel={this.handleCancelModal}
                                 confirmLoading={this.state.modalConfirmLoading}
                                 model={this.state.model}
+                                brokerCount={this.state.brokerCount}
                             /> : undefined
                     }
 
@@ -417,9 +422,9 @@ class Topic extends Component {
                                })
                            }}>
                         <Form ref={this.form} {...formItemLayout}>
-                            <Form.Item label="总分区数量" name='totalCount' rules={[{required: true}]}>
-                                <InputNumber min={this.state.selectedRow['partitionsSize']}
-                                             placeholder={'不能小于当前分区数量：' + this.state.selectedRow["partitionsSize"]}
+                            <Form.Item label="分区数量" name='totalCount' rules={[{required: true}]}>
+                                <InputNumber min={this.state.selectedRow['partitionsCount']}
+                                             placeholder={'不能小于当前分区数量：' + this.state.selectedRow["partitionsCount"]}
                                              style={{width: '100%'}}/>
                             </Form.Item>
                         </Form>