Topic消息查询支持Timestamp排序，支持查询最新消息或最早消息 #534

Author: SuperWein

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

@@ -36,6 +36,7 @@
 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.kafka.clients.admin.OffsetSpec;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -161,7 +162,11 @@ public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String to
             maxMessage = Math.min(maxMessage, dto.getMaxRecords());
             kafkaConsumer.assign(partitionList);
             for (TopicPartition partition : partitionList) {
-                kafkaConsumer.seek(partition, Math.max(beginOffsetsMapResult.getData().get(partition), endOffsetsMapResult.getData().get(partition) - dto.getMaxRecords()));
+                if (Constant.EARLIEST.equals(dto.getFilterOffsetReset())) {
+                    kafkaConsumer.seek(partition, beginOffsetsMapResult.getData().get(partition));
+                } 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 +190,15 @@ public Result<List<TopicRecordVO>> getTopicMessages(Long clusterPhyId, String to
                     }
             }
 
+            // 排序
+            if (ObjectUtils.isNotEmpty(voList)) {
+                if (Constant.ASC.equals(dto.getSortType())) {
+                    voList.sort((o1, o2) -> (int) (o1.getTimestampUnitMs() - o2.getTimestampUnitMs()));
+                } else if (Constant.DESC.equals(dto.getSortType())) {
+                    voList.sort((o1, o2) -> (int) (o2.getTimestampUnitMs() - o1.getTimestampUnitMs()));
+                }
+            }
+
             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/topic/TopicRecordDTO.java

@@ -34,4 +34,11 @@ public class TopicRecordDTO extends BaseDTO {
 
     @ApiModelProperty(value = "预览超时时间", example = "10000")
     private Long pullTimeoutUnitMs = 8000L;
+
+    @ApiModelProperty(value = "排序", example = "desc")
+    private String sortType;
+
+    @ApiModelProperty(value = "offset", example = "latest")
+    private String filterOffsetReset;
+
 }

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

@@ -64,4 +64,17 @@ private Constant() {}
     public static final Float COLLECT_METRICS_ERROR_COST_TIME = -1.0F;
 
     public static final Integer DEFAULT_RETRY_TIME = 3;
+
+    /**
+     * 排序
+     */
+    public static final String ASC = "asc";
+    public static final String DESC = "desc";
+
+    /**
+     * 消费策略
+     */
+    public static final String LATEST = "latest";
+    public static final String EARLIEST = "earliest";
+
 }

km-console/packages/layout-clusters-fe/src/pages/TopicDetail/Messages.tsx

@@ -10,6 +10,7 @@ const defaultParams: any = {
   maxRecords: 100,
   pullTimeoutUnitMs: 5000,
   // filterPartitionId: 1,
+  filterOffsetReset: 'latest'
 };
 const defaultpaPagination = {
   current: 1,
@@ -29,6 +30,12 @@ const TopicMessages = (props: any) => {
   const [pagination, setPagination] = useState<any>(defaultpaPagination);
   const [form] = Form.useForm();
 
+  // 获取消息开始位置
+  const offsetResetList = [
+    { 'label': 'latest', value: 'latest' },
+    { 'label': 'earliest', value: 'earliest' }
+  ];
+
   // 默认排序
   const defaultSorter = {
     sortField: 'timestampUnitMs',
@@ -88,7 +95,10 @@ const TopicMessages = (props: any) => {
   };
 
   const onTableChange = (pagination: any, filters: any, sorter: any) => {
+    defaultSorter.sortField = sorter.field || '';
+    defaultSorter.sortType = sorter.order ? sorter.order.substring(0, sorter.order.indexOf('end')) : '';
     setPagination(pagination);
+    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 });
@@ -119,6 +129,15 @@ const TopicMessages = (props: any) => {
         </div>
         <div className="messages-query">
           <Form form={form} layout="inline" onFinish={onFinish}>
+            <Form.Item name="filterOffsetReset">
+              <Select
+                  options={offsetResetList}
+                  size="small"
+                  style={{ width: '120px' }}
+                  className={'detail-table-select'}
+                  placeholder="请选择offset"
+              />
+            </Form.Item>
             <Form.Item name="filterPartitionId">
               <Select
                 options={partitionIdList}
@@ -158,7 +177,7 @@ const TopicMessages = (props: any) => {
         showQueryForm={false}
         tableProps={{
           showHeader: false,
-          rowKey: 'path',
+          rowKey: 'offset',
           loading: loading,
           columns: getTopicMessagesColmns(),
           dataSource: data,
@@ -169,6 +188,7 @@ const TopicMessages = (props: any) => {
             bordered: false,
             onChange: onTableChange,
             scroll: { x: 'max-content' },
+            sortDirections: ['descend', 'ascend', 'default']
           },
         }}
       />