fuweng11 commented on code in PR #9029:
URL: https://github.com/apache/inlong/pull/9029#discussion_r1350134369
##########
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:
topicId
##########
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);
+ clsSinkDTO.setTopicId(topicID);
sinkInfo.setExtParams(JsonUtils.toJsonString(clsSinkDTO));
// create topic index by tokenizer
- this.createTopicIndex(sinkInfo);
- StreamSinkEntity streamSinkEntity = new StreamSinkEntity();
- CommonBeanUtils.copyProperties(sinkInfo, streamSinkEntity, true);
- streamSinkEntityMapper.updateByIdSelective(streamSinkEntity);
+ this.createTopicIndex(sinkInfo, clsDataNode);
+ // update set topic id into sink info
+ updateSinkInfo(sinkInfo, clsSinkDTO);
String info = "success to create cls resource";
sinkService.updateStatus(sinkInfo.getId(),
SinkStatus.CONFIG_SUCCESSFUL.getCode(), info);
- LOG.info("update cls sink = {}info status success ,topicName {}",
streamSinkEntity.getSinkName(),
+ LOG.info("update cls info status success for sinkId= {},topicName
={}", sinkInfo.getSinkName(),
Review Comment:
for sinkId = {}, topicName = {}
##########
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());
Review Comment:
tokenizer is blank for topic = {}
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/cls/ClsResourceOperator.java:
##########
@@ -152,15 +205,87 @@ private void createTopicIndex(SinkInfo sinkInfo) throws
BusinessException {
CreateIndexRequest req = new CreateIndexRequest();
req.setTopicId(clsSinkDTO.getTopicId());
req.setRule(ruleInfo);
+ return req;
+ }
+
+ private void updateTopicIndex(SinkInfo sinkInfo, ClsDataNodeDTO
clsDataNode) {
+ ClsClient clsClient = getClsClient(clsDataNode);
+ ClsSinkDTO clsSinkDTO = JsonUtils.parseObject(sinkInfo.getExtParams(),
ClsSinkDTO.class);
+ RuleInfo ruleInfo = new RuleInfo();
+ FullTextInfo fullTextInfo1 = new FullTextInfo();
Review Comment:
FullTextInfo fullTextInfo1?
##########
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:
Strings.isBlank(topicID)
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/cls/ClsResourceOperator.java:
##########
@@ -152,15 +205,87 @@ private void createTopicIndex(SinkInfo sinkInfo) throws
BusinessException {
CreateIndexRequest req = new CreateIndexRequest();
req.setTopicId(clsSinkDTO.getTopicId());
req.setRule(ruleInfo);
+ return req;
+ }
+
+ private void updateTopicIndex(SinkInfo sinkInfo, ClsDataNodeDTO
clsDataNode) {
+ ClsClient clsClient = getClsClient(clsDataNode);
+ ClsSinkDTO clsSinkDTO = JsonUtils.parseObject(sinkInfo.getExtParams(),
ClsSinkDTO.class);
+ RuleInfo ruleInfo = new RuleInfo();
+ FullTextInfo fullTextInfo1 = new FullTextInfo();
+ fullTextInfo1.setTokenizer(clsSinkDTO.getTokenizer());
+ ruleInfo.setFullText(fullTextInfo1);
+
+ ModifyIndexRequest req = new ModifyIndexRequest();
+ req.setTopicId(clsSinkDTO.getTopicId());
+ req.setRule(ruleInfo);
try {
- clsClient.CreateIndex(req);
+ clsClient.ModifyIndex(req);
} catch (TencentCloudSDKException e) {
- String errMsg = "Create cls topic index failed: " + e.getMessage();
+ String errMsg = "update 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("topic {} create index success tokenizer is {}",
clsSinkDTO.getTopicName(), clsSinkDTO.getTokenizer());
+ }
+
+ /**
+ * Describe cls topicId by topic name
+ */
+ private String describeTopicIDByTopicName(SinkInfo sinkInfo,
ClsDataNodeDTO clsDataNode) {
+ ClsClient clsClient = getClsClient(clsDataNode);
+ ClsSinkDTO clsSinkDTO = JsonUtils.parseObject(sinkInfo.getExtParams(),
ClsSinkDTO.class);
+ Filter[] filters = getDescribeFilters(clsDataNode, clsSinkDTO);
+ DescribeTopicsRequest req = new DescribeTopicsRequest();
+ req.setFilters(filters);
+ req.setPreciseSearch(PRECISE_SEARCH);
+ try {
+ DescribeTopicsResponse describeTopicsResponse =
clsClient.DescribeTopics(req);
+ LOG.info("sink {} describe cls topic success topic count {}",
sinkInfo.getSinkName(),
Review Comment:
success describe cls topic count = {} for sinkId = {}
##########
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:
create index success for topic = {}, tokenizer = {}
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
