fuweng11 commented on code in PR #9029:
URL: https://github.com/apache/inlong/pull/9029#discussion_r1349839871
##########
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 sink = {}info status success ,topicName {}",
sinkInfo.getSinkName(),
Review Comment:
update cls info status success for sinkId= {}, topicName ={}
--
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
