[Feature]增加Truncate数据功能(didi#1043)

duanxiaoqiu · duanxiaoqiu · commit dddd1b481a4d · 2023-06-27T10:32:21.000+08:00
diff --git a/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/impl/OpTopicManagerImpl.java b/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/impl/OpTopicManagerImpl.java
@@ -10,10 +10,12 @@
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicCreateParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicPartitionExpandParam;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicTruncateParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.partition.Partition;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.Topic;
+import com.xiaojukeji.know.streaming.km.common.constant.KafkaConstant;
 import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
 import com.xiaojukeji.know.streaming.km.common.utils.BackoffUtils;
 import com.xiaojukeji.know.streaming.km.common.utils.FutureUtil;
@@ -159,7 +161,7 @@ public Result<Void> expandTopic(TopicExpansionDTO dto, String operator) {
     @Override
     public Result<Void> truncateTopic(Long clusterPhyId, String topicName, String operator) {
         // 清空Topic
-        Result<Void> rv = opTopicService.truncateTopic(new TopicParam(clusterPhyId, topicName), operator);
+        Result<Void> rv = opTopicService.truncateTopic(new TopicTruncateParam(clusterPhyId, topicName, KafkaConstant.TOPICK_TRUNCATE_DEFAULT_OFFSET), operator);
         if (rv.failed()) {
             return rv;
         }
diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/param/topic/TopicTruncateParam.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/entity/param/topic/TopicTruncateParam.java
@@ -0,0 +1,29 @@
+package com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic;
+
+import com.xiaojukeji.know.streaming.km.common.bean.entity.param.cluster.ClusterPhyParam;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class TopicTruncateParam extends ClusterPhyParam {
+    protected String topicName;
+    protected long offset;
+
+    public TopicTruncateParam(Long clusterPhyId, String topicName, long offset) {
+        super(clusterPhyId);
+        this.topicName = topicName;
+        this.offset = offset;
+    }
+
+    @Override
+    public String toString() {
+        return "TopicParam{" +
+                "clusterPhyId=" + clusterPhyId +
+                ", topicName='" + topicName + '\'' +
+                ", offset='" + offset + '\'' +
+                '}';
+    }
+}
diff --git a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/KafkaConstant.java b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/KafkaConstant.java
@@ -49,6 +49,8 @@ public class KafkaConstant {
 
     public static final Map<String, ConfigDef.ConfigKey> KAFKA_ALL_CONFIG_DEF_MAP = new ConcurrentHashMap<>();
 
+    public static final Integer TOPICK_TRUNCATE_DEFAULT_OFFSET = -1;
+
     static {
         try {
             KAFKA_ALL_CONFIG_DEF_MAP.putAll(CollectionConverters.asJava(LogConfig$.MODULE$.configKeys()));
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/OpTopicService.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/OpTopicService.java
@@ -3,6 +3,7 @@
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicCreateParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicPartitionExpandParam;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicTruncateParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.topic.Topic;
 
@@ -25,5 +26,5 @@ public interface OpTopicService {
     /**
      * 清空topic消息
      */
-    Result<Void> truncateTopic(TopicParam param, String operator);
+    Result<Void> truncateTopic(TopicTruncateParam param, String operator);
 }
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/OpTopicServiceImpl.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/topic/impl/OpTopicServiceImpl.java
@@ -8,6 +8,7 @@
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicCreateParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicPartitionExpandParam;
+import com.xiaojukeji.know.streaming.km.common.bean.entity.param.topic.TopicTruncateParam;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.ResultStatus;
 import com.xiaojukeji.know.streaming.km.common.constant.KafkaConstant;
@@ -208,7 +209,7 @@ public Result<Void> expandTopic(TopicPartitionExpandParam expandParam, String op
     }
 
     @Override
-    public Result<Void> truncateTopic(TopicParam param, String operator) {
+    public Result<Void> truncateTopic(TopicTruncateParam param, String operator) {
         try {
             // 清空topic数据
             Result<Void> rv = (Result<Void>) doVCHandler(param.getClusterPhyId(), TOPIC_TRUNCATE, param);
@@ -233,26 +234,26 @@ public Result<Void> truncateTopic(TopicParam param, String operator) {
     /**************************************************** private method ****************************************************/
 
     private Result<Void> truncateTopicByKafkaClient(VersionItemParam itemParam) {
-        TopicParam param = (TopicParam) itemParam;
+        TopicTruncateParam param = (TopicTruncateParam) itemParam;
         try {
             AdminClient adminClient = kafkaAdminClient.getClient(param.getClusterPhyId());
             //获取topic的分区信息
-            DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(param.getTopicName()));
+            DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(Arrays.asList(param.getTopicName()), new DescribeTopicsOptions().timeoutMs(KafkaConstant.ADMIN_CLIENT_REQUEST_TIME_OUT_UNIT_MS));
             Map<String, TopicDescription> descriptionMap = describeTopicsResult.all().get();
 
             Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();
-            RecordsToDelete recordsToDeleteOffset = RecordsToDelete.beforeOffset(-1);
+            RecordsToDelete recordsToDeleteOffset = RecordsToDelete.beforeOffset(param.getOffset());
 
             descriptionMap.forEach((topicName, topicDescription) -> {
                 for (TopicPartitionInfo topicPartition : topicDescription.partitions()) {
                     recordsToDelete.put(new TopicPartition(topicName, topicPartition.partition()), recordsToDeleteOffset);
                 }
             });
 
-            DeleteRecordsResult deleteRecordsResult = adminClient.deleteRecords(recordsToDelete);
+            DeleteRecordsResult deleteRecordsResult = adminClient.deleteRecords(recordsToDelete, new DeleteRecordsOptions().timeoutMs(KafkaConstant.ADMIN_CLIENT_REQUEST_TIME_OUT_UNIT_MS));
             deleteRecordsResult.all().get();
         } catch (Exception e) {
-            log.error("truncate topic by kafka-client failed，clusterPhyId:{} topicName:{}", param.getClusterPhyId(), param.getTopicName(), e);
+            log.error("truncate topic by kafka-client failed，clusterPhyId:{} topicName:{} offset:{}", param.getClusterPhyId(), param.getTopicName(), param.getOffset(), e);
 
             return Result.buildFromRSAndMsg(ResultStatus.KAFKA_OPERATE_FAILED, e.getMessage());
         }
diff --git a/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/fe/FrontEndControlVersionItems.java b/km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/fe/FrontEndControlVersionItems.java
@@ -36,6 +36,8 @@ public class FrontEndControlVersionItems extends BaseMetricVersionMetric {
     private static final String FE_HA_CREATE_MIRROR_TOPIC                               = "FEHaCreateMirrorTopic";
     private static final String FE_HA_DELETE_MIRROR_TOPIC                               = "FEHaDeleteMirrorTopic";
 
+    private static final String FE_TRUNCATE_TOPIC                                       = "FETruncateTopic";
+
     public FrontEndControlVersionItems(){}
 
     @Override
@@ -89,6 +91,10 @@ public List<VersionControlItem> init(){
         itemList.add(buildItem().minVersion(VersionEnum.V_2_5_0_D_300).maxVersion(VersionEnum.V_2_5_0_D_MAX)
                 .name(FE_HA_DELETE_MIRROR_TOPIC).desc("HA-取消Topic复制"));
 
+        //truncate topic
+        itemList.add(buildItem().minVersion(VersionEnum.V_0_11_0_0).maxVersion(VersionEnum.V_MAX)
+                .name(FE_TRUNCATE_TOPIC).desc("清空topic"));
+
         return itemList;
     }
 }
