topic 限流

src/main/java/com/xuxd/kafka/console/beans/dto/TopicThrottleDTO.java

@@ -0,0 +1,21 @@
+package com.xuxd.kafka.console.beans.dto;
+
+import com.xuxd.kafka.console.beans.enums.TopicThrottleSwitch;
+import java.util.List;
+import lombok.Data;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-11-26 15:33:37
+ **/
+@Data
+public class TopicThrottleDTO {
+
+    private String topic;
+
+    private List<Integer> partitions;
+
+    private TopicThrottleSwitch operation;
+}

src/main/java/com/xuxd/kafka/console/beans/enums/TopicThrottleSwitch.java

@@ -0,0 +1,11 @@
+package com.xuxd.kafka.console.beans.enums;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-11-26 15:33:07
+ **/
+public enum TopicThrottleSwitch {
+    ON,OFF;
+}

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

@@ -3,6 +3,7 @@ package com.xuxd.kafka.console.controller;
 import com.xuxd.kafka.console.beans.ReplicaAssignment;
 import com.xuxd.kafka.console.beans.dto.AddPartitionDTO;
 import com.xuxd.kafka.console.beans.dto.NewTopicDTO;
+import com.xuxd.kafka.console.beans.dto.TopicThrottleDTO;
 import com.xuxd.kafka.console.beans.enums.TopicType;
 import com.xuxd.kafka.console.service.TopicService;
 import java.util.ArrayList;
@@ -82,4 +83,9 @@ public class TopicController {
     public Object updateReplicaAssignment(@RequestBody ReplicaAssignment assignment) {
         return topicService.updateReplicaAssignment(assignment);
     }
+
+    @PostMapping("/replica/throttle")
+    public Object configThrottle(@RequestBody TopicThrottleDTO dto) {
+        return topicService.configThrottle(dto.getTopic(), dto.getPartitions(), dto.getOperation());
+    }
 }

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

@@ -2,6 +2,7 @@ package com.xuxd.kafka.console.service;
 
 import com.xuxd.kafka.console.beans.ReplicaAssignment;
 import com.xuxd.kafka.console.beans.ResponseData;
+import com.xuxd.kafka.console.beans.enums.TopicThrottleSwitch;
 import com.xuxd.kafka.console.beans.enums.TopicType;
 import java.util.List;
 import java.util.Map;
@@ -31,4 +32,6 @@ public interface TopicService {
     ResponseData getCurrentReplicaAssignment(String topic);
 
     ResponseData updateReplicaAssignment(ReplicaAssignment assignment);
+
+    ResponseData configThrottle(String topic, List<Integer> partitions,  TopicThrottleSwitch throttleSwitch);
 }

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

@@ -3,6 +3,7 @@ package com.xuxd.kafka.console.service.impl;
 import com.google.gson.Gson;
 import com.xuxd.kafka.console.beans.ReplicaAssignment;
 import com.xuxd.kafka.console.beans.ResponseData;
+import com.xuxd.kafka.console.beans.enums.TopicThrottleSwitch;
 import com.xuxd.kafka.console.beans.enums.TopicType;
 import com.xuxd.kafka.console.beans.vo.TopicDescriptionVO;
 import com.xuxd.kafka.console.beans.vo.TopicPartitionVO;
@@ -148,4 +149,19 @@ public class TopicServiceImpl implements TopicService {
         boolean success = (boolean) tuple2._1();
         return success ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2());
     }
+
+    @Override public ResponseData configThrottle(String topic, List<Integer> partitions, TopicThrottleSwitch throttleSwitch) {
+        Tuple2<Object, String> tuple2 = null;
+        switch (throttleSwitch) {
+            case ON:
+                tuple2 = topicConsole.configThrottle(topic, partitions);
+                break;
+            case OFF:
+                break;
+            default:
+                throw new IllegalArgumentException("switch is unknown.");
+        }
+        boolean success = (boolean) tuple2._1();
+        return success ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2());
+    }
 }

src/main/scala/kafka/console/TopicConsole.scala

@@ -172,7 +172,7 @@ class TopicConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig
 
             if (interBrokerThrottle >= 0) {
                 val moveMap = calculateProposedMoveMap(currentReassignments, proposedParts, currentParts)
-                modifyReassignmentThrottle(adminClient, moveMap, interBrokerThrottle)
+                modifyReassignmentThrottle(adminClient, moveMap)
             }
 
             if (logDirThrottle >= 0) {
@@ -195,6 +195,35 @@ class TopicConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig
         }
     }
 
+    def configThrottle(topic: String, partitions: util.List[Integer]): (Boolean, String) = {
+        withAdminClientAndCatchError(admin => {
+            val throttles = {
+                if (partitions.get(0) == -1) {
+                    Map(topic -> "*")
+                } else {
+                    val topicDescription = admin.describeTopics(Collections.singleton(topic), withTimeoutMs(new DescribeTopicsOptions))
+                        .all().get().values().asScala.toList
+
+                    def convert(partition: Integer, replicas: scala.List[Int]): String = {
+                        replicas.map("%d:%d".format(partition, _)).toSet.mkString(",")
+                    }
+
+                    val ptor = topicDescription.head.partitions().asScala.map(info => (info.partition(), info.replicas().asScala.map(_.id()))).toMap
+                    val conf = partitions.asScala.map(partition => convert(partition, ptor.get(partition) match {
+                        case Some(v) => v.toList
+                        case None => throw new IllegalArgumentException
+                    })).toList
+                    Map(topic -> conf.mkString(","))
+                }
+            }
+            modifyTopicThrottles(admin, throttles, throttles)
+            (true, "")
+        }, e => {
+            log.error("configThrottle error, ", e)
+            (false, e.getMessage)
+        }).asInstanceOf[(Boolean, String)]
+    }
+
     /**
      * Get the current replica assignments for some topics.
      *
@@ -242,12 +271,9 @@ class TopicConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig
         }
     }
 
-    private def modifyReassignmentThrottle(admin: Admin, moveMap: MoveMap, interBrokerThrottle: Long): Unit = {
+    private def modifyReassignmentThrottle(admin: Admin, moveMap: MoveMap): Unit = {
         val leaderThrottles = calculateLeaderThrottles(moveMap)
         val followerThrottles = calculateFollowerThrottles(moveMap)
         modifyTopicThrottles(admin, leaderThrottles, followerThrottles)
-
-//        val reassigningBrokers = calculateReassigningBrokers(moveMap)
-//        modifyInterBrokerThrottle(admin, reassigningBrokers, interBrokerThrottle)
     }
 }

ui/src/utils/api.js

@@ -125,6 +125,10 @@ export const KafkaTopicApi = {
     url: "/topic/replica/assignment",
     method: "post",
   },
+  configThrottle: {
+    url: "/topic/replica/throttle",
+    method: "post",
+  },
 };
 
 export const KafkaConsumerApi = {

ui/src/views/topic/ConfigTopicThrottle.vue

@@ -0,0 +1,156 @@
+<template>
+  <a-modal
+    :title="topic + '限流'"
+    :visible="show"
+    :width="1000"
+    :mask="false"
+    :maskClosable="false"
+    okText="确认"
+    cancelText="取消"
+    :destroyOnClose="true"
+    @cancel="handleCancel"
+    @ok="ok"
+  >
+    <div>
+      <a-spin :spinning="loading">
+        <a-form
+          :form="form"
+          :label-col="{ span: 5 }"
+          :wrapper-col="{ span: 12 }"
+        >
+          <a-form-item label="选择分区">
+            <a-select
+              mode="multiple"
+              option-filter-prop="children"
+              v-decorator="[
+                'partitions',
+                {
+                  initialValue: [-1],
+                  rules: [{ required: true, message: '请选择一个分区!' }],
+                },
+              ]"
+              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-form-item label="操作">
+            <a-radio-group
+              v-decorator="[
+                'operation',
+                {
+                  initialValue: 'ON',
+                  rules: [{ required: true, message: '请选择一个操作!' }],
+                },
+              ]"
+            >
+              <a-radio value="ON"> 开启限流配置 </a-radio>
+              <a-radio value="OFF"> 移除限流配置 </a-radio>
+            </a-radio-group>
+          </a-form-item>
+        </a-form>
+        <hr />
+        <h4>说明：</h4>
+        该限流表示topic的副本的在不同broker之间数据同步占用带宽的限制，该配置是一个topic级别的配置项。如未配置速率，即使配置了这个限流也不会进行实际的限流操作。配置速率在
+        <span style="color: red">运维->配置限流</span> 处进行操作.
+        <h4>如何检查是否对哪些分区启用限流：</h4>
+        topic的限流是通过下面这两项配置的：
+        <ul>
+          <li>leader.replication.throttled.replicas</li>
+          <li>follower.replication.throttled.replicas</li>
+        </ul>
+        只需通过
+        <strong>属性配置</strong>
+        查看这两项配置的值，格式为："0:0,1:0"，左侧为分区，右侧为broker
+        id。示例表示：[分区0的副本：在broker 0上，分区1的副本：在broker 0上]。
+      </a-spin>
+    </div>
+  </a-modal>
+</template>
+
+<script>
+import request from "@/utils/request";
+import { KafkaTopicApi } from "@/utils/api";
+import notification from "ant-design-vue/lib/notification";
+
+export default {
+  name: "ConfigTopicThrottle",
+  props: {
+    visible: {
+      type: Boolean,
+      default: false,
+    },
+    topic: {
+      type: String,
+      default: "",
+    },
+  },
+  data() {
+    return {
+      show: this.visible,
+      loading: false,
+      form: this.$form.createForm(this, { name: "RemoveThrottleForm" }),
+      partitions: [],
+    };
+  },
+  watch: {
+    visible(v) {
+      this.show = v;
+      if (this.show) {
+        this.getPartitionInfo();
+      }
+    },
+  },
+  methods: {
+    handleCancel() {
+      this.$emit("closeThrottleDialog", { refresh: false });
+    },
+    getPartitionInfo() {
+      this.loading = true;
+      request({
+        url: KafkaTopicApi.getPartitionInfo.url + "?topic=" + this.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((e) => e.partition);
+          this.partitions.splice(0, 0, -1);
+        }
+      });
+    },
+    ok() {
+      this.form.validateFields((err, values) => {
+        if (!err) {
+          const data = Object.assign({}, values, {topic: this.topic});
+          this.loading = true;
+          request({
+            url: KafkaTopicApi.configThrottle.url,
+            method: KafkaTopicApi.configThrottle.method,
+            data: data,
+          }).then((res) => {
+            this.loading = false;
+            if (res.code == 0) {
+              this.$message.success(res.msg);
+              this.$emit("closeThrottleDialog", { refresh: false });
+            } else {
+              notification.error({
+                message: "error",
+                description: res.msg,
+              });
+            }
+          });
+        }
+      });
+    },
+  },
+};
+</script>
+
+<style scoped></style>

ui/src/views/topic/Topic.vue

@@ -116,7 +116,7 @@
               size="small"
               href="javascript:;"
               class="operation-btn"
-              @click="openThrottleDialog"
+              @click="openThrottleDialog(record.name)"
               >限流
             </a-button>
           </div>
@@ -152,6 +152,11 @@
           :topic="selectDetail.resourceName"
           @closeUpdateReplicaDialog="closeUpdateReplicaDialog"
         ></UpdateReplica>
+        <ConfigTopicThrottle
+          :visible="showThrottleDialog"
+          :topic="selectDetail.resourceName"
+          @closeThrottleDialog="closeThrottleDialog"
+        ></ConfigTopicThrottle>
       </div>
     </a-spin>
   </div>
@@ -167,6 +172,7 @@ import AddPartition from "@/views/topic/AddPartition";
 import ConsumedDetail from "@/views/topic/ConsumedDetail";
 import TopicConfig from "@/views/topic/TopicConfig";
 import UpdateReplica from "@/views/topic/UpdateReplica";
+import ConfigTopicThrottle from "@/views/topic/ConfigTopicThrottle";
 
 export default {
   name: "Topic",
@@ -177,6 +183,7 @@ export default {
     ConsumedDetail,
     TopicConfig,
     UpdateReplica,
+    ConfigTopicThrottle,
   },
   data() {
     return {
@@ -199,6 +206,7 @@ export default {
       showConsumedDetailDialog: false,
       showTopicConfigDialog: false,
       showUpdateReplicaDialog: false,
+      showThrottleDialog: false,
     };
   },
   methods: {
@@ -289,8 +297,12 @@ export default {
     openMessageStatsDialog() {
       this.$message.info("此功能尚不支持");
     },
-    openThrottleDialog() {
+    openThrottleDialog(topic) {
-      this.$message.info("此功能尚不支持");
+      this.showThrottleDialog = true;
+      this.selectDetail.resourceName = topic;
+    },
+    closeThrottleDialog() {
+      this.showThrottleDialog = false;
     },
   },
   created() {