chestnut-c commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018684017
##########
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java:
##########
@@ -96,4 +128,39 @@ public ListenerResult listen(WorkflowContext context)
throws WorkflowListenerExc
return ListenerResult.success("success");
}
+ private void createQueueForStreams(InlongGroupInfo groupInfo,
List<InlongStreamInfo> streamInfos, String operator) {
+ String groupId = groupInfo.getInlongGroupId();
+ log.info("success to start stream process for groupId={}", groupId);
+
+ for (InlongStreamInfo stream : streamInfos) {
+ StreamResourceProcessForm form =
StreamResourceProcessForm.getProcessForm(groupInfo, stream, INIT);
+ String streamId = stream.getInlongStreamId();
+ final String errMsg = "failed to start stream process for
groupId=" + groupId + " streamId=" + streamId;
+
+ CompletableFuture<WorkflowResult> future = CompletableFuture
+ .supplyAsync(() ->
workflowService.start(CREATE_STREAM_RESOURCE, operator, form), EXECUTOR_SERVICE)
+ .whenComplete((result, ex) -> {
+ if (ex != null) {
+ log.error(errMsg + ": " + ex.getMessage());
+ throw new WorkflowListenerException(errMsg, ex);
+ } else {
+ List<TaskResponse> tasks = result.getNewTasks();
+ if (TaskStatus.FAILED == tasks.get(tasks.size() -
1).getStatus()) {
+ log.error(errMsg);
+ throw new WorkflowListenerException(errMsg);
+ }
+ }
+ });
+ try {
+ future.get(180, TimeUnit.SECONDS);
Review Comment:
Waiting for the result in the for loop is equivalent to synchronization. Why
use the asynchronous method?
--
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
