Merge pull request #603 from didi/dev

合并开发分支

Author: ZQKC

bin/startup.sh

@@ -9,7 +9,7 @@ error_exit ()
 [ ! -e "$JAVA_HOME/bin/java" ] && unset JAVA_HOME
 
 if [ -z "$JAVA_HOME" ]; then
-  if $darwin; then
+  if [ "Darwin" = "$(uname -s)" ]; then
 
     if [ -x '/usr/libexec/java_home' ] ; then
       export JAVA_HOME=`/usr/libexec/java_home`

docs/install_guide/版本升级手册.md

@@ -40,8 +40,7 @@ thread-pool:
 
 ```
 
-
-**SQL变更**
+**SQL 变更**
 
 ```sql
 -- 多集群管理权限2022-09-06新增
@@ -80,12 +79,11 @@ ALTER TABLE `logi_security_oplog`
 
 ### 6.2.2、升级至 `v3.0.0-beta.1`版本
 
-
-**SQL变更**
+**SQL 变更**
 
 1、在`ks_km_broker`表增加了一个监听信息字段。
-2、为`logi_security_oplog`表operation_methods字段设置默认值''。
-因此需要执行下面的sql对数据库表进行更新。
+2、为`logi_security_oplog`表 operation_methods 字段设置默认值''。
+因此需要执行下面的 sql 对数据库表进行更新。
 
 ```sql
 ALTER TABLE `ks_km_broker`
@@ -98,7 +96,6 @@ ALTER COLUMN `operation_methods` set default '';
 
 ---
 
-
 ### 6.2.3、`2.x`版本 升级至 `v3.0.0-beta.0`版本
 
 **升级步骤：**
@@ -123,14 +120,14 @@ ALTER COLUMN `operation_methods` set default '';
 UPDATE ks_km_topic
     INNER JOIN
     (SELECT
-        topic.cluster_id AS cluster_id,
-        topic.topic_name AS topic_name,
-        topic.description AS description
+    topic.cluster_id AS cluster_id,
+    topic.topic_name AS topic_name,
+    topic.description AS description
     FROM topic WHERE description != ''
     ) AS t
 
-    ON ks_km_topic.cluster_phy_id = t.cluster_id
-        AND ks_km_topic.topic_name = t.topic_name
-        AND ks_km_topic.id > 0
-SET ks_km_topic.description = t.description;
+ON ks_km_topic.cluster_phy_id = t.cluster_id
+    AND ks_km_topic.topic_name = t.topic_name
+    AND ks_km_topic.id > 0
+    SET ks_km_topic.description = t.description;
 ```

docs/user_guide/用户使用手册.md

@@ -11,7 +11,7 @@
 
 下面是用户第一次使用我们产品的典型体验路径：
 
-![text](http://img-ys011.didistatic.com/static/dc2img/do1_YehqxqmsVaqU5gf3XphI)
+![text](http://img-ys011.didistatic.com/static/dc2img/do1_qgqPsAY46sZeBaPUCwXY)
 
 ## 5.3、常用功能
 

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/group/impl/GroupManagerImpl.java

@@ -19,7 +19,7 @@
 import com.xiaojukeji.know.streaming.km.common.bean.vo.group.GroupTopicOverviewVO;
 import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
 import com.xiaojukeji.know.streaming.km.common.enums.AggTypeEnum;
-import com.xiaojukeji.know.streaming.km.common.enums.GroupOffsetResetEnum;
+import com.xiaojukeji.know.streaming.km.common.enums.OffsetTypeEnum;
 import com.xiaojukeji.know.streaming.km.common.enums.group.GroupStateEnum;
 import com.xiaojukeji.know.streaming.km.common.exception.AdminOperateException;
 import com.xiaojukeji.know.streaming.km.common.exception.NotExistException;
@@ -199,12 +199,12 @@ private Result<Void> checkFieldLegal(GroupOffsetResetDTO dto) {
             return Result.buildFromRSAndMsg(ResultStatus.NOT_EXIST, MsgConstant.getTopicNotExist(dto.getClusterId(), dto.getTopicName()));
         }
 
-        if (GroupOffsetResetEnum.PRECISE_OFFSET.getResetType() == dto.getResetType()
+        if (OffsetTypeEnum.PRECISE_OFFSET.getResetType() == dto.getResetType()
             && ValidateUtils.isEmptyList(dto.getOffsetList())) {
             return Result.buildFromRSAndMsg(ResultStatus.PARAM_ILLEGAL, "参数错误，指定offset重置需传offset信息");
         }
 
-        if (GroupOffsetResetEnum.PRECISE_TIMESTAMP.getResetType() == dto.getResetType()
+        if (OffsetTypeEnum.PRECISE_TIMESTAMP.getResetType() == dto.getResetType()
                 && ValidateUtils.isNull(dto.getTimestamp())) {
             return Result.buildFromRSAndMsg(ResultStatus.PARAM_ILLEGAL, "参数错误，指定时间重置需传时间信息");
         }
@@ -213,7 +213,7 @@ private Result<Void> checkFieldLegal(GroupOffsetResetDTO dto) {
     }
 
     private Result<Map<TopicPartition, Long>> getPartitionOffset(GroupOffsetResetDTO dto) {
-        if (GroupOffsetResetEnum.PRECISE_OFFSET.getResetType() == dto.getResetType()) {
+        if (OffsetTypeEnum.PRECISE_OFFSET.getResetType() == dto.getResetType()) {
             return Result.buildSuc(dto.getOffsetList().stream().collect(Collectors.toMap(
                     elem -> new TopicPartition(dto.getTopicName(), elem.getPartitionId()),
                     PartitionOffsetDTO::getOffset,
@@ -222,9 +222,9 @@ private Result<Map<TopicPartition, Long>> getPartitionOffset(GroupOffsetResetDTO
         }
 
         OffsetSpec offsetSpec = null;
-        if (GroupOffsetResetEnum.PRECISE_TIMESTAMP.getResetType() == dto.getResetType()) {
+        if (OffsetTypeEnum.PRECISE_TIMESTAMP.getResetType() == dto.getResetType()) {
             offsetSpec = OffsetSpec.forTimestamp(dto.getTimestamp());
-        } else if (GroupOffsetResetEnum.EARLIEST.getResetType() == dto.getResetType()) {
+        } else if (OffsetTypeEnum.EARLIEST.getResetType() == dto.getResetType()) {
             offsetSpec = OffsetSpec.earliest();
         } else {
             offsetSpec = OffsetSpec.latest();

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/TopicStateManager.java

@@ -1,5 +1,6 @@
 package com.xiaojukeji.know.streaming.km.biz.topic;
 
+import com.xiaojukeji.know.streaming.km.common.bean.dto.pagination.PaginationSortDTO;
 import com.xiaojukeji.know.streaming.km.common.bean.dto.topic.TopicRecordDTO;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.result.Result;
 import com.xiaojukeji.know.streaming.km.common.bean.vo.topic.TopicBrokersPartitionsSummaryVO;

km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/impl/TopicStateManagerImpl.java

@@ -22,25 +22,26 @@
 import com.xiaojukeji.know.streaming.km.common.constant.Constant;
 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.converter.PartitionConverter;
 import com.xiaojukeji.know.streaming.km.common.converter.TopicVOConverter;
+import com.xiaojukeji.know.streaming.km.common.enums.OffsetTypeEnum;
+import com.xiaojukeji.know.streaming.km.common.enums.SortTypeEnum;
 import com.xiaojukeji.know.streaming.km.common.exception.AdminOperateException;
 import com.xiaojukeji.know.streaming.km.common.exception.NotExistException;
 import com.xiaojukeji.know.streaming.km.common.utils.ConvertUtil;
+import com.xiaojukeji.know.streaming.km.common.utils.PaginationUtil;
 import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
 import com.xiaojukeji.know.streaming.km.core.service.broker.BrokerService;
 import com.xiaojukeji.know.streaming.km.core.service.cluster.ClusterPhyService;
 import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionMetricService;
-import com.xiaojukeji.know.streaming.km.core.service.topic.TopicConfigService;
 import com.xiaojukeji.know.streaming.km.core.service.partition.PartitionService;
+import com.xiaojukeji.know.streaming.km.core.service.topic.TopicConfigService;
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicMetricService;
 import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
 import com.xiaojukeji.know.streaming.km.core.service.version.metrics.TopicMetricVersionItems;
+import org.apache.commons.lang3.ObjectUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.admin.OffsetSpec;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.*;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.config.TopicConfig;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -160,8 +161,31 @@ public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String to
             }
             maxMessage = Math.min(maxMessage, dto.getMaxRecords());
             kafkaConsumer.assign(partitionList);
+
+            Map<TopicPartition, OffsetAndTimestamp> partitionOffsetAndTimestampMap = new HashMap<>();
+            // 获取指定时间每个分区的offset（按指定开始时间查询消息时）
+            if (OffsetTypeEnum.PRECISE_TIMESTAMP.getResetType() == dto.getFilterOffsetReset()) {
+                Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
+                partitionList.forEach(topicPartition -> {
+                    timestampsToSearch.put(topicPartition, dto.getStartTimestampUnitMs());
+                });
+                partitionOffsetAndTimestampMap = kafkaConsumer.offsetsForTimes(timestampsToSearch);
+            }
+
             for (TopicPartition partition : partitionList) {
-                kafkaConsumer.seek(partition, Math.max(beginOffsetsMapResult.getData().get(partition), endOffsetsMapResult.getData().get(partition) - dto.getMaxRecords()));
+                if (OffsetTypeEnum.EARLIEST.getResetType() == dto.getFilterOffsetReset()) {
+                    // 重置到最旧
+                    kafkaConsumer.seek(partition, beginOffsetsMapResult.getData().get(partition));
+                } else if (OffsetTypeEnum.PRECISE_TIMESTAMP.getResetType() == dto.getFilterOffsetReset()) {
+                    // 重置到指定时间
+                    kafkaConsumer.seek(partition, partitionOffsetAndTimestampMap.get(partition).offset());
+                } else if (OffsetTypeEnum.PRECISE_OFFSET.getResetType() == dto.getFilterOffsetReset()) {
+                    // 重置到指定位置
+
+                } else {
+                    // 默认，重置到最新
+                    kafkaConsumer.seek(partition, Math.max(beginOffsetsMapResult.getData().get(partition), endOffsetsMapResult.getData().get(partition) - dto.getMaxRecords()));
+                }
             }
 
             // 这里需要减去 KafkaConstant.POLL_ONCE_TIMEOUT_UNIT_MS 是因为poll一次需要耗时，如果这里不减去，则可能会导致poll之后，超过要求的时间
@@ -185,6 +209,15 @@ public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String to
                     }
             }
 
+            // 排序
+            if (ObjectUtils.isNotEmpty(voList)) {
+                // 默认按时间倒序排序
+                if (StringUtils.isBlank(dto.getSortType())) {
+                    dto.setSortType(SortTypeEnum.DESC.getSortType());
+                }
+                PaginationUtil.pageBySort(voList, dto.getSortField(), dto.getSortType());
+            }
+
             return Result.buildSuc(voList.subList(0, Math.min(dto.getMaxRecords(), voList.size())));
         } catch (Exception e) {
             log.error("method=getTopicMessages||clusterPhyId={}||topicName={}||param={}||errMsg=exception", clusterPhyId, topicName, dto, e);

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/dto/group/GroupOffsetResetDTO.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.xiaojukeji.know.streaming.km.common.bean.dto.partition.PartitionOffsetDTO;
 import com.xiaojukeji.know.streaming.km.common.bean.dto.topic.ClusterTopicDTO;
+import com.xiaojukeji.know.streaming.km.common.enums.OffsetTypeEnum;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
@@ -23,7 +24,7 @@ public class GroupOffsetResetDTO extends ClusterTopicDTO {
     private String groupName;
 
     /**
-     * @see com.xiaojukeji.know.streaming.km.common.enums.GroupOffsetResetEnum
+     * @see OffsetTypeEnum
      */
     @NotNull(message = "resetType不允许为空")
     @ApiModelProperty(value = "重置方式", example = "1")

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/dto/topic/TopicRecordDTO.java

@@ -1,7 +1,8 @@
 package com.xiaojukeji.know.streaming.km.common.bean.dto.topic;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.xiaojukeji.know.streaming.km.common.bean.dto.BaseDTO;
+import com.xiaojukeji.know.streaming.km.common.bean.dto.pagination.PaginationSortDTO;
+import com.xiaojukeji.know.streaming.km.common.enums.OffsetTypeEnum;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
@@ -15,7 +16,7 @@
 @Data
 @JsonIgnoreProperties(ignoreUnknown = true)
 @ApiModel(description = "Topic记录")
-public class TopicRecordDTO extends BaseDTO {
+public class TopicRecordDTO extends PaginationSortDTO {
     @NotNull(message = "truncate不允许为空")
     @ApiModelProperty(value = "是否截断", example = "true")
     private Boolean truncate;
@@ -34,4 +35,13 @@ public class TopicRecordDTO extends BaseDTO {
 
     @ApiModelProperty(value = "预览超时时间", example = "10000")
     private Long pullTimeoutUnitMs = 8000L;
+
+    /**
+     * @see OffsetTypeEnum
+     */
+    @ApiModelProperty(value = "offset", example = "")
+    private Integer filterOffsetReset = 0;
+
+    @ApiModelProperty(value = "开始日期时间戳", example = "")
+    private Long startTimestampUnitMs;
 }

km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/constant/Constant.java

@@ -64,4 +64,5 @@ private Constant() {}
     public static final Float COLLECT_METRICS_ERROR_COST_TIME = -1.0F;
 
     public static final Integer DEFAULT_RETRY_TIME = 3;
+
 }

…m/common/enums/GroupOffsetResetEnum.java …ming/km/common/enums/OffsetTypeEnum.javakm-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enums/GroupOffsetResetEnum.java renamed to km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enums/OffsetTypeEnum.java km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enums/GroupOffsetResetEnum.java renamed to km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enums/OffsetTypeEnum.java

@@ -3,27 +3,27 @@
 import lombok.Getter;
 
 /**
- * 重置offset
+ * offset类型
  * @author zengqiao
  * @date 19/4/8
  */
 @Getter
-public enum GroupOffsetResetEnum {
-    LATEST(0, "重置到最新"),
+public enum OffsetTypeEnum {
+    LATEST(0, "最新"),
 
-    EARLIEST(1, "重置到最旧"),
+    EARLIEST(1, "最旧"),
 
-    PRECISE_TIMESTAMP(2, "按时间进行重置"),
+    PRECISE_TIMESTAMP(2, "指定时间"),
 
-    PRECISE_OFFSET(3, "重置到指定位置"),
+    PRECISE_OFFSET(3, "指定位置"),
 
     ;
 
     private final int resetType;
 
     private final String message;
 
-    GroupOffsetResetEnum(int resetType, String message) {
+    OffsetTypeEnum(int resetType, String message) {
         this.resetType = resetType;
         this.message = message;
     }