[INLONG-10711][Manager] Fix the problem of the stream source did not …

…update its status correctly after executing the workflow (#10712)
apache · Jul 24, 2024 · 93e9a32 · 93e9a32
1 parent 35c72c1
commit 93e9a32
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/.../src/main/java/org/apache/inlong/manager/service/listener/source/SourceStartListener.java b/.../src/main/java/org/apache/inlong/manager/service/listener/source/SourceStartListener.java
@@ -19,6 +19,7 @@
 
 import org.apache.inlong.manager.common.consts.InlongConstants;
 import org.apache.inlong.manager.common.enums.GroupOperateType;
+import org.apache.inlong.manager.common.enums.SourceStatus;
 import org.apache.inlong.manager.common.enums.TaskEvent;
 import org.apache.inlong.manager.pojo.source.SourceRequest;
 import org.apache.inlong.manager.pojo.source.StreamSource;
@@ -73,6 +74,12 @@ public ListenerResult listen(WorkflowContext context) throws Exception {
         InlongStreamInfo streamInfo = form.getStreamInfo();
         final String groupId = streamInfo.getInlongGroupId();
         final String streamId = streamInfo.getInlongStreamId();
+        if (InlongConstants.DATASYNC_REALTIME_MODE.equals(form.getGroupInfo().getInlongGroupMode())
+                || InlongConstants.DATASYNC_OFFLINE_MODE.equals(form.getGroupInfo().getInlongGroupMode())) {
+            streamSourceService.updateStatus(groupId, streamId, SourceStatus.SOURCE_NORMAL.getCode(), operator);
+        } else {
+            streamSourceService.updateStatus(groupId, streamId, SourceStatus.TO_BE_ISSUED_ADD.getCode(), operator);
+        }
         log.info("begin to update agent task config for groupId={}, streamId={}", groupId, streamId);
         List<StreamSource> sources = streamSourceService.listSource(groupId, streamId);
         for (StreamSource source : sources) {