castorqin commented on code in PR #9029:
URL: https://github.com/apache/inlong/pull/9029#discussion_r1355977648
##########
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:
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);
+ 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(),
clsSinkDTO.getTopicName());
} catch (TencentCloudSDKException e) {
String errMsg = "Create cls topic failed: " + e.getMessage();
Review Comment:
done
##########
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);
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
