支持在线删除消息

2022-07-04 17:16:00 +08:00
parent b163e5f776
commit 979859b232
7 changed files with 248 additions and 7 deletions
--- a/src/main/java/com/xuxd/kafka/console/controller/MessageController.java
+++ b/src/main/java/com/xuxd/kafka/console/controller/MessageController.java
@@ -1,14 +1,15 @@
 package com.xuxd.kafka.console.controller;

+import com.xuxd.kafka.console.beans.QueryMessage;
+import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.SendMessage;
 import com.xuxd.kafka.console.beans.dto.QueryMessageDTO;
 import com.xuxd.kafka.console.service.MessageService;
+import org.apache.commons.collections.CollectionUtils;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;

 /**
 * kafka-console-ui.
@@ -52,4 +53,12 @@ public class MessageController {
    public Object resend(@RequestBody SendMessage message) {
        return messageService.resend(message);
    }
+
+    @DeleteMapping
+    public Object delete(@RequestBody List<QueryMessage> messages) {
+        if (CollectionUtils.isEmpty(messages)) {
+            return ResponseData.create().failed("params is null");
+        }
+        return messageService.delete(messages);
+    }
 }
--- a/src/main/java/com/xuxd/kafka/console/service/MessageService.java
+++ b/src/main/java/com/xuxd/kafka/console/service/MessageService.java
@@ -4,6 +4,8 @@ import com.xuxd.kafka.console.beans.QueryMessage;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.SendMessage;

+import java.util.List;
+
 /**
 * kafka-console-ui.
 *
@@ -23,4 +25,6 @@ public interface MessageService {
    ResponseData send(SendMessage message);

    ResponseData resend(SendMessage message);
+
+    ResponseData delete(List<QueryMessage> messages);
 }
--- a/src/main/java/com/xuxd/kafka/console/service/impl/MessageServiceImpl.java
+++ b/src/main/java/com/xuxd/kafka/console/service/impl/MessageServiceImpl.java
@@ -24,6 +24,7 @@ import kafka.console.TopicConsole;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.admin.RecordsToDelete;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
@@ -242,6 +243,18 @@ public class MessageServiceImpl implements MessageService, ApplicationContextAwa
        return success ? ResponseData.create().success("success: " + tuple2._2()) : ResponseData.create().failed(tuple2._2());
    }

+    @Override
+    public ResponseData delete(List<QueryMessage> messages) {
+        Map<TopicPartition, RecordsToDelete> params = new HashMap<>(messages.size(), 1f);
+
+        messages.forEach(message -> {
+            params.put(new TopicPartition(message.getTopic(), message.getPartition()), RecordsToDelete.beforeOffset(message.getOffset()));
+        });
+        Tuple2<Object, String> tuple2 = messageConsole.delete(params);
+        boolean success = (boolean) tuple2._1();
+        return success ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2());
+    }
+
    private Map<TopicPartition, ConsumerRecord<byte[], byte[]>> searchRecordByOffset(QueryMessage queryMessage) {
        Set<TopicPartition> partitions = getPartitions(queryMessage);

--- a/src/main/scala/kafka/console/MessageConsole.scala
+++ b/src/main/scala/kafka/console/MessageConsole.scala
@@ -4,13 +4,14 @@ import com.xuxd.kafka.console.beans.MessageFilter
 import com.xuxd.kafka.console.beans.enums.FilterType
 import com.xuxd.kafka.console.config.{ContextConfigHolder, KafkaConfig}
 import org.apache.commons.lang3.StringUtils
+import org.apache.kafka.clients.admin.{DeleteRecordsOptions, RecordsToDelete}
 import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
 import org.apache.kafka.clients.producer.ProducerRecord
 import org.apache.kafka.common.TopicPartition

 import java.time.Duration
 import java.util
-import java.util.Properties
+import java.util.{Properties}
 import scala.collection.immutable
 import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsScala, SeqHasAsJava}

@@ -236,4 +237,14 @@ class MessageConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConf
            (false, e.getMessage)
        }).asInstanceOf[(Boolean, String)]
    }
+
+    def delete(recordsToDelete: util.Map[TopicPartition, RecordsToDelete]): (Boolean, String) = {
+      withAdminClientAndCatchError(admin => {
+        admin.deleteRecords(recordsToDelete, withTimeoutMs(new DeleteRecordsOptions())).all().get()
+        (true, "")
+      }, e => {
+        log.error("delete message error.", e)
+        (false, "delete error :" + e.getMessage)
+      }).asInstanceOf[(Boolean, String)]
+    }
 }
--- a/ui/src/utils/api.js
+++ b/ui/src/utils/api.js
@@ -276,4 +276,8 @@ export const KafkaMessageApi = {
    url: "/message/resend",
    method: "post",
  },
+  delete: {
+    url: "/message",
+    method: "delete",
+  },
 };
--- a/ui/src/views/message/DeleteMessage.vue
+++ b/ui/src/views/message/DeleteMessage.vue
@@ -0,0 +1,196 @@
+<template>
+  <div class="tab-content">
+    <a-spin :spinning="loading">
+      <div>
+        <h4 class="hint-content">
+          注意：以下删除，将删除该分区比该偏移位点小的所有消息（不包含该位点）
+        </h4>
+        <hr />
+      </div>
+      <div id="search-offset-form-advanced-search">
+        <a-form
+          class="ant-advanced-search-form"
+          :form="form"
+          @submit="handleSearch"
+        >
+          <a-row :gutter="24">
+            <a-col :span="9">
+              <a-form-item label="topic">
+                <a-select
+                  class="topic-select"
+                  @change="handleTopicChange"
+                  show-search
+                  option-filter-prop="children"
+                  v-decorator="[
+                    'topic',
+                    {
+                      rules: [{ required: true, message: '请选择一个topic!' }],
+                    },
+                  ]"
+                  placeholder="请选择一个topic"
+                >
+                  <a-select-option v-for="v in topicList" :key="v" :value="v">
+                    {{ v }}
+                  </a-select-option>
+                </a-select>
+              </a-form-item>
+            </a-col>
+            <a-col :span="6">
+              <a-form-item label="分区">
+                <a-select
+                  class="type-select"
+                  show-search
+                  option-filter-prop="children"
+                  v-model="selectPartition"
+                  placeholder="请选择一个分区"
+                >
+                  <a-select-option v-for="v in partitions" :key="v" :value="v">
+                    <span v-if="v == -1">全部</span> <span v-else>{{ v }}</span>
+                  </a-select-option>
+                </a-select>
+              </a-form-item>
+            </a-col>
+            <a-col :span="7">
+              <a-form-item label="偏移">
+                <a-input
+                  v-decorator="[
+                    'offset',
+                    {
+                      rules: [{ required: true, message: '请输入消息偏移!' }],
+                    },
+                  ]"
+                  placeholder="消息偏移"
+                />
+              </a-form-item>
+            </a-col>
+            <a-col :span="2" :style="{ textAlign: 'right' }">
+              <a-form-item>
+                <a-button type="primary" html-type="submit"> 执行删除</a-button>
+              </a-form-item>
+            </a-col>
+          </a-row>
+        </a-form>
+      </div>
+    </a-spin>
+  </div>
+</template>
+
+<script>
+import request from "@/utils/request";
+import { KafkaMessageApi, KafkaTopicApi } from "@/utils/api";
+import notification from "ant-design-vue/lib/notification";
+
+export default {
+  name: "DeleteMessage",
+  props: {
+    topicList: {
+      type: Array,
+    },
+  },
+  data() {
+    return {
+      loading: false,
+      form: this.$form.createForm(this, { name: "message_search_offset" }),
+      partitions: [],
+      selectPartition: undefined,
+    };
+  },
+  methods: {
+    handleSearch(e) {
+      e.preventDefault();
+      this.form.validateFields((err, values) => {
+        if (!err) {
+          const data = Object.assign({}, values, {
+            partition: this.selectPartition,
+          });
+          this.loading = true;
+          request({
+            url: KafkaMessageApi.delete.url,
+            method: KafkaMessageApi.delete.method,
+            data: [data],
+          }).then((res) => {
+            this.loading = false;
+            if (res.code == 0) {
+              this.$message.success(res.msg);
+            } else {
+              notification.error({
+                message: "error",
+                description: res.msg,
+              });
+            }
+          });
+        }
+      });
+    },
+    getPartitionInfo(topic) {
+      this.loading = true;
+      request({
+        url: KafkaTopicApi.getPartitionInfo.url + "?topic=" + topic,
+        method: KafkaTopicApi.getPartitionInfo.method,
+      }).then((res) => {
+        this.loading = false;
+        if (res.code != 0) {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        } else {
+          this.partitions = res.data.map((v) => v.partition);
+        }
+      });
+    },
+    handleTopicChange(topic) {
+      this.selectPartition =
+        this.partitions.length > 0 ? this.partitions[0] : 0;
+      this.getPartitionInfo(topic);
+    },
+  },
+};
+</script>
+
+<style scoped>
+.tab-content {
+  width: 100%;
+  height: 100%;
+}
+
+.ant-advanced-search-form {
+  padding: 24px;
+  background: #fbfbfb;
+  border: 1px solid #d9d9d9;
+  border-radius: 6px;
+}
+
+.ant-advanced-search-form .ant-form-item {
+  display: flex;
+}
+
+.ant-advanced-search-form .ant-form-item-control-wrapper {
+  flex: 1;
+}
+
+#components-form-topic-advanced-search .ant-form {
+  max-width: none;
+  margin-bottom: 1%;
+}
+
+#search-offset-form-advanced-search .search-result-list {
+  margin-top: 16px;
+  border: 1px dashed #e9e9e9;
+  border-radius: 6px;
+  background-color: #fafafa;
+  min-height: 200px;
+  text-align: center;
+  padding-top: 80px;
+}
+.topic-select {
+  width: 400px !important;
+}
+.type-select {
+  width: 200px !important;
+}
+
+.hint-content {
+  color: red;
+}
+</style>
--- a/ui/src/views/message/Message.vue
+++ b/ui/src/views/message/Message.vue
@@ -11,6 +11,9 @@
        <a-tab-pane key="3" tab="在线发送">
          <SendMessage :topic-list="topicList"></SendMessage>
        </a-tab-pane>
+        <a-tab-pane key="4" tab="在线删除">
+          <DeleteMessage :topic-list="topicList"></DeleteMessage>
+        </a-tab-pane>
      </a-tabs>
    </a-spin>
  </div>
@@ -23,9 +26,10 @@ import request from "@/utils/request";
 import { KafkaTopicApi } from "@/utils/api";
 import notification from "ant-design-vue/lib/notification";
 import SendMessage from "@/views/message/SendMessage";
+import DeleteMessage from "./DeleteMessage";
 export default {
  name: "Message",
-  components: { SearchByTime, SearchByOffset, SendMessage },
+  components: { DeleteMessage, SearchByTime, SearchByOffset, SendMessage },
  data() {
    return {
      loading: false,