You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by go...@apache.org on 2021/12/16 14:39:08 UTC
[incubator-inlong] branch master updated: [INLONG-2012][InLong-Manager] Inlong-Agent could not fetch file agent task through "fileAgent/getTaskConf" (#2013)
This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 0f048a3 [INLONG-2012][InLong-Manager] Inlong-Agent could not fetch file agent task through "fileAgent/getTaskConf" (#2013)
0f048a3 is described below
commit 0f048a3a8fadb5a2fe533ae4948c6693edc7ec1f
Author: healchow <he...@gmail.com>
AuthorDate: Thu Dec 16 22:39:02 2021 +0800
[INLONG-2012][InLong-Manager] Inlong-Agent could not fetch file agent task through "fileAgent/getTaskConf" (#2013)
---
.../inlong/manager/common/enums/BizConstant.java | 2 +
.../inlong/manager/common/enums/EntityStatus.java | 19 ++--
.../common/pojo/agent/FileAgentCommandInfo.java | 51 +---------
.../dao/mapper/SourceFileDetailEntityMapper.java | 12 +++
.../mappers/SourceFileDetailEntityMapper.xml | 16 ++-
.../service/core/impl/AgentTaskServiceImpl.java | 112 ++++++++++-----------
.../service/core/impl/SourceDbServiceImpl.java | 2 +-
.../service/core/impl/SourceFileServiceImpl.java | 105 +++++++++----------
.../listener/BusinessCompleteProcessListener.java | 5 +
.../stream/StreamCompleteProcessListener.java | 12 ++-
10 files changed, 165 insertions(+), 171 deletions(-)
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/BizConstant.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/BizConstant.java
index d2eef5e..262fb3b 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/BizConstant.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/BizConstant.java
@@ -50,6 +50,8 @@ public class BizConstant {
public static final String CLUSTER_HIVE_TOPO = "HIVE_TOPO";
+ public static final String ID_IS_EMPTY = "id cannot empty during the update/delete operation";
+
public static final String GROUP_ID_IS_EMPTY = "business group id is empty";
public static final String STREAM_ID_IS_EMPTY = "data stream id is empty";
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/EntityStatus.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/EntityStatus.java
index cfaa927..12796c8 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/EntityStatus.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/EntityStatus.java
@@ -56,16 +56,15 @@ public enum EntityStatus {
DATA_STORAGE_CONFIG_FAILED(120, "configuration failed"),
DATA_STORAGE_CONFIG_SUCCESSFUL(130, "configuration successful"),
- // Data source related status
- DATA_RESOURCE_NEW(200, "new"),
- DATA_RESOURCE_DELETE(201, "deleted"),
-
- // Agent related status
- AGENT_WAIT_CREATE(200, "wait create"),
- AGENT_WAIT_STOP(201, "wait stop"),
- AGENT_WAIT_START(203, "wait start"),
- AGENT_WAIT_DELETE(204, "wait delete"),
- AGENT_WAIT_UPDATE(205, "wait update"),
+ // Data source (or Agent) related status
+ AGENT_DISABLE(99, "disable"),
+ AGENT_NORMAL(101, "normal"),
+ AGENT_FREEZE(102, "stopped"),
+ AGENT_FAILURE(103, "failed"),
+
+ // ADD(0), DEL(1), RETRY(2), BACKTRACK(3), FROZEN(4), ACTIVE(5), CHECK(6), REDOMETRIC(7), MAKEUP(8);
+ AGENT_ADD(200, "wait add"),
+ AGENT_DELETE(201, "wait delete"),
;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/agent/FileAgentCommandInfo.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/agent/FileAgentCommandInfo.java
index 4df91e4..f406ad3 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/agent/FileAgentCommandInfo.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/agent/FileAgentCommandInfo.java
@@ -35,6 +35,8 @@ public class FileAgentCommandInfo {
@ApiModelProperty(value = "agent result details")
private List<CommandInfoBean> commandInfo;
+ @ApiModel("File agent command info")
+ @Data
public static class CommandInfoBean {
@ApiModelProperty(value = "operation result")
@@ -47,7 +49,7 @@ public class FileAgentCommandInfo {
private int taskId;
@ApiModelProperty(value = "operation type")
- private int opType;
+ private int op;
@ApiModelProperty(value = "data time")
private String dataTime;
@@ -55,52 +57,5 @@ public class FileAgentCommandInfo {
@ApiModelProperty(value = "operation id")
private int id;
- public int getCommandResult() {
- return commandResult;
- }
-
- public void setCommandResult(int commandResult) {
- this.commandResult = commandResult;
- }
-
- public int getTaskId() {
- return taskId;
- }
-
- public void setTaskId(int taskId) {
- this.taskId = taskId;
- }
-
- public int getOpType() {
- return opType;
- }
-
- public void setOpType(int opType) {
- this.opType = opType;
- }
-
- public String getDataTime() {
- return dataTime;
- }
-
- public void setDataTime(String dataTime) {
- this.dataTime = dataTime;
- }
-
- public int getId() {
- return id;
- }
-
- public void setId(int id) {
- this.id = id;
- }
-
- public long getDeliveryTime() {
- return deliveryTime;
- }
-
- public void setDeliveryTime(long deliveryTime) {
- this.deliveryTime = deliveryTime;
- }
}
}
diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/SourceFileDetailEntityMapper.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/SourceFileDetailEntityMapper.java
index 0475659..067b53e 100644
--- a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/SourceFileDetailEntityMapper.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/SourceFileDetailEntityMapper.java
@@ -39,6 +39,18 @@ public interface SourceFileDetailEntityMapper {
int updateByPrimaryKey(SourceFileDetailEntity record);
+ /**
+ * Update source db detail status after approving business
+ *
+ * @param groupId Business group id
+ * @param streamId Data stream id
+ * @param status Modified status
+ * @param operator Edit person's name
+ * @return whether succeed
+ * @apiNote If stream id is null, update all data stream associated with group id
+ */
+ boolean updateStatusAfterApprove(String groupId, String streamId, Integer status, String operator);
+
List<SourceFileDetailEntity> selectByCondition(SourceFileDetailPageRequest request);
/**
diff --git a/inlong-manager/manager-dao/src/main/resources/mappers/SourceFileDetailEntityMapper.xml b/inlong-manager/manager-dao/src/main/resources/mappers/SourceFileDetailEntityMapper.xml
index c67f95e..e83c7a1 100644
--- a/inlong-manager/manager-dao/src/main/resources/mappers/SourceFileDetailEntityMapper.xml
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/SourceFileDetailEntityMapper.xml
@@ -372,6 +372,20 @@
and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
and is_deleted = 0
</update>
+ <update id="updateStatusAfterApprove">
+ update source_file_detail
+ <set>
+ status = #{status, jdbcType=INTEGER},
+ modifier = #{operator, jdbcType=VARCHAR}
+ </set>
+ <where>
+ inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ <if test="streamId != null and streamId != ''">
+ and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
+ </if>
+ and is_deleted = 0
+ </where>
+ </update>
<select id="selectFileAgentTaskByIp" resultMap="FileAgentTask">
select detail.inlong_group_id,
@@ -395,7 +409,7 @@
data_stream stream
where detail.ip = #{ip,jdbcType=VARCHAR}
and detail.is_deleted = 0
- and (floor(detail.STATUS / 100) = 2 OR floor(detail.STATUS / 100) = 3)
+ and (floor(detail.STATUS / 100) = 2)
and basic.inlong_group_id = detail.inlong_group_id
and basic.inlong_stream_id = detail.inlong_stream_id
and biz.inlong_group_id = detail.inlong_group_id
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentTaskServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentTaskServiceImpl.java
index 3ac5c99..7a3faa8 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentTaskServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentTaskServiceImpl.java
@@ -27,12 +27,14 @@ import java.util.Map;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
+import org.apache.inlong.manager.common.enums.EntityStatus;
import org.apache.inlong.manager.common.enums.FileAgentDataGenerateRule;
import org.apache.inlong.manager.common.pojo.agent.AgentStatusReportRequest;
import org.apache.inlong.manager.common.pojo.agent.CheckAgentTaskConfRequest;
import org.apache.inlong.manager.common.pojo.agent.ConfirmAgentIpRequest;
import org.apache.inlong.manager.common.pojo.agent.FileAgentCMDConfig;
import org.apache.inlong.manager.common.pojo.agent.FileAgentCommandInfo;
+import org.apache.inlong.manager.common.pojo.agent.FileAgentCommandInfo.CommandInfoBean;
import org.apache.inlong.manager.common.pojo.agent.FileAgentTaskConfig;
import org.apache.inlong.manager.common.pojo.agent.FileAgentTaskInfo;
import org.apache.inlong.manager.dao.entity.DataSourceCmdConfigEntity;
@@ -64,12 +66,17 @@ public class AgentTaskServiceImpl implements AgentTaskService {
@Override
public FileAgentTaskInfo getFileAgentTask(FileAgentCommandInfo info) {
+ LOGGER.debug("begin to get file agent task by info={}", info);
+ if (info == null || info.getAgentIp() == null) {
+ LOGGER.error("agent command info cannot be empty");
+ return null;
+ }
// Process the status of the postback task
- dealCommandResult(info);
+ this.dealCommandResult(info);
// Query pending tasks
- List<FileAgentTaskConfig> taskConfigs = getFileAgentTaskConfigs(info);
+ List<FileAgentTaskConfig> taskConfigs = this.getFileAgentTaskConfigs(info);
// Query pending special commands
List<FileAgentCMDConfig> cmdConfigs = getFileAgentCMDConfigs(info);
@@ -117,67 +124,60 @@ public class AgentTaskServiceImpl implements AgentTaskService {
}
}
- config.setAdditionalAttr(s.toString().substring(0, s.toString().length() - 1));
-
- int currentStatus = config.getStatus();
- if (currentStatus / 100 == 2) { // Modify status 20x -> 30x
- int nextStatus = currentStatus % 100 + 300;
- SourceFileDetailEntity update = new SourceFileDetailEntity();
- update.setId(config.getTaskId());
- update.setStatus(nextStatus);
- update.setPreviousStatus(currentStatus);
- fileDetailMapper.updateByPrimaryKeySelective(update);
- }
+ config.setAdditionalAttr(s.substring(0, s.toString().length() - 1));
}
return taskConfigs;
}
private void dealCommandResult(FileAgentCommandInfo info) {
- if (CollectionUtils.isNotEmpty(info.getCommandInfo())) {
- List<FileAgentCommandInfo.CommandInfoBean> commandInfos = info.getCommandInfo();
-
- for (FileAgentCommandInfo.CommandInfoBean command : commandInfos) {
- SourceFileDetailEntity current = fileDetailMapper.selectByPrimaryKey(command.getTaskId());
-
- if (current != null) {
- int opType = command.getOpType();
- if (opType == 2 || opType == 6 || opType == 8) { // Channel results issued by special orders
- DataSourceCmdConfigEntity cmd = new DataSourceCmdConfigEntity();
- if (command.getId() > 0) { // Modify the data result status of special commands
- cmd.setId(command.getId());
- cmd.setBsend(true);
- cmd.setModifyTime(new Date());
- cmd.setResultInfo(String.valueOf(command.getCommandResult()));
- sourceCmdConfigMapper.updateByPrimaryKeySelective(cmd);
- }
+ if (CollectionUtils.isEmpty(info.getCommandInfo())) {
+ LOGGER.warn("command info is empty, just return");
+ return;
+ }
- } else { // Modify the result status of the data collection task
- if (current.getModifyTime().getTime() - command.getDeliveryTime() > 1000 * 5) {
- log.warn(" task id {} receive heartbeat time delay more than 5's, skip it!",
- command.getTaskId());
- continue;
- }
- int nextStatus = 101;
- if (current.getStatus() != null && current.getStatus() / 100 == 3) { // Modify 30x -> 10x
- if (command.getCommandResult() == 0) { // Processed successfully
- if (current.getStatus() == 300 || current.getStatus() == 305) {
- nextStatus = 100;
- } else if (current.getStatus() == 301) { // To be deleted status becomes invalid 99
- nextStatus = 101;
- } else if (current.getStatus() == 304) { // To be deleted status becomes invalid 99
- nextStatus = 104;
- }
- } else if (command.getCommandResult() == 1) { // Processing failed
- nextStatus = 103;
- }
-
- SourceFileDetailEntity update = new SourceFileDetailEntity();
- update.setId(command.getTaskId());
- update.setStatus(nextStatus);
- update.setPreviousStatus(current.getStatus());
- fileDetailMapper.updateByPrimaryKeySelective(update);
+ for (CommandInfoBean command : info.getCommandInfo()) {
+ SourceFileDetailEntity current = fileDetailMapper.selectByPrimaryKey(command.getTaskId());
+ if (current == null) {
+ continue;
+ }
+
+ int op = command.getOp();
+ if (op == 2 || op == 6 || op == 8) { // Channel results issued by special orders
+ DataSourceCmdConfigEntity cmd = new DataSourceCmdConfigEntity();
+ if (command.getId() > 0) { // Modify the data result status of special commands
+ cmd.setId(command.getId());
+ cmd.setBsend(true);
+ cmd.setModifyTime(new Date());
+ cmd.setResultInfo(String.valueOf(command.getCommandResult()));
+ sourceCmdConfigMapper.updateByPrimaryKeySelective(cmd);
+ }
+
+ } else { // Modify the result status of the data collection task
+ if (current.getModifyTime().getTime() - command.getDeliveryTime() > 1000 * 5) {
+ log.warn(" task id {} receive heartbeat time delay more than 5's, skip it!",
+ command.getTaskId());
+ continue;
+ }
+
+ int result = command.getCommandResult();
+ int nextStatus = EntityStatus.AGENT_NORMAL.getCode();
+ int previousStatus = current.getStatus();
+ if (previousStatus / 100 == 2) { // Modify 30x -> 10x
+ if (result == 0) { // Processed successfully
+ if (previousStatus == EntityStatus.AGENT_ADD.getCode()) {
+ nextStatus = EntityStatus.AGENT_NORMAL.getCode();
+ } else if (previousStatus == EntityStatus.AGENT_DELETE.getCode()) {
+ nextStatus = EntityStatus.AGENT_DISABLE.getCode();
}
+ } else if (result == 1) { // Processing failed
+ nextStatus = EntityStatus.AGENT_FAILURE.getCode();
}
+
+ SourceFileDetailEntity update = new SourceFileDetailEntity();
+ update.setId(command.getTaskId());
+ update.setStatus(nextStatus);
+ update.setPreviousStatus(previousStatus);
+ fileDetailMapper.updateByPrimaryKeySelective(update);
}
}
}
@@ -273,7 +273,7 @@ public class AgentTaskServiceImpl implements AgentTaskService {
}
}
- config.setAdditionalAttr(s.toString().substring(0, s.toString().length() - 1));
+ config.setAdditionalAttr(s.substring(0, s.toString().length() - 1));
}
@Override
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceDbServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceDbServiceImpl.java
index 7f9af99..b6b1ead 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceDbServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceDbServiceImpl.java
@@ -338,7 +338,7 @@ public class SourceDbServiceImpl implements SourceDbService {
}
SourceDbDetailEntity dbEntity = CommonBeanUtils.copyProperties(detailInfo, SourceDbDetailEntity::new);
- dbEntity.setStatus(EntityStatus.DATA_RESOURCE_NEW.getCode());
+ dbEntity.setStatus(EntityStatus.AGENT_ADD.getCode());
dbEntity.setCreator(operator);
dbEntity.setModifier(operator);
dbEntity.setCreateTime(new Date());
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceFileServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceFileServiceImpl.java
index f641e78..12ee4bb 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceFileServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/SourceFileServiceImpl.java
@@ -33,6 +33,7 @@ import org.apache.inlong.manager.common.pojo.datasource.SourceFileDetailInfo;
import org.apache.inlong.manager.common.pojo.datasource.SourceFileDetailListVO;
import org.apache.inlong.manager.common.pojo.datasource.SourceFileDetailPageRequest;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.dao.entity.BusinessEntity;
import org.apache.inlong.manager.dao.entity.SourceFileBasicEntity;
import org.apache.inlong.manager.dao.entity.SourceFileDetailEntity;
@@ -40,7 +41,6 @@ import org.apache.inlong.manager.dao.mapper.BusinessEntityMapper;
import org.apache.inlong.manager.dao.mapper.SourceFileBasicEntityMapper;
import org.apache.inlong.manager.dao.mapper.SourceFileDetailEntityMapper;
import org.apache.inlong.manager.service.core.SourceFileService;
-import org.apache.inlong.manager.common.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
@@ -179,11 +179,31 @@ public class SourceFileServiceImpl implements SourceFileService {
Preconditions.checkNotNull(detailInfo.getInlongStreamId(), BizConstant.STREAM_ID_IS_EMPTY);
// Check if it can be added
- this.checkBizIsTempStatus(detailInfo.getInlongGroupId());
+ BusinessEntity businessEntity = this.checkBizIsTempStatus(detailInfo.getInlongGroupId());
+
+ // If there are data sources under the same groupId, streamId, ip, username, the addition fails
+ String groupId = detailInfo.getInlongGroupId();
+ String streamId = detailInfo.getInlongStreamId();
+ String ip = detailInfo.getIp();
+ String username = detailInfo.getUsername();
+ Integer count = fileDetailMapper.selectDetailExist(groupId, streamId, ip, username);
+ if (count > 0) {
+ LOGGER.error("file data source already exists: groupId=" + groupId + ", streamId=" + streamId
+ + ", ip=" + ip + ", username=" + username);
+ throw new BusinessException(BizErrorCodeEnum.DATA_SOURCE_DUPLICATE);
+ }
+
+ detailInfo.setStatus(EntityStatus.AGENT_ADD.getCode());
+ SourceFileDetailEntity detailEntity = CommonBeanUtils.copyProperties(detailInfo, SourceFileDetailEntity::new);
+ detailEntity.setCreator(operator);
+ detailEntity.setModifier(operator);
+ Date now = new Date();
+ detailEntity.setCreateTime(now);
+ detailEntity.setModifyTime(now);
+ fileDetailMapper.insertSelective(detailEntity);
- int id = saveDetailOpt(detailInfo, operator);
LOGGER.info("success to save file data source detail");
- return id;
+ return detailEntity.getId();
}
@Override
@@ -240,28 +260,25 @@ public class SourceFileServiceImpl implements SourceFileService {
public boolean updateDetail(SourceFileDetailInfo detailInfo, String operator) {
LOGGER.info("begin to update file data source detail={}", detailInfo);
Preconditions.checkNotNull(detailInfo, "file data source detail is empty");
- Preconditions.checkNotNull(detailInfo.getInlongGroupId(), BizConstant.GROUP_ID_IS_EMPTY);
- Preconditions.checkNotNull(detailInfo.getInlongStreamId(), BizConstant.STREAM_ID_IS_EMPTY);
- // The groupId may be modified, it is necessary to determine whether the business status of
- // the modified groupId supports modification
- this.checkBizIsTempStatus(detailInfo.getInlongGroupId());
+ Integer id = detailInfo.getId();
+ Preconditions.checkNotNull(id, BizConstant.ID_IS_EMPTY);
- // id exists, update, otherwise add
- if (detailInfo.getId() != null) {
- SourceFileDetailEntity entity = fileDetailMapper.selectByPrimaryKey(detailInfo.getId());
- if (entity == null) {
- LOGGER.error("file data source detail not found by id=" + detailInfo.getIp());
- throw new BusinessException(BizErrorCodeEnum.DATA_SOURCE_DETAIL_NOT_FOUND);
- }
- BeanUtils.copyProperties(detailInfo, entity);
- entity.setStatus(EntityStatus.BIZ_CONFIG_ING.getCode());
- entity.setModifier(operator);
- fileDetailMapper.updateByPrimaryKeySelective(entity);
- } else {
- saveDetailOpt(detailInfo, operator);
+ SourceFileDetailEntity entity = fileDetailMapper.selectByPrimaryKey(id);
+ if (entity == null) {
+ LOGGER.error("file data source detail not found by id=" + id);
+ throw new BusinessException(BizErrorCodeEnum.DATA_SOURCE_DETAIL_NOT_FOUND);
}
+ // After the approval is passed, the status needs to be revised to be revised to be issued: 205
+ this.checkBizIsTempStatus(detailInfo.getInlongGroupId());
+ detailInfo.setStatus(EntityStatus.AGENT_ADD.getCode());
+
+ SourceFileDetailEntity updateEntity = CommonBeanUtils.copyProperties(detailInfo, SourceFileDetailEntity::new);
+ updateEntity.setModifier(operator);
+ updateEntity.setModifyTime(new Date());
+ fileDetailMapper.updateByPrimaryKeySelective(updateEntity);
+
LOGGER.info("success to update file data source detail");
return true;
}
@@ -279,9 +296,18 @@ public class SourceFileServiceImpl implements SourceFileService {
}
// Check if it can be deleted
- this.checkBizIsTempStatus(entity.getInlongGroupId());
+ BusinessEntity bizEntity = this.checkBizIsTempStatus(entity.getInlongGroupId());
- entity.setIsDeleted(1);
+ // After the approval is passed, the status needs to be modified to delete to be issued: 204
+ if (EntityStatus.BIZ_CONFIG_SUCCESSFUL.getCode().equals(bizEntity.getStatus())) {
+ entity.setPreviousStatus(entity.getStatus());
+ entity.setStatus(EntityStatus.AGENT_DELETE.getCode());
+ } else {
+ entity.setPreviousStatus(entity.getStatus());
+ entity.setStatus(EntityStatus.AGENT_DISABLE.getCode());
+ }
+
+ entity.setIsDeleted(EntityStatus.IS_DELETED.getCode());
entity.setModifier(operator);
int resultCount = fileDetailMapper.updateByPrimaryKey(entity);
@@ -322,39 +348,12 @@ public class SourceFileServiceImpl implements SourceFileService {
}
/**
- * Save data source details
- */
- @Transactional(rollbackFor = Throwable.class)
- int saveDetailOpt(SourceFileDetailInfo detailInfo, String operator) {
- // If there are data sources under the same groupId, streamId, ip, username, the addition fails
- String groupId = detailInfo.getInlongGroupId();
- String streamId = detailInfo.getInlongStreamId();
- String ip = detailInfo.getIp();
- String username = detailInfo.getUsername();
- Integer count = fileDetailMapper.selectDetailExist(groupId, streamId, ip, username);
- if (count > 0) {
- LOGGER.error("file data source already exists: groupId=" + groupId + ", streamId=" + streamId
- + ", ip=" + ip + ", username=" + username);
- throw new BusinessException(BizErrorCodeEnum.DATA_SOURCE_DUPLICATE);
- }
-
- SourceFileDetailEntity detailEntity = CommonBeanUtils.copyProperties(detailInfo, SourceFileDetailEntity::new);
- detailEntity.setStatus(EntityStatus.DATA_RESOURCE_NEW.getCode());
- detailEntity.setCreator(operator);
- detailEntity.setModifier(operator);
- detailEntity.setCreateTime(new Date());
- fileDetailMapper.insertSelective(detailEntity);
-
- return detailEntity.getId();
- }
-
- /**
* Check whether the business status is temporary
*
* @param groupId Business group id
* @return Business entity for caller reuse
*/
- private void checkBizIsTempStatus(String groupId) {
+ private BusinessEntity checkBizIsTempStatus(String groupId) {
BusinessEntity businessEntity = businessMapper.selectByIdentifier(groupId);
Preconditions.checkNotNull(businessEntity, "groupId is invalid");
// Add/modify/delete is not allowed under certain business status
@@ -362,6 +361,8 @@ public class SourceFileServiceImpl implements SourceFileService {
LOGGER.error("business status was not allowed to add/update/delete data source info");
throw new BusinessException(BizErrorCodeEnum.DATA_SOURCE_OPT_NOT_ALLOWED);
}
+
+ return businessEntity;
}
}
\ No newline at end of file
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/business/listener/BusinessCompleteProcessListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/business/listener/BusinessCompleteProcessListener.java
index ad56542..c1e41c1 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/business/listener/BusinessCompleteProcessListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/business/listener/BusinessCompleteProcessListener.java
@@ -19,6 +19,7 @@ package org.apache.inlong.manager.service.workflow.business.listener;
import lombok.extern.slf4j.Slf4j;
import org.apache.inlong.manager.common.enums.EntityStatus;
+import org.apache.inlong.manager.dao.mapper.SourceFileDetailEntityMapper;
import org.apache.inlong.manager.service.core.BusinessService;
import org.apache.inlong.manager.service.core.DataStreamService;
import org.apache.inlong.manager.service.workflow.business.BusinessResourceWorkflowForm;
@@ -41,6 +42,8 @@ public class BusinessCompleteProcessListener implements ProcessEventListener {
private BusinessService businessService;
@Autowired
private DataStreamService dataStreamService;
+ @Autowired
+ private SourceFileDetailEntityMapper fileDetailMapper;
@Override
public ProcessEvent event() {
@@ -62,6 +65,8 @@ public class BusinessCompleteProcessListener implements ProcessEventListener {
businessService.updateStatus(groupId, EntityStatus.BIZ_CONFIG_SUCCESSFUL.getCode(), username);
// update data stream status
dataStreamService.updateStatus(groupId, null, EntityStatus.DATA_STREAM_CONFIG_SUCCESSFUL.getCode(), username);
+ // update file data source status
+ fileDetailMapper.updateStatusAfterApprove(groupId, null, EntityStatus.AGENT_ADD.getCode(), username);
return ListenerResult.success();
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/StreamCompleteProcessListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/StreamCompleteProcessListener.java
index 08cdff8..eab472f 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/StreamCompleteProcessListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/StreamCompleteProcessListener.java
@@ -19,6 +19,7 @@ package org.apache.inlong.manager.service.workflow.stream;
import lombok.extern.slf4j.Slf4j;
import org.apache.inlong.manager.common.enums.EntityStatus;
+import org.apache.inlong.manager.dao.mapper.SourceFileDetailEntityMapper;
import org.apache.inlong.manager.service.core.BusinessService;
import org.apache.inlong.manager.service.core.DataStreamService;
import org.apache.inlong.manager.service.workflow.business.BusinessResourceWorkflowForm;
@@ -41,6 +42,8 @@ public class StreamCompleteProcessListener implements ProcessEventListener {
private BusinessService businessService;
@Autowired
private DataStreamService streamService;
+ @Autowired
+ private SourceFileDetailEntityMapper fileDetailMapper;
@Override
public ProcessEvent event() {
@@ -56,12 +59,15 @@ public class StreamCompleteProcessListener implements ProcessEventListener {
BusinessResourceWorkflowForm form = (BusinessResourceWorkflowForm) context.getProcessForm();
String groupId = form.getInlongGroupId();
String streamId = form.getInlongStreamId();
- String username = context.getApplicant();
+ String user = context.getApplicant();
// update business status
- businessService.updateStatus(groupId, EntityStatus.BIZ_CONFIG_SUCCESSFUL.getCode(), username);
+ businessService.updateStatus(groupId, EntityStatus.BIZ_CONFIG_SUCCESSFUL.getCode(), user);
// update data stream status
- streamService.updateStatus(groupId, streamId, EntityStatus.DATA_STREAM_CONFIG_SUCCESSFUL.getCode(), username);
+ streamService.updateStatus(groupId, streamId, EntityStatus.DATA_STREAM_CONFIG_SUCCESSFUL.getCode(), user);
+ // update file data source status
+ fileDetailMapper.updateStatusAfterApprove(groupId, streamId, EntityStatus.AGENT_ADD.getCode(), user);
+
return ListenerResult.success();
}