Skip to content

Commit

Permalink
Merge branch 'fix_1043' into ve_3.x_dev
Browse files Browse the repository at this point in the history
  • Loading branch information
qiao.zeng committed Nov 12, 2023
2 parents a8be274 + 251f7f7 commit 6e9dc4f
Showing 1 changed file with 63 additions and 4 deletions.
67 changes: 63 additions & 4 deletions km-biz/src/main/java/com/xiaojukeji/know/streaming/km/biz/topic/impl/OpTopicManagerImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import com.xiaojukeji.know.streaming.km.common.bean.dto.topic.TopicExpansionDTO;
import com.xiaojukeji.know.streaming.km.common.bean.entity.broker.Broker;
import com.xiaojukeji.know.streaming.km.common.bean.entity.cluster.ClusterPhy;
import com.xiaojukeji.know.streaming.km.common.bean.entity.param.config.KafkaTopicConfigParam;
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;
Expand All @@ -17,17 +18,17 @@
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;
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
import com.xiaojukeji.know.streaming.km.common.utils.*;
import com.xiaojukeji.know.streaming.km.common.utils.kafka.KafkaReplicaAssignUtil;
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.PartitionService;
import com.xiaojukeji.know.streaming.km.core.service.topic.OpTopicService;
import com.xiaojukeji.know.streaming.km.core.service.topic.TopicConfigService;
import com.xiaojukeji.know.streaming.km.core.service.topic.TopicService;
import kafka.admin.AdminUtils;
import kafka.admin.BrokerMetadata;
import org.apache.kafka.common.config.TopicConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
Expand Down Expand Up @@ -61,6 +62,9 @@ public class OpTopicManagerImpl implements OpTopicManager {
@Autowired
private PartitionService partitionService;

@Autowired
private TopicConfigService topicConfigService;

@Override
public Result<Void> createTopic(TopicCreateDTO dto, String operator) {
log.info("method=createTopic||param={}||operator={}.", dto, operator);
Expand Down Expand Up @@ -160,17 +164,72 @@ public Result<Void> expandTopic(TopicExpansionDTO dto, String operator) {

@Override
public Result<Void> truncateTopic(Long clusterPhyId, String topicName, String operator) {
// 增加delete配置
Result<Tuple<Boolean, String>> rt = this.addDeleteConfigIfNotExist(clusterPhyId, topicName, operator);
if (rt.failed()) {
log.error("method=truncateTopic||clusterPhyId={}||topicName={}||operator={}||result={}||msg=get config from kafka failed", clusterPhyId, topicName, operator, rt);
return Result.buildFromIgnoreData(rt);
}

// 清空Topic
Result<Void> rv = opTopicService.truncateTopic(new TopicTruncateParam(clusterPhyId, topicName, KafkaConstant.TOPICK_TRUNCATE_DEFAULT_OFFSET), operator);
if (rv.failed()) {
return rv;
log.error("method=truncateTopic||clusterPhyId={}||topicName={}||originConfig={}||operator={}||result={}||msg=truncate topic failed", clusterPhyId, topicName, rt.getData().v2(), operator, rv);
// config被修改了，则错误提示需要提醒一下，否则直接返回错误
return rt.getData().v1() ? Result.buildFailure(rv.getCode(), rv.getMessage() + "\t\n" + String.format("Topic的CleanupPolicy已被修改，需要手动恢复为%s", rt.getData().v2())) : rv;
}

// 恢复compact配置
rv = this.recoverConfigIfChanged(clusterPhyId, topicName, rt.getData().v1(), rt.getData().v2(), operator);
if (rv.failed()) {
log.error("method=truncateTopic||clusterPhyId={}||topicName={}||originConfig={}||operator={}||result={}||msg=truncate topic success but recover config failed", clusterPhyId, topicName, rt.getData().v2(), operator, rv);
// config被修改了，则错误提示需要提醒一下，否则直接返回错误
return Result.buildFailure(rv.getCode(), String.format("Topic清空操作已成功，但是恢复CleanupPolicy配置失败，需要手动恢复为%s。", rt.getData().v2()) + "\t\n" + rv.getMessage());
}

return Result.buildSuc();
}

/**************************************************** private method ****************************************************/

private Result<Tuple<Boolean, String>> addDeleteConfigIfNotExist(Long clusterPhyId, String topicName, String operator) {
// 获取Topic配置
Result<Map<String, String>> configMapResult = topicConfigService.getTopicConfigFromKafka(clusterPhyId, topicName);
if (configMapResult.failed()) {
return Result.buildFromIgnoreData(configMapResult);
}

String cleanupPolicyValue = configMapResult.getData().getOrDefault(TopicConfig.CLEANUP_POLICY_CONFIG, "");
List<String> cleanupPolicyValueList = CommonUtils.string2StrList(cleanupPolicyValue);
if (cleanupPolicyValueList.size() == 1 && cleanupPolicyValueList.contains(TopicConfig.CLEANUP_POLICY_DELETE)) {
// 不需要修改
return Result.buildSuc(new Tuple<>(Boolean.FALSE, cleanupPolicyValue));
}

Map<String, String> changedConfigMap = new HashMap<>(1);
changedConfigMap.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE);

Result<Void> rv = topicConfigService.modifyTopicConfig(new KafkaTopicConfigParam(clusterPhyId, topicName, changedConfigMap), operator);
if (rv.failed()) {
// 修改失败
return Result.buildFromIgnoreData(rv);
}

return Result.buildSuc(new Tuple<>(Boolean.TRUE, cleanupPolicyValue));
}

private Result<Void> recoverConfigIfChanged(Long clusterPhyId, String topicName, Boolean changed, String originValue, String operator) {
if (!changed) {
// 没有修改，直接返回
return Result.buildSuc();
}

// 恢复配置
Map<String, String> changedConfigMap = new HashMap<>(1);
changedConfigMap.put(TopicConfig.CLEANUP_POLICY_CONFIG, originValue);

return topicConfigService.modifyTopicConfig(new KafkaTopicConfigParam(clusterPhyId, topicName, changedConfigMap), operator);
}

private Seq<BrokerMetadata> buildBrokerMetadataSeq(Long clusterPhyId, final List<Integer> selectedBrokerIdList) {
// 选取Broker列表
Expand Down

0 comments on commit 6e9dc4f

Please sign in to comment.