castorqin commented on code in PR #9029:
URL: https://github.com/apache/inlong/pull/9029#discussion_r1354806108
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/cls/ClsResourceOperator.java:
##########
@@ -113,6 +120,32 @@ private void createTopicID(SinkInfo sinkInfo) {
}
}
+ private String getTopicID(SinkInfo sinkInfo, ClsDataNodeDTO clsDataNode,
ClsSinkDTO clsSinkDTO)
+ throws TencentCloudSDKException {
+ String topicID = describeTopicIDByTopicName(sinkInfo, clsDataNode);
+ if (Strings.isEmpty(topicID)) {
Review Comment:
done
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/cls/ClsResourceOperator.java:
##########
@@ -78,32 +90,27 @@ public void createSinkResource(SinkInfo sinkInfo) {
LOG.warn("create resource was disabled, skip to create for [" +
sinkInfo.getId() + "]");
return;
}
- this.createTopicID(sinkInfo);
+ this.createClsResource(sinkInfo);
this.assignCluster(sinkInfo);
}
/**
* Create cloud log service topic
*/
- private void createTopicID(SinkInfo sinkInfo) {
+ private void createClsResource(SinkInfo sinkInfo) {
ClsDataNodeDTO clsDataNode = getClsDataNode(sinkInfo);
ClsSinkDTO clsSinkDTO = JsonUtils.parseObject(sinkInfo.getExtParams(),
ClsSinkDTO.class);
try {
- ClsClient client = getClsClient(clsDataNode);
- CreateTopicRequest req = getCreateTopicRequest(clsDataNode,
clsSinkDTO);
- CreateTopicResponse resp = client.CreateTopic(req);
- LOG.info("create cls topic {} success ,topicId {}",
clsSinkDTO.getTopicName(), resp.getTopicId());
- // update set topic id into sink info
- clsSinkDTO.setTopicId(resp.getTopicId());
+ String topicID = getTopicID(sinkInfo, clsDataNode, clsSinkDTO);
Review Comment:
done
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/cls/ClsResourceOperator.java:
##########
@@ -136,14 +169,34 @@ private ClsClient getClsClient(ClsDataNodeDTO
clsDataNode) {
/**
* Create topic index by tokenizer
*/
- private void createTopicIndex(SinkInfo sinkInfo) throws BusinessException {
+ private void createTopicIndex(SinkInfo sinkInfo, ClsDataNodeDTO
clsDataNode) throws BusinessException {
+
ClsSinkDTO clsSinkDTO = JsonUtils.parseObject(sinkInfo.getExtParams(),
ClsSinkDTO.class);
- if (StringUtils.isNotBlank(clsSinkDTO.getTokenizer())) {
- LOG.warn("topic {} tokenizer is empty", clsSinkDTO.getTopicName());
+ if (StringUtils.isBlank(clsSinkDTO.getTokenizer())) {
+ LOG.warn("topic {} tokenizer is blank", clsSinkDTO.getTopicName());
+ return;
+ }
+ FullTextInfo topicIndexFullText = getTopicIndexFullText(sinkInfo,
clsDataNode);
+ if (ObjectUtils.anyNotNull(topicIndexFullText)) {
+ // if topic index exist,update
+ updateTopicIndex(sinkInfo, clsDataNode);
return;
}
- ClsDataNodeDTO clsDataNode = getClsDataNode(sinkInfo);
ClsClient clsClient = getClsClient(clsDataNode);
+ CreateIndexRequest req = getCreateIndexRequest(clsSinkDTO);
+ try {
+ clsClient.CreateIndex(req);
+ } catch (TencentCloudSDKException e) {
+ String errMsg = "Create cls topic index failed: " + e.getMessage();
+ LOG.error(errMsg, e);
+ sinkService.updateStatus(sinkInfo.getId(),
SinkStatus.CONFIG_FAILED.getCode(), errMsg);
+ throw new BusinessException(errMsg);
+ }
+ LOG.info("create index success for topic ={}, tokenizer = {}",
clsSinkDTO.getTopicName(),
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
