8x48/KnowStreaming
1
0
Fork 0
mirror of https://github.com/didi/KnowStreaming.git synced 2026-01-05 13:08:48 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

修改GroupOffsetResetEnum类名为OffsetTypeEnum

Browse Source
This commit is contained in:
zengqiao
2022-09-19 13:55:59 +08:00
parent 0bf85c97b5
commit 375c6f56c9
5 changed files with 24 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@@ -23,7 +23,7 @@ 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.KafkaConstant;
import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant; import com.xiaojukeji.know.streaming.km.common.constant.MsgConstant;
import com.xiaojukeji.know.streaming.km.common.converter.TopicVOConverter; import com.xiaojukeji.know.streaming.km.common.converter.TopicVOConverter;
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.SortTypeEnum; 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.AdminOperateException;
import com.xiaojukeji.know.streaming.km.common.exception.NotExistException; import com.xiaojukeji.know.streaming.km.common.exception.NotExistException;
@@ -164,7 +164,7 @@ public class TopicStateManagerImpl implements TopicStateManager {
Map<TopicPartition, OffsetAndTimestamp> partitionOffsetAndTimestampMap = new HashMap<>(); Map<TopicPartition, OffsetAndTimestamp> partitionOffsetAndTimestampMap = new HashMap<>();
// 获取指定时间每个分区的offset（按指定开始时间查询消息时） // 获取指定时间每个分区的offset（按指定开始时间查询消息时）
if (GroupOffsetResetEnum.PRECISE_TIMESTAMP.getResetType() == dto.getFilterOffsetReset()) { if (OffsetTypeEnum.PRECISE_TIMESTAMP.getResetType() == dto.getFilterOffsetReset()) {
Map<TopicPartition, Long> timestampsToSearch = new HashMap<>(); Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
partitionList.forEach(topicPartition -> { partitionList.forEach(topicPartition -> {
timestampsToSearch.put(topicPartition, dto.getStartTimestampUnitMs()); timestampsToSearch.put(topicPartition, dto.getStartTimestampUnitMs());
@@ -173,13 +173,13 @@ public class TopicStateManagerImpl implements TopicStateManager {
} }
for (TopicPartition partition : partitionList) { for (TopicPartition partition : partitionList) {
if (GroupOffsetResetEnum.EARLIEST.getResetType() == dto.getFilterOffsetReset()) { if (OffsetTypeEnum.EARLIEST.getResetType() == dto.getFilterOffsetReset()) {
// 重置到最旧 // 重置到最旧
kafkaConsumer.seek(partition, beginOffsetsMapResult.getData().get(partition)); kafkaConsumer.seek(partition, beginOffsetsMapResult.getData().get(partition));
} else if (GroupOffsetResetEnum.PRECISE_TIMESTAMP.getResetType() == dto.getFilterOffsetReset()) { } else if (OffsetTypeEnum.PRECISE_TIMESTAMP.getResetType() == dto.getFilterOffsetReset()) {
// 重置到指定时间 // 重置到指定时间
kafkaConsumer.seek(partition, partitionOffsetAndTimestampMap.get(partition).offset()); kafkaConsumer.seek(partition, partitionOffsetAndTimestampMap.get(partition).offset());
} else if (GroupOffsetResetEnum.PRECISE_OFFSET.getResetType() == dto.getFilterOffsetReset()) { } else if (OffsetTypeEnum.PRECISE_OFFSET.getResetType() == dto.getFilterOffsetReset()) {
// 重置到指定位置 // 重置到指定位置
} else { } else {

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

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

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

@@ -1,8 +1,8 @@
package com.xiaojukeji.know.streaming.km.common.bean.dto.topic; package com.xiaojukeji.know.streaming.km.common.bean.dto.topic;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties; 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.bean.dto.pagination.PaginationSortDTO;
import com.xiaojukeji.know.streaming.km.common.enums.OffsetTypeEnum;
import io.swagger.annotations.ApiModel; import io.swagger.annotations.ApiModel;
import io.swagger.annotations.ApiModelProperty; import io.swagger.annotations.ApiModelProperty;
import lombok.Data; import lombok.Data;
@@ -36,13 +36,12 @@ public class TopicRecordDTO extends PaginationSortDTO {
@ApiModelProperty(value = "预览超时时间", example = "10000") @ApiModelProperty(value = "预览超时时间", example = "10000")
private Long pullTimeoutUnitMs = 8000L; private Long pullTimeoutUnitMs = 8000L;
/**
* @see OffsetTypeEnum
*/
@ApiModelProperty(value = "offset", example = "") @ApiModelProperty(value = "offset", example = "")
private Integer filterOffsetReset = 0; private Integer filterOffsetReset = 0;
@ApiModelProperty(value = "开始日期时间戳", example = "") @ApiModelProperty(value = "开始日期时间戳", example = "")
private Long startTimestampUnitMs; private Long startTimestampUnitMs;
@ApiModelProperty(value = "结束日期时间戳", example = "")
private Long utilTimestampUnitMs;
} }

14
km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enums/GroupOffsetResetEnum.java → km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/enums/OffsetTypeEnum.java
View File

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