fuweng11 commented on code in PR #6046:
URL: https://github.com/apache/inlong/pull/6046#discussion_r1018697052
##########
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:
This is because we need to call the stream workflow to configure. If we add
a stream later, we do not need to configure the group workflow, but directly
execute the stream workflow.
--
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]
