[INLONG-9921][Manager] Fix the problem of manager can't stop data sync job

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/GroupStatus.java

@@ -41,13 +41,13 @@ public enum GroupStatus {
     CONFIG_FAILED(120, "configuration failed"),
     CONFIG_SUCCESSFUL(130, "configuration successful"),
 
-    CONFIG_OFFLINE_ING(141, "in configure offline"),
-    CONFIGURATION_OFFLINE(140, "configure offline successful"),
+    CONFIG_OFFLINE_ING(141, "configuration is going offline"),
+    CONFIG_OFFLINE_SUCCESSFUL(140, "configuration offline successful"),
 
-    CONFIG_ONLINE_ING(151, "in configure online"),
+    CONFIG_ONLINE_ING(151, "configuration is going online"),
 
-    CONFIG_DELETING(41, "configure deleting"),
-    CONFIG_DELETED(40, "configure deleted"),
+    CONFIG_DELETING(41, "configuration deleting"),
+    CONFIG_DELETED(40, "configuration deleted"),
 
     // FINISH is used for batch task.
     FINISH(131, "finish");
@@ -71,9 +71,10 @@ public enum GroupStatus {
                 Sets.newHashSet(CONFIG_SUCCESSFUL, TO_BE_APPROVAL, CONFIG_ING, CONFIG_OFFLINE_ING, CONFIG_DELETING));
 
         GROUP_STATE_AUTOMATON.put(
-                CONFIG_OFFLINE_ING, Sets.newHashSet(CONFIG_OFFLINE_ING, CONFIGURATION_OFFLINE, CONFIG_FAILED));
-        GROUP_STATE_AUTOMATON.put(CONFIGURATION_OFFLINE, Sets.newHashSet(CONFIGURATION_OFFLINE, CONFIG_ONLINE_ING,
-                CONFIG_DELETING));
+                CONFIG_OFFLINE_ING, Sets.newHashSet(CONFIG_OFFLINE_ING, CONFIG_OFFLINE_SUCCESSFUL, CONFIG_FAILED));
+        GROUP_STATE_AUTOMATON.put(
+                CONFIG_OFFLINE_SUCCESSFUL, Sets.newHashSet(CONFIG_OFFLINE_SUCCESSFUL, CONFIG_ONLINE_ING,
+                        CONFIG_DELETING));
 
         GROUP_STATE_AUTOMATON.put(CONFIG_ONLINE_ING,
                 Sets.newHashSet(CONFIG_ONLINE_ING, CONFIG_FAILED, CONFIG_SUCCESSFUL));
@@ -143,7 +144,7 @@ public static boolean deleteStreamFirst(GroupStatus status) {
         return status == GroupStatus.APPROVE_PASSED
                 || status == GroupStatus.CONFIG_FAILED
                 || status == GroupStatus.CONFIG_SUCCESSFUL
-                || status == GroupStatus.CONFIGURATION_OFFLINE
+                || status == GroupStatus.CONFIG_OFFLINE_SUCCESSFUL
                 || status == GroupStatus.FINISH;
     }
 
@@ -163,7 +164,7 @@ public static boolean allowedDeleteSubInfos(GroupStatus status) {
      */
     public static boolean allowedSuspend(GroupStatus status) {
         return status == GroupStatus.CONFIG_SUCCESSFUL
-                || status == GroupStatus.CONFIGURATION_OFFLINE
+                || status == GroupStatus.CONFIG_OFFLINE_SUCCESSFUL
                 || status == GroupStatus.FINISH;
     }
 

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/SimpleGroupStatus.java

@@ -52,7 +52,7 @@ public static SimpleGroupStatus parseStatusByCode(int code) {
                 return FAILED;
             case CONFIG_SUCCESSFUL:
                 return STARTED;
-            case CONFIGURATION_OFFLINE:
+            case CONFIG_OFFLINE_SUCCESSFUL:
                 return STOPPED;
             case FINISH:
                 return FINISHED;
@@ -101,7 +101,7 @@ public static List<Integer> parseStatusCodeByStr(String status) {
                 statusList.add(GroupStatus.CONFIG_SUCCESSFUL.getCode());
                 return statusList;
             case STOPPED:
-                statusList.add(GroupStatus.CONFIGURATION_OFFLINE.getCode());
+                statusList.add(GroupStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode());
                 return statusList;
             case FINISHED:
                 statusList.add(GroupStatus.FINISH.getCode());

inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/StreamStatus.java

@@ -27,11 +27,10 @@ public enum StreamStatus {
     CONFIG_FAILED(120, "configuration failed"),
     CONFIG_SUCCESSFUL(130, "configuration successful"),
 
-    SUSPENDING(141, "suspending"),
-    SUSPENDED(140, "suspended"),
+    CONFIG_OFFLINE_ING(141, "configuration is going offline"),
+    CONFIG_OFFLINE_SUCCESSFUL(140, "configuration offline successful"),
 
-    RESTARTING(151, "restarting"),
-    RESTARTED(150, "restarted"),
+    CONFIG_ONLINE_ING(151, "configuration is going online"),
 
     DELETING(41, "deleting"),
     DELETED(40, "deleted");
@@ -48,17 +47,17 @@ public enum StreamStatus {
      * Checks whether the given status allows updating operate.
      */
     public static boolean notAllowedUpdate(StreamStatus status) {
-        return status == StreamStatus.CONFIG_ING || status == StreamStatus.SUSPENDING
-                || status == StreamStatus.RESTARTING || status == StreamStatus.DELETING;
+        return status == StreamStatus.CONFIG_ING || status == StreamStatus.CONFIG_OFFLINE_ING
+                || status == StreamStatus.CONFIG_ONLINE_ING || status == StreamStatus.DELETING;
     }
 
     /**
      * Checks whether the given status allows deleting operate.
      */
     public static boolean notAllowedDelete(StreamStatus status) {
         return status == StreamStatus.CONFIG_ING
-                || status == StreamStatus.RESTARTING
-                || status == StreamStatus.SUSPENDING;
+                || status == StreamStatus.CONFIG_ONLINE_ING
+                || status == StreamStatus.CONFIG_OFFLINE_ING;
     }
 
     public static StreamStatus forCode(int code) {

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java

@@ -525,7 +525,7 @@ private void preProcessTemplateFileTask(TaskRequest taskRequest) {
         List<StreamSourceEntity> sourceEntities = sourceMapper.selectTemplateSourceByCluster(needCopiedStatusList,
                 Lists.newArrayList(SourceType.FILE), agentClusterName);
         Set<GroupStatus> noNeedAddTask = Sets.newHashSet(
-                GroupStatus.CONFIGURATION_OFFLINE, GroupStatus.CONFIG_OFFLINE_ING, GroupStatus.CONFIG_DELETING,
+                GroupStatus.CONFIG_OFFLINE_SUCCESSFUL, GroupStatus.CONFIG_OFFLINE_ING, GroupStatus.CONFIG_DELETING,
                 GroupStatus.CONFIG_DELETED);
         sourceEntities.stream()
                 .forEach(sourceEntity -> {

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupProcessService.java

@@ -336,7 +336,7 @@ private GroupStatus getFinalStatus(GroupStatus pendingStatus) {
             case CONFIG_ONLINE_ING:
                 return GroupStatus.CONFIG_SUCCESSFUL;
             case CONFIG_OFFLINE_ING:
-                return GroupStatus.CONFIGURATION_OFFLINE;
+                return GroupStatus.CONFIG_OFFLINE_SUCCESSFUL;
             default:
                 return GroupStatus.CONFIG_DELETED;
         }

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java

@@ -22,19 +22,25 @@
 import org.apache.inlong.manager.common.enums.GroupStatus;
 import org.apache.inlong.manager.common.enums.ProcessEvent;
 import org.apache.inlong.manager.common.enums.SourceStatus;
+import org.apache.inlong.manager.common.enums.StreamStatus;
 import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
 import org.apache.inlong.manager.pojo.group.InlongGroupRequest;
+import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
 import org.apache.inlong.manager.pojo.workflow.form.process.GroupResourceProcessForm;
 import org.apache.inlong.manager.service.group.InlongGroupService;
 import org.apache.inlong.manager.service.source.StreamSourceService;
+import org.apache.inlong.manager.service.stream.InlongStreamService;
 import org.apache.inlong.manager.workflow.WorkflowContext;
 import org.apache.inlong.manager.workflow.event.ListenerResult;
 import org.apache.inlong.manager.workflow.event.process.ProcessEventListener;
 
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import java.util.List;
+
 /**
  * The listener of InlongGroup when update operates successfully.
  */
@@ -46,6 +52,8 @@ public class UpdateGroupCompleteListener implements ProcessEventListener {
     private InlongGroupService groupService;
     @Autowired
     private StreamSourceService sourceService;
+    @Autowired
+    private InlongStreamService streamService;
 
     @Override
     public ProcessEvent event() {
@@ -63,12 +71,18 @@ public ListenerResult listen(WorkflowContext context) {
         InlongGroupInfo groupInfo = form.getGroupInfo();
         InlongGroupRequest groupRequest = groupInfo.genRequest();
         String operator = context.getOperator();
+        List<InlongStreamInfo> streamInfos = form.getStreamInfos();
+        if (CollectionUtils.isNotEmpty(streamInfos)) {
+            streamInfos.forEach(streamInfo -> streamService.updateWithoutCheck(streamInfo.genRequest(), operator));
+        }
         switch (operateType) {
             case SUSPEND:
-                groupService.updateStatus(groupId, GroupStatus.CONFIGURATION_OFFLINE.getCode(), operator);
+                streamService.updateStatus(groupId, null, StreamStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode(), operator);
+                groupService.updateStatus(groupId, GroupStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode(), operator);
                 groupService.update(groupRequest, operator);
                 break;
             case RESTART:
+                streamService.updateStatus(groupId, null, StreamStatus.CONFIG_SUCCESSFUL.getCode(), operator);
                 groupService.updateStatus(groupId, GroupStatus.CONFIG_SUCCESSFUL.getCode(), operator);
                 groupService.update(groupRequest, operator);
                 break;

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/UpdateStreamCompleteListener.java

@@ -57,10 +57,10 @@ public ListenerResult listen(WorkflowContext context) throws Exception {
         StreamStatus status;
         switch (operateType) {
             case RESTART:
-                status = StreamStatus.RESTARTED;
+                status = StreamStatus.CONFIG_SUCCESSFUL;
                 break;
             case SUSPEND:
-                status = StreamStatus.SUSPENDED;
+                status = StreamStatus.CONFIG_OFFLINE_SUCCESSFUL;
                 break;
             case DELETE:
                 status = StreamStatus.DELETED;

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/UpdateStreamListener.java

@@ -54,10 +54,10 @@ public ListenerResult listen(WorkflowContext context) throws Exception {
         final String streamId = streamInfo.getInlongStreamId();
         switch (operateType) {
             case SUSPEND:
-                streamService.updateStatus(groupId, streamId, StreamStatus.SUSPENDING.getCode(), operator);
+                streamService.updateStatus(groupId, streamId, StreamStatus.CONFIG_OFFLINE_ING.getCode(), operator);
                 break;
             case RESTART:
-                streamService.updateStatus(groupId, streamId, StreamStatus.RESTARTING.getCode(), operator);
+                streamService.updateStatus(groupId, streamId, StreamStatus.CONFIG_ONLINE_ING.getCode(), operator);
                 break;
             case DELETE:
                 streamService.updateStatus(groupId, streamId, StreamStatus.DELETING.getCode(), operator);

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamProcessService.java

@@ -134,12 +134,12 @@ public boolean suspendProcess(String groupId, String streamId, String operator,
         InlongStreamInfo streamInfo = streamService.get(groupId, streamId);
         Preconditions.expectNotNull(streamInfo, ErrorCodeEnum.STREAM_NOT_FOUND.getMessage());
         StreamStatus status = StreamStatus.forCode(streamInfo.getStatus());
-        if (status == StreamStatus.SUSPENDED || status == StreamStatus.SUSPENDING) {
+        if (status == StreamStatus.CONFIG_OFFLINE_SUCCESSFUL || status == StreamStatus.CONFIG_OFFLINE_ING) {
             log.warn("groupId={}, streamId={} is already in {}", groupId, streamId, status);
             return true;
         }
 
-        if (status != StreamStatus.CONFIG_SUCCESSFUL && status != StreamStatus.RESTARTED) {
+        if (status != StreamStatus.CONFIG_SUCCESSFUL) {
             throw new BusinessException(String.format("stream status=%s not support suspend stream"
                     + " for groupId=%s streamId=%s", status, groupId, streamId));
         }
@@ -176,12 +176,12 @@ public boolean restartProcess(String groupId, String streamId, String operator,
         InlongStreamInfo streamInfo = streamService.get(groupId, streamId);
         Preconditions.expectNotNull(streamInfo, ErrorCodeEnum.STREAM_NOT_FOUND.getMessage());
         StreamStatus status = StreamStatus.forCode(streamInfo.getStatus());
-        if (status == StreamStatus.RESTARTED || status == StreamStatus.RESTARTING) {
+        if (status == StreamStatus.CONFIG_ONLINE_ING) {
             log.warn("inlong stream was already in {} for groupId={}, streamId={}", status, groupId, streamId);
             return true;
         }
 
-        if (status != StreamStatus.SUSPENDED) {
+        if (status != StreamStatus.CONFIG_OFFLINE_SUCCESSFUL) {
             throw new BusinessException(String.format("stream status=%s not support restart stream"
                     + " for groupId=%s streamId=%s", status, groupId, streamId));
         }

inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java

@@ -143,8 +143,9 @@ public void suspendSource(String groupId, String streamId) {
         sources.stream()
                 .filter(source -> source.getTaskMapId() != null)
                 .forEach(source -> sourceService.stop(source.getId(), GLOBAL_OPERATOR));
-        groupMapper.updateStatus(groupId, GroupStatus.CONFIGURATION_OFFLINE.getCode(), GLOBAL_OPERATOR);
-        streamMapper.updateStatusByIdentifier(groupId, streamId, StreamStatus.SUSPENDED.getCode(), GLOBAL_OPERATOR);
+        groupMapper.updateStatus(groupId, GroupStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode(), GLOBAL_OPERATOR);
+        streamMapper.updateStatusByIdentifier(groupId, streamId, StreamStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode(),
+                GLOBAL_OPERATOR);
     }
 
     /**
@@ -156,7 +157,8 @@ public void restartSource(String groupId, String streamId) {
                 .filter(source -> source.getTaskMapId() != null)
                 .forEach(source -> sourceService.restart(source.getId(), GLOBAL_OPERATOR));
         groupMapper.updateStatus(groupId, GroupStatus.CONFIG_SUCCESSFUL.getCode(), GLOBAL_OPERATOR);
-        streamMapper.updateStatusByIdentifier(groupId, streamId, StreamStatus.RESTARTED.getCode(), GLOBAL_OPERATOR);
+        streamMapper.updateStatusByIdentifier(groupId, streamId, StreamStatus.CONFIG_SUCCESSFUL.getCode(),
+                GLOBAL_OPERATOR);
     }
 
     public void deleteSource(String groupId, String streamId) {

inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/group/InlongGroupProcessServiceTest.java

@@ -96,7 +96,7 @@ private void testSuspendProcess() {
         ProcessResponse response = result.getProcessInfo();
         Assertions.assertSame(response.getStatus(), ProcessStatus.COMPLETED);
         InlongGroupInfo groupInfo = groupService.get(GROUP_ID);
-        Assertions.assertEquals(groupInfo.getStatus(), GroupStatus.CONFIGURATION_OFFLINE.getCode());
+        Assertions.assertEquals(groupInfo.getStatus(), GroupStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode());
     }
 
     private void testRestartProcess() {

inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/source/listener/StreamSourceListenerTest.java

@@ -103,7 +103,7 @@ private void testFrozenSource(Integer sourceId) {
     private void testRestartSource(Integer sourceId) {
         groupService.updateStatus(GROUP_ID, GroupStatus.CONFIG_OFFLINE_ING.getCode(), GLOBAL_OPERATOR);
         groupService.update(groupInfo.genRequest(), GLOBAL_OPERATOR);
-        groupService.updateStatus(GROUP_ID, GroupStatus.CONFIGURATION_OFFLINE.getCode(), GLOBAL_OPERATOR);
+        groupService.updateStatus(GROUP_ID, GroupStatus.CONFIG_OFFLINE_SUCCESSFUL.getCode(), GLOBAL_OPERATOR);
         groupService.update(groupInfo.genRequest(), GLOBAL_OPERATOR);
 
         sourceService.updateStatus(GROUP_ID, null, SourceStatus.SOURCE_NORMAL.getCode(), GLOBAL_OPERATOR);