Topic消息查询支持Timestamp排序，接口支持按指定日期查询

2026-01-01 17:52:11 +08:00 · 2022-09-09 15:45:31 +08:00
parent 8e50d145d5
commit 0e49002f42
6 changed files with 50 additions and 21 deletions
--- a/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/TopicStateManager.java
+++ b/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/TopicStateManager.java
@@ -16,7 +16,7 @@ import java.util.List;
 public interface TopicStateManager {
    TopicBrokerAllVO getTopicBrokerAll(Long clusterPhyId, String topicName, String searchBrokerHost) throws NotExistException;

-    Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String topicName, TopicRecordDTO dto, PaginationSortDTO sortDto) throws AdminOperateException;
+    Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String topicName, TopicRecordDTO dto) throws AdminOperateException;

    Result<TopicStateVO> getTopicState(Long clusterPhyId, String topicName);

--- a/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/impl/TopicStateManagerImpl.java
+++ b/km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/impl/TopicStateManagerImpl.java
@@ -3,7 +3,6 @@ package com.xiaojukeji.know.streaming.km.biz.topic.impl;
 import com.didiglobal.logi.log.ILog;
 import com.didiglobal.logi.log.LogFactory;
 import com.xiaojukeji.know.streaming.km.biz.topic.TopicStateManager;
-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.broker.Broker;
 import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
@@ -40,10 +39,7 @@ 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.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;
@@ -123,7 +119,7 @@ public class TopicStateManagerImpl implements TopicStateManager {
    }

    @Override
-    public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String topicName, TopicRecordDTO dto, PaginationSortDTO sortDto) throws AdminOperateException {
+    public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String topicName, TopicRecordDTO dto) throws AdminOperateException {
        long startTime = System.currentTimeMillis();

        // 获取集群
@@ -163,10 +159,29 @@ public class TopicStateManagerImpl implements TopicStateManager {
            }
            maxMessage = Math.min(maxMessage, dto.getMaxRecords());
            kafkaConsumer.assign(partitionList);
+
+            Map<TopicPartition, OffsetAndTimestamp> partitionOffsetAndTimestampMap = new HashMap<>();
+            // 获取指定时间每个分区的offset（按指定开始时间查询消息时）
+            if (GroupOffsetResetEnum.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) {
                if (GroupOffsetResetEnum.EARLIEST.getResetType() == dto.getFilterOffsetReset()) {
+                    // 重置到最旧
                    kafkaConsumer.seek(partition, beginOffsetsMapResult.getData().get(partition));
+                } else if (GroupOffsetResetEnum.PRECISE_TIMESTAMP.getResetType() == dto.getFilterOffsetReset()) {
+                    // 重置到指定时间
+                    kafkaConsumer.seek(partition, partitionOffsetAndTimestampMap.get(partition).offset());
+                } else if (GroupOffsetResetEnum.PRECISE_OFFSET.getResetType() == dto.getFilterOffsetReset()) {
+                    // 重置到指定位置
+
                } else {
+                    // 默认，重置到最新
                    kafkaConsumer.seek(partition, Math.max(beginOffsetsMapResult.getData().get(partition), endOffsetsMapResult.getData().get(partition) - dto.getMaxRecords()));
                }
            }
@@ -194,7 +209,7 @@ public class TopicStateManagerImpl implements TopicStateManager {

            // 排序
            if (ObjectUtils.isNotEmpty(voList)) {
-                PaginationUtil.pageBySort(voList, sortDto.getSortField(), sortDto.getSortField());
+                PaginationUtil.pageBySort(voList, dto.getSortField(), dto.getSortType());
            }

            return Result.buildSuc(voList.subList(0, Math.min(dto.getMaxRecords(), voList.size())));
--- a/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/dto/topic/TopicRecordDTO.java
+++ b/km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/dto/topic/TopicRecordDTO.java
@@ -2,6 +2,7 @@ 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 io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
@@ -15,7 +16,7 @@ import javax.validation.constraints.NotNull;
@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;
@@ -38,4 +39,10 @@ public class TopicRecordDTO extends BaseDTO {
    @ApiModelProperty(value = "offset", example = "")
    private Integer filterOffsetReset = 0;

+    @ApiModelProperty(value = "开始日期时间戳", example = "")
+    private Long startTimestampUnitMs;
+
+    @ApiModelProperty(value = "结束日期时间戳", example = "")
+    private Long utilTimestampUnitMs;
+
 }
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/Messages.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/Messages.tsx
@@ -32,8 +32,8 @@ const TopicMessages = (props: any) => {

  // 获取消息开始位置
  const offsetResetList = [
-    { 'label': 'latest', value: '0' },
-    { 'label': 'earliest', value: '1' }
+    { 'label': 'latest', value: 0 },
+    { 'label': 'earliest', value: 1 }
  ];

  // 默认排序
@@ -42,6 +42,8 @@ const TopicMessages = (props: any) => {
    sortType: 'desc',
  };

+  const [sorter, setSorter] = useState<any>(defaultSorter);
+
  // 请求接口获取数据
  const genData = async () => {
    if (urlParams?.clusterId === undefined || hashData?.topicName === undefined) return;
@@ -56,7 +58,7 @@ const TopicMessages = (props: any) => {
      });
      setPartitionIdList(newPartitionIdList || []);
    });
-    request(Api.getTopicMessagesList(hashData?.topicName, urlParams?.clusterId), { data: { ...params, ...defaultSorter }, method: 'POST' })
+    request(Api.getTopicMessagesList(hashData?.topicName, urlParams?.clusterId), { data: { ...params, ...sorter }, method: 'POST' })
      .then((res: any) => {
        // setPagination({
        //   current: res.pagination?.pageNo,
@@ -94,11 +96,16 @@ const TopicMessages = (props: any) => {
    history.push(`/cluster/${urlParams?.clusterId}/testing/consumer`);
  };

-  const onTableChange = (pagination: any, filters: any, sorter: any) => {
-    defaultSorter.sortField = sorter.field || '';
-    defaultSorter.sortType = sorter.order ? sorter.order.substring(0, sorter.order.indexOf('end')) : '';
+  const onTableChange = (pagination: any, filters: any, sorter: any, extra: any) => {
    setPagination(pagination);
-    genData();
+    // 只有排序事件时，触发重新请求后端数据
+    if(extra.action === 'sort') {
+      setSorter({
+        sortField: sorter.field || '',
+        sortType: sorter.order ? sorter.order.substring(0, sorter.order.indexOf('end')) : ''
+      });
+      genData();
+    }
    // const asc = sorter?.order && sorter?.order === 'ascend' ? true : false;
    // const sortColumn = sorter.field && toLine(sorter.field);
    // genData({ pageNo: pagination.current, pageSize: pagination.pageSize, filters, asc, sortColumn, queryTerm: searchResult, ...allParams });
@@ -106,7 +113,7 @@ const TopicMessages = (props: any) => {

  useEffect(() => {
    props.positionType === 'Messages' && genData();
-  }, [props, params]);
+  }, [props, params, sorter]);

  return (
    <>
--- a/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/config.tsx
+++ b/km-console/packages/layout-clusters-fe/src/pages/TopicDetail/config.tsx
@@ -85,7 +85,8 @@ export const getTopicMessagesColmns = () => {
      title: 'Timestamp',
      dataIndex: 'timestampUnitMs',
      key: 'timestampUnitMs',
-      render: (t: number) => (t ? moment(t).format(timeFormat) : '-'),
+      sorter: true,
+      render: (t: number) => (t ? moment(t).format(timeFormat) + '.' + moment(t).millisecond() : '-'),
    },
    {
      title: 'Key',
--- a/km-rest/src/main/java/com/xiaojukeji/know/streaming/km/rest/api/v3/topic/TopicStateController.java
+++ b/km-rest/src/main/java/com/xiaojukeji/know/streaming/km/rest/api/v3/topic/TopicStateController.java
@@ -92,9 +92,8 @@ public class TopicStateController {
    @ResponseBody
    public Result<List<TopicRecordVO>> getTopicMessages(@PathVariable Long clusterPhyId,
                                                        @PathVariable String topicName,
-                                                        @Validated @RequestBody TopicRecordDTO dto,
-                                                        @Validated PaginationSortDTO sortDto) throws Exception {
-        return topicStateManager.getTopicMessages(clusterPhyId, topicName, dto, sortDto);
+                                                        @Validated @RequestBody TopicRecordDTO dto) throws Exception {
+        return topicStateManager.getTopicMessages(clusterPhyId, topicName, dto);
    }

    @ApiOperation(value = "Topic-ACL信息", notes = "")