You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2023/05/10 01:41:11 UTC
[inlong] branch master updated: [INLONG-7981][Manager] Failed to stop source correctly when suspend a group (#7982)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong.git
The following commit(s) were added to refs/heads/master by this push:
new 3c5f8e93f [INLONG-7981][Manager] Failed to stop source correctly when suspend a group (#7982)
3c5f8e93f is described below
commit 3c5f8e93fd1855d5e0f597ee7840fbc6283f5ebb
Author: fuweng11 <76...@users.noreply.github.com>
AuthorDate: Wed May 10 09:41:04 2023 +0800
[INLONG-7981][Manager] Failed to stop source correctly when suspend a group (#7982)
---
CHANGES.md | 1 +
.../manager/service/core/impl/AgentServiceImpl.java | 20 +++++++++++++-------
.../manager/service/core/impl/AgentServiceTest.java | 3 +++
3 files changed, 17 insertions(+), 7 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 908a75a46..cd4d5144d 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -47,6 +47,7 @@
### Manager
| ISSUE | Summary |
|:-----------------------------------------------------------:|:-----------------------------------------------------------------------------------------------|
+| [INLONG-7981](https://github.com/apache/inlong/issues/7981) | [Bug][Manager] Failed to stop source correctly when suspend a group |
| [INLONG-7948](https://github.com/apache/inlong/issues/7948) | [Improve][Manager] Add user authentication when operate inlong consume |
| [INLONG-7946](https://github.com/apache/inlong/issues/7946) | [Improve][Manager] Add user authentication when bind clusterTag |
| [INLONG-7941](https://github.com/apache/inlong/issues/7941) | [Improve][Manager][Dashborad] Query group to distinguish lightweight |
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
index b3a2d7e92..ff44bfb9c 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AgentServiceImpl.java
@@ -38,8 +38,8 @@ import org.apache.inlong.common.pojo.dataproxy.MQClusterInfo;
import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.consts.SourceType;
import org.apache.inlong.manager.common.enums.ClusterType;
+import org.apache.inlong.manager.common.enums.GroupStatus;
import org.apache.inlong.manager.common.enums.SourceStatus;
-import org.apache.inlong.manager.common.enums.StreamStatus;
import org.apache.inlong.manager.common.exceptions.BusinessException;
import org.apache.inlong.manager.common.util.CommonBeanUtils;
import org.apache.inlong.manager.common.util.JsonUtils;
@@ -336,7 +336,12 @@ public class AgentServiceImpl implements AgentService {
// find those node whose tag match stream_source tag and agent ip match stream_source agent ip
List<StreamSourceEntity> sourceEntities = sourceMapper.selectTemplateSourceByCluster(needCopiedStatusList,
Lists.newArrayList(SourceType.FILE), agentClusterName);
+ Set<GroupStatus> noNeedAddTask = Sets.newHashSet(GroupStatus.SUSPENDED, GroupStatus.SUSPENDING);
sourceEntities.forEach(sourceEntity -> {
+ InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId());
+ if (groupEntity != null && noNeedAddTask.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
+ return;
+ }
StreamSourceEntity subSource = sourceMapper.selectOneByTemplatedIdAndAgentIp(sourceEntity.getId(),
agentIp);
if (subSource == null) {
@@ -396,16 +401,17 @@ public class AgentServiceImpl implements AgentService {
}
// case: agent tag rebind and match source task again and stream is not in 'SUSPENDED' status
- InlongStreamEntity streamEntity = streamMapper.selectByIdentifier(
- sourceEntity.getInlongGroupId(), sourceEntity.getInlongStreamId());
+ InlongGroupEntity groupEntity = groupMapper.selectByGroupId(sourceEntity.getInlongGroupId());
Set<SourceStatus> exceptedMatchedSourceStatus = Sets.newHashSet(
SourceStatus.SOURCE_NORMAL,
SourceStatus.TO_BE_ISSUED_ADD,
SourceStatus.TO_BE_ISSUED_ACTIVE);
- Set<StreamStatus> exceptedMatchedStreamStatus = Sets.newHashSet(StreamStatus.SUSPENDED);
+ Set<GroupStatus> exceptedMatchedGroupStatus = Sets.newHashSet(GroupStatus.SUSPENDED,
+ GroupStatus.SUSPENDING);
if (matchGroup(sourceEntity, clusterNodeEntity)
+ && groupEntity != null
&& !exceptedMatchedSourceStatus.contains(SourceStatus.forCode(sourceEntity.getStatus()))
- && !exceptedMatchedStreamStatus.contains(StreamStatus.forCode(streamEntity.getStatus()))) {
+ && !exceptedMatchedGroupStatus.contains(GroupStatus.forCode(groupEntity.getStatus()))) {
LOGGER.info("Transform task({}) from {} to {} because tag rematch "
+ "for agent({}) in cluster({})", sourceEntity.getAgentIp(),
sourceEntity.getStatus(), SourceStatus.TO_BE_ISSUED_ACTIVE.getCode(),
@@ -463,14 +469,13 @@ public class AgentServiceImpl implements AgentService {
InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
InlongStreamEntity streamEntity = streamMapper.selectByIdentifier(groupId, streamId);
+ String extParams = entity.getExtParams();
if (groupEntity != null && streamEntity != null) {
- String extParams = entity.getExtParams();
dataConfig.setSyncSend(streamEntity.getSyncSend());
if (SourceType.FILE.equalsIgnoreCase(streamEntity.getDataType())) {
String dataSeparator = streamEntity.getDataSeparator();
extParams = (null != dataSeparator ? getExtParams(extParams, dataSeparator) : extParams);
}
- dataConfig.setExtParams(extParams);
int dataReportType = groupEntity.getDataReportType();
dataConfig.setDataReportType(dataReportType);
@@ -529,6 +534,7 @@ public class AgentServiceImpl implements AgentService {
LOGGER.warn("set syncSend=[0] as the stream not exists for groupId={}, streamId={}", groupId, streamId);
}
}
+ dataConfig.setExtParams(extParams);
return dataConfig;
}
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
index 2f85d1c9a..6670b87e6 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/core/impl/AgentServiceTest.java
@@ -30,6 +30,7 @@ import org.apache.inlong.common.pojo.agent.TaskResult;
import org.apache.inlong.common.pojo.agent.TaskSnapshotMessage;
import org.apache.inlong.common.pojo.agent.TaskSnapshotRequest;
import org.apache.inlong.manager.common.consts.SourceType;
+import org.apache.inlong.manager.common.enums.GroupStatus;
import org.apache.inlong.manager.common.enums.SourceStatus;
import org.apache.inlong.manager.common.enums.StreamStatus;
import org.apache.inlong.manager.common.exceptions.BusinessException;
@@ -141,6 +142,7 @@ class AgentServiceTest extends ServiceBaseTest {
sources.stream()
.filter(source -> source.getTemplateId() != null)
.forEach(source -> sourceService.stop(source.getId(), GLOBAL_OPERATOR));
+ groupMapper.updateStatus(groupId, GroupStatus.SUSPENDED.getCode(), GLOBAL_OPERATOR);
streamMapper.updateStatusByIdentifier(groupId, streamId, StreamStatus.SUSPENDED.getCode(), GLOBAL_OPERATOR);
}
@@ -152,6 +154,7 @@ class AgentServiceTest extends ServiceBaseTest {
sources.stream()
.filter(source -> source.getTemplateId() != null)
.forEach(source -> sourceService.restart(source.getId(), GLOBAL_OPERATOR));
+ groupMapper.updateStatus(groupId, GroupStatus.RESTARTED.getCode(), GLOBAL_OPERATOR);
streamMapper.updateStatusByIdentifier(groupId, streamId, StreamStatus.RESTARTED.getCode(), GLOBAL_OPERATOR);
}