[INLONG-1858][Manager] Fix adding columns for Hive not take effect error (#4554)

Author: woofyzhao

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

@@ -42,7 +42,7 @@ public enum FieldType {
     public static FieldType forName(String name) {
         Preconditions.checkNotNull(name, "FieldType should not be null");
         for (FieldType value : values()) {
-            if (value.toString().equals(name)) {
+            if (value.toString().equalsIgnoreCase(name)) {
                 return value;
             }
         }

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/operation/InlongStreamProcessOperation.java

@@ -86,11 +86,12 @@ public boolean startProcess(String groupId, String streamId, String operator, bo
             throw new BusinessException(ErrorCodeEnum.STREAM_NOT_FOUND);
         }
         StreamStatus status = StreamStatus.forCode(streamInfo.getStatus());
-        if (status == StreamStatus.CONFIG_ING || status == StreamStatus.CONFIG_SUCCESSFUL) {
+        if (status == StreamStatus.CONFIG_ING) {
             log.warn("GroupId={}, StreamId={} is already in {}", groupId, streamId, status);
             return true;
         }
-        if (status != StreamStatus.NEW || status != StreamStatus.CONFIG_FAILED) {
+        if (status != StreamStatus.NEW && status != StreamStatus.CONFIG_FAILED
+                && status != StreamStatus.CONFIG_SUCCESSFUL) {
             throw new BusinessException(
                     String.format("GroupId=%s, StreamId=%s, status=%s not correct for stream start", groupId, streamId,
                             status));

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/StreamSinkServiceImpl.java

@@ -26,6 +26,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
 import org.apache.inlong.manager.common.enums.GlobalConstants;
+import org.apache.inlong.manager.common.enums.GroupStatus;
 import org.apache.inlong.manager.common.enums.SinkStatus;
 import org.apache.inlong.manager.common.enums.SinkType;
 import org.apache.inlong.manager.common.exceptions.BusinessException;
@@ -41,10 +42,12 @@
 import org.apache.inlong.manager.dao.entity.StreamSinkEntity;
 import org.apache.inlong.manager.dao.mapper.StreamSinkEntityMapper;
 import org.apache.inlong.manager.dao.mapper.StreamSinkFieldEntityMapper;
+import org.apache.inlong.manager.service.core.operation.InlongStreamProcessOperation;
 import org.apache.inlong.manager.service.group.GroupCheckService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -62,7 +65,6 @@
 public class StreamSinkServiceImpl implements StreamSinkService {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(StreamSinkServiceImpl.class);
-
     @Autowired
     private SinkOperationFactory operationFactory;
     @Autowired
@@ -71,6 +73,9 @@ public class StreamSinkServiceImpl implements StreamSinkService {
     private StreamSinkEntityMapper sinkMapper;
     @Autowired
     private StreamSinkFieldEntityMapper sinkFieldMapper;
+    @Autowired
+    private AutowireCapableBeanFactory autowireCapableBeanFactory;
+    private InlongStreamProcessOperation streamProcessOperation;
 
     @Override
     @Transactional(rollbackFor = Throwable.class)
@@ -192,7 +197,7 @@ public Boolean update(SinkRequest request, String operator) {
         String streamId = request.getInlongStreamId();
         String sinkName = request.getSinkName();
         String sinkType = request.getSinkType();
-        InlongGroupEntity groupEntity = groupCheckService.checkGroupStatus(groupId, operator);
+        final InlongGroupEntity groupEntity = groupCheckService.checkGroupStatus(groupId, operator);
 
         // Check whether the sink name exists with the same groupId and streamId
         List<StreamSinkEntity> sinkList = sinkMapper.selectByRelatedId(groupId, streamId, sinkName);
@@ -214,9 +219,14 @@ public Boolean update(SinkRequest request, String operator) {
 
         // The inlong group status is [Configuration successful], then asynchronously initiate
         // the [Single inlong stream resource creation] workflow
-//        if (EntityStatus.GROUP_CONFIG_SUCCESSFUL.getCode().equals(groupEntity.getStatus())) {
-//            executorService.execute(new WorkflowStartRunnable(operator, groupEntity, streamId));
-//        }
+        if (GroupStatus.CONFIG_SUCCESSFUL.getCode().equals(groupEntity.getStatus())) {
+            // To work around the circular reference check we manually instantiate and wire
+            if (streamProcessOperation == null) {
+                streamProcessOperation = new InlongStreamProcessOperation();
+                autowireCapableBeanFactory.autowireBean(streamProcessOperation);
+            }
+            streamProcessOperation.startProcess(groupId, streamId, operator, true);
+        }
         LOGGER.info("success to update sink info: {}", request);
         return true;
     }