集群同步-》同步消费位点

src/main/java/com/xuxd/kafka/console/controller/ConsumerController.java

@@ -106,4 +106,9 @@ public class ConsumerController {
 
         return res;
     }
+
+    @GetMapping("/group/id/list")
+    public Object getGroupIdList() {
+        return consumerService.getGroupIdList();
+    }
 }

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

@@ -28,4 +28,6 @@ public interface ConsumerService {
     ResponseData resetOffsetToEndpoint(String groupId, String topic, OffsetResetStrategy strategy);
 
     ResponseData resetPartitionToTargetOffset(String groupId, TopicPartition partition, long offset);
+
+    ResponseData getGroupIdList();
 }

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

@@ -133,4 +133,9 @@ public class ConsumerServiceImpl implements ConsumerService {
         Tuple2<Object, String> tuple2 = consumerConsole.resetPartitionToTargetOffset(groupId, partition, offset);
         return (boolean) tuple2._1() ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2());
     }
+
+    @Override public ResponseData getGroupIdList() {
+        Set<String> stateGroup = consumerConsole.getConsumerGroupIdList(null);
+        return ResponseData.create().data(stateGroup).success();
+    }
 }

ui/src/utils/api.js

@@ -108,6 +108,10 @@ export const KafkaConsumerApi = {
     url: "/consumer/reset/offset",
     method: "post",
   },
+  getGroupIdList: {
+    url: "/consumer/group/id/list",
+    method: "get",
+  },
 };
 
 export const KafkaClusterApi = {

ui/src/views/op/Operation.vue

@@ -3,7 +3,9 @@
     <div class="content-module">
       <a-card title="数据同步" style="width: 100%; text-align: left">
         <p>
-          <a-button type="primary"> 同步消费位点 </a-button>
+          <a-button type="primary" @click="openSyncConsumerOffsetDialog">
+            同步消费位点
+          </a-button>
           <label>说明：</label>
           <span
             >同步其它集群中指定消费组与订阅的topic的消费位点到当前集群上，该消费组在当前集群已存在，且双方订阅的topic分区信息一致</span
@@ -20,13 +22,34 @@
         </p>
       </a-card>
     </div>
+    <SyncConsumerOffset
+      :visible="syncData.showSyncConsumerOffsetDialog"
+      @closeSyncConsumerOffsetDialog="closeSyncConsumerOffsetDialog"
+    >
+    </SyncConsumerOffset>
   </div>
 </template>
 
 <script>
+import SyncConsumerOffset from "@/views/op/SyncConsumerOffset";
 export default {
   name: "Operation",
-  components: {},
+  components: { SyncConsumerOffset },
+  data() {
+    return {
+      syncData: {
+        showSyncConsumerOffsetDialog: false,
+      },
+    };
+  },
+  methods: {
+    openSyncConsumerOffsetDialog() {
+      this.syncData.showSyncConsumerOffsetDialog = true;
+    },
+    closeSyncConsumerOffsetDialog() {
+      this.syncData.showSyncConsumerOffsetDialog = false;
+    },
+  },
 };
 </script>
 

ui/src/views/op/SyncConsumerOffset.vue

@@ -0,0 +1,153 @@
+<template>
+  <a-modal
+    title="同步消费位点"
+    :visible="show"
+    :width="800"
+    :mask="false"
+    :destroyOnClose="true"
+    :footer="null"
+    :maskClosable="false"
+    @cancel="handleCancel"
+  >
+    <div>
+      <a-spin :spinning="loading">
+        <a-form
+          :form="form"
+          :label-col="{ span: 5 }"
+          :wrapper-col="{ span: 12 }"
+          @submit="handleSubmit"
+        >
+          <a-form-item label="消费组">
+            <a-select
+              show-search
+              option-filter-prop="children"
+              v-decorator="[
+                'groupId',
+                { rules: [{ required: true, message: '请选择一个消费组!' }] },
+              ]"
+              placeholder="请选择一个消费组"
+            >
+              <a-select-option v-for="v in groupIdList" :key="v" :value="v">
+                {{ v }}
+              </a-select-option>
+            </a-select>
+          </a-form-item>
+          <a-form-item label="topic">
+            <a-select
+              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-form-item label="kafka地址">
+            <a-input
+              v-decorator="[
+                'address',
+                {
+                  rules: [{ required: true, message: '输入待同步kafka地址!' }],
+                },
+              ]"
+              placeholder="输入待同步kafka地址"
+            />
+          </a-form-item>
+
+          <a-form-item :wrapper-col="{ span: 12, offset: 5 }">
+            <a-button type="primary" html-type="submit"> 提交 </a-button>
+          </a-form-item>
+        </a-form>
+      </a-spin>
+    </div>
+  </a-modal>
+</template>
+
+<script>
+import request from "@/utils/request";
+import { KafkaConsumerApi } from "@/utils/api";
+import notification from "ant-design-vue/es/notification";
+export default {
+  name: "SyncConsumerOffset",
+  props: {
+    topic: {
+      type: String,
+      default: "",
+    },
+    visible: {
+      type: Boolean,
+      default: false,
+    },
+  },
+  data() {
+    return {
+      show: this.visible,
+      data: [],
+      loading: false,
+      form: this.$form.createForm(this, { name: "coordinated" }),
+      topicList: [],
+      groupIdList: [],
+    };
+  },
+  watch: {
+    visible(v) {
+      this.show = v;
+      if (this.show) {
+        this.getGroupIdList();
+      }
+    },
+  },
+  methods: {
+    getGroupIdList() {
+      request({
+        url: KafkaConsumerApi.getGroupIdList.url,
+        method: KafkaConsumerApi.getGroupIdList.method,
+      }).then((res) => {
+        if (res.code == 0) {
+          this.groupIdList = res.data;
+        } else {
+          notification.error({
+            message: "error",
+            description: res.msg,
+          });
+        }
+      });
+    },
+    handleSubmit(e) {
+      e.preventDefault();
+      this.form.validateFields((err, values) => {
+        if (!err) {
+          this.loading = true;
+          request({
+            url: KafkaConsumerApi.addSubscription.url,
+            method: KafkaConsumerApi.addSubscription.method,
+            data: values,
+          }).then((res) => {
+            this.loading = false;
+            if (res.code == 0) {
+              this.$message.success(res.msg);
+              this.$emit("closeSyncConsumerOffsetDialog", { refresh: true });
+            } else {
+              notification.error({
+                message: "error",
+                description: res.msg,
+              });
+            }
+          });
+        }
+      });
+    },
+    handleCancel() {
+      this.data = [];
+      this.$emit("closeSyncConsumerOffsetDialog", { refresh: false });
+    },
+  },
+};
+</script>
+
+<style scoped></style>