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 895f42ca3b [INLONG-8564][Manager] Fix the task not working of deleting
StreamSources (#8593)
895f42ca3b is described below
commit 895f42ca3b8726c0399dc1041b2bfc11253af744
Author: healchow <[email protected]>
AuthorDate: Fri Jul 28 10:39:41 2023 +0800
[INLONG-8564][Manager] Fix the task not working of deleting StreamSources
(#8593)
---
.../dao/mapper/InlongGroupEntityMapper.java | 7 ++-
.../dao/mapper/StreamSourceEntityMapper.java | 8 ++-
.../resources/mappers/InlongGroupEntityMapper.xml | 18 +-----
.../resources/mappers/StreamSourceEntityMapper.xml | 41 +++++++++-----
.../service/group/coordinator/Coordinator.java | 31 ----------
.../service/source/StreamSourceServiceImpl.java | 3 +-
...inatorTask.java => DeleteStreamSourceTask.java} | 66 ++++++++++++----------
.../src/main/resources/application-dev.properties | 5 +-
.../src/main/resources/application-prod.properties | 6 +-
.../src/main/resources/application-test.properties | 5 ++
10 files changed, 92 insertions(+), 98 deletions(-)
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java
index 7d5b7bf324..3e8189b301 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/InlongGroupEntityMapper.java
@@ -76,6 +76,7 @@ public interface InlongGroupEntityMapper {
* @param limit max item count
* @return all matched group ids
*/
+ @MultiTenantQuery(with = false)
List<String> selectDeletedGroupIdsWithTimeBefore(@Param("timeBefore") Date
timeBefore,
@Param("limit") Integer limit);
@@ -84,10 +85,10 @@ public interface InlongGroupEntityMapper {
*
* @param timeAfter the latest modify time after which to select
* @param limit max item count
- * @return
+ * @return all matched group ids
*/
- List<String> selectDeletedGroupIdsWithTimeAfter(@Param("timeAfter") Date
timeAfter,
- @Param("limit") Integer limit);
+ @MultiTenantQuery(with = false)
+ List<String> selectDeletedGroupIdsWithTimeAfter(@Param("timeAfter") Date
timeAfter, @Param("limit") Integer limit);
int updateByPrimaryKey(InlongGroupEntity record);
diff --git
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceEntityMapper.java
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceEntityMapper.java
index 805c73206a..ca984fb04b 100644
---
a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceEntityMapper.java
+++
b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/StreamSourceEntityMapper.java
@@ -135,9 +135,6 @@ public interface StreamSourceEntityMapper {
int updateByPrimaryKeySelective(StreamSourceEntity record);
- int updateByRelatedId(@Param("groupId") String groupId, @Param("streamId")
String streamId,
- @Param("status") Integer status);
-
int updateByPrimaryKey(StreamSourceEntity record);
/**
@@ -193,6 +190,11 @@ public interface StreamSourceEntityMapper {
*/
void updateStatusByDeleted();
+ int logicalDeleteByRelatedId(@Param("groupId") String groupId,
@Param("streamId") String streamId,
+ @Param("status") Integer status);
+
+ int logicalDeleteByIds(@Param("idList") List<Integer> idList,
@Param("status") Integer status);
+
/**
* Logical delete stream source by agentIp, change status at same time.
*
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
index 600638c721..923b5f1b53 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
@@ -391,32 +391,20 @@
</where>
</select>
<select id="selectDeletedGroupIdsWithTimeBefore"
resultType="java.lang.String">
- <bind name="_isInlongService" value="LoginUser.InlongService"/>
select inlong_group_id
from inlong_group
- <where>
- modify_time <= #{timeBefore, jdbcType=TIMESTAMP}
- <if test="_isInlongService == false">
- and tenant = #{tenant,jdbcType=VARCHAR}
- </if>
- </where>
+ where modify_time <= #{timeBefore, jdbcType=TIMESTAMP}
group by inlong_group_id
having min(is_deleted) > 0
limit #{limit, jdbcType=INTEGER}
</select>
<select id="selectDeletedGroupIdsWithTimeAfter"
resultType="java.lang.String">
- <bind name="_isInlongService" value="LoginUser.InlongService"/>
select inlong_group_id
from inlong_group
- <where>
- modify_time >= #{timeAfter, jdbcType=TIMESTAMP}
- <if test="_isInlongService == false">
- and tenant = #{tenant,jdbcType=VARCHAR}
- </if>
- </where>
+ where modify_time >= #{timeAfter, jdbcType=TIMESTAMP}
group by inlong_group_id
having min(is_deleted) > 0
- limit #{limit, jdbcType=INTEGER}
+ limit #{limit, jdbcType=INTEGER}
</select>
<update id="updateByPrimaryKey"
parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
diff --git
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml
index 7ae871cad4..d927608c57 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/StreamSourceEntityMapper.xml
@@ -360,18 +360,6 @@
and inlong_cluster_name = #{clusterName, jdbcType=VARCHAR}
</where>
</select>
- <update id="updateByRelatedId">
- update stream_source
- <set>
- is_deleted = id,
- previous_status = status,
- status = #{status, jdbcType=INTEGER},
- version = version + 1
- </set>
- where is_deleted = 0
- and inlong_group_id = #{groupId, jdbcType=VARCHAR}
- and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
- </update>
<update id="updateByPrimaryKeySelective"
parameterType="org.apache.inlong.manager.dao.entity.StreamSourceEntity">
update stream_source
@@ -551,7 +539,6 @@
and status not in (101, 102, 104, 105, 110)
</where>
</update>
-
<update id="updateStatusByDeleted">
update stream_source
<set>
@@ -563,6 +550,33 @@
and status not in (99, 201, 301)
</where>
</update>
+
+ <update id="logicalDeleteByRelatedId">
+ update stream_source
+ <set>
+ is_deleted = id,
+ previous_status = status,
+ status = #{status, jdbcType=INTEGER},
+ version = version + 1
+ </set>
+ where is_deleted = 0
+ and inlong_group_id = #{groupId, jdbcType=VARCHAR}
+ and inlong_stream_id = #{streamId, jdbcType=VARCHAR}
+ </update>
+
+ <update id="logicalDeleteByIds">
+ update stream_source
+ <set>
+ is_deleted = id,
+ previous_status = status,
+ status = #{status, jdbcType=INTEGER},
+ version = version + 1
+ </set>
+ where id in
+ <foreach item="item" index="index" collection="idList" open="("
close=")" separator=",">
+ #{item}
+ </foreach>
+ </update>
<update id="logicalDeleteByAgentIp">
update stream_source
<set>
@@ -577,6 +591,7 @@
and status = #{targetStatus, jdbcType=INTEGER}
</if>
</update>
+
<delete id="deleteByRelatedId">
delete
from stream_source
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/coordinator/Coordinator.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/coordinator/Coordinator.java
deleted file mode 100644
index 95ba2fc8e8..0000000000
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/coordinator/Coordinator.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.inlong.manager.service.group.coordinator;
-
-/**
- * Operator which ensure the consistency of the state of all components within
the group
- */
-public interface Coordinator {
-
- /**
- * Make inner state of one group to eventual consistency
- * @param inlongGroupId
- */
- public void coordinate(String inlongGroupId);
-
-}
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
index c18aef1fe9..fe6109fba1 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
@@ -494,7 +494,8 @@ public class StreamSourceServiceImpl implements
StreamSourceService {
Preconditions.expectNotBlank(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY);
Preconditions.expectNotBlank(streamId,
ErrorCodeEnum.STREAM_ID_IS_EMPTY);
- int sourceCount = sourceMapper.updateByRelatedId(groupId, streamId,
SourceStatus.TO_BE_ISSUED_DELETE.getCode());
+ int sourceCount = sourceMapper.logicalDeleteByRelatedId(groupId,
streamId,
+ SourceStatus.TO_BE_ISSUED_DELETE.getCode());
int fieldCount = sourceFieldMapper.updateByRelatedId(groupId,
streamId);
LOGGER.info("success to force delete source for groupId={} and
streamId={} by user={},"
+ " update {} sources and {} fields", groupId, streamId,
operator, sourceCount, fieldCount);
diff --git
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DelGroupCoordinatorTask.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DeleteStreamSourceTask.java
similarity index 58%
rename from
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DelGroupCoordinatorTask.java
rename to
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DeleteStreamSourceTask.java
index dcbd21e84d..f4def6697f 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DelGroupCoordinatorTask.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DeleteStreamSourceTask.java
@@ -21,7 +21,6 @@ import org.apache.inlong.manager.common.enums.SourceStatus;
import org.apache.inlong.manager.dao.entity.StreamSourceEntity;
import org.apache.inlong.manager.dao.mapper.InlongGroupEntityMapper;
import org.apache.inlong.manager.dao.mapper.StreamSourceEntityMapper;
-import org.apache.inlong.manager.service.group.coordinator.Coordinator;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
@@ -35,67 +34,76 @@ import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.TimerTask;
-import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
import java.util.concurrent.TimeUnit;
/**
- * Stop all stream source which is running when group is deleted
+ * Delete all stream source which is running but its inlong group was deleted.
*/
@Slf4j
@Service
-public class DelGroupCoordinatorTask extends TimerTask implements Coordinator,
InitializingBean {
+public class DeleteStreamSourceTask extends TimerTask implements
InitializingBean {
- private static final int INITIAL_DELAY = 300;
- private static final int INTERVAL = 1800;
+ private static final int INITIAL_DELAY_MINUTES = 5;
+ private static final int INTERVAL_MINUTES = 60;
- @Value("${group.compromise.batchSize:100}")
+ @Value("${group.deleted.batchSize:100}")
private Integer batchSize;
+ @Value("${group.deleted.latest.hours:10}")
+ private Integer latestHours;
+
@Autowired
private InlongGroupEntityMapper groupMapper;
@Autowired
private StreamSourceEntityMapper sourceMapper;
@Override
- public void afterPropertiesSet() throws Exception {
- log.info("start delete group compromise task");
- ScheduledExecutorService executor =
- Executors.newSingleThreadScheduledExecutor(new
ThreadFactoryBuilder().setNameFormat("group-compromise"
- + "-%s").build());
- executor.scheduleWithFixedDelay(this, INITIAL_DELAY, INTERVAL,
TimeUnit.SECONDS);
+ public void afterPropertiesSet() {
+ ThreadFactory threadFactory = new
ThreadFactoryBuilder().setNameFormat("inlong-group-delete-%s").build();
+ ScheduledExecutorService executor = new ScheduledThreadPoolExecutor(1,
threadFactory, new AbortPolicy());
+ executor.scheduleWithFixedDelay(this, INITIAL_DELAY_MINUTES,
INTERVAL_MINUTES, TimeUnit.MINUTES);
+
+ log.info("success to start the delete stream source task");
}
@Override
public void run() {
- LocalDateTime now = LocalDateTime.now();
- LocalDateTime twoHoursAgo =
now.minusHours(2).truncatedTo(ChronoUnit.HOURS);
- Date modifyTime =
Date.from(twoHoursAgo.atZone(ZoneId.systemDefault()).toInstant());
+ LocalDateTime currentTime = LocalDateTime.now();
+ LocalDateTime latestTime =
currentTime.minusHours(latestHours).truncatedTo(ChronoUnit.HOURS);
+ Date modifyTime =
Date.from(latestTime.atZone(ZoneId.systemDefault()).toInstant());
+
List<String> groupIds =
groupMapper.selectDeletedGroupIdsWithTimeAfter(modifyTime, batchSize);
if (CollectionUtils.isEmpty(groupIds)) {
return;
}
- for (String groupId : groupIds) {
- coordinate(groupId);
- }
+
+ deleteSources(groupIds);
}
- @Override
- public void coordinate(String inlongGroupId) {
- List<StreamSourceEntity> sourceList =
sourceMapper.selectByRelatedId(inlongGroupId, null, null);
+ private void deleteSources(List<String> inlongGroupIds) {
+ List<StreamSourceEntity> sourceList =
sourceMapper.selectByGroupIds(inlongGroupIds);
if (CollectionUtils.isEmpty(sourceList)) {
return;
}
+
+ List<Integer> idList = new ArrayList<>();
for (StreamSourceEntity source : sourceList) {
- if
(SourceStatus.SOURCE_NORMAL.getCode().equals(source.getStatus()) &&
StringUtils.isNotBlank(
- source.getInlongClusterNodeGroup())) {
- source.setPreviousStatus(source.getStatus());
- source.setStatus(SourceStatus.TO_BE_ISSUED_DELETE.getCode());
- source.setIsDeleted(source.getId());
- sourceMapper.updateByPrimaryKey(source);
+ if (SourceStatus.SOURCE_NORMAL.getCode().equals(source.getStatus())
+ &&
StringUtils.isNotBlank(source.getInlongClusterNodeGroup())) {
+ idList.add(source.getId());
}
}
+
+ if (CollectionUtils.isNotEmpty(idList)) {
+ sourceMapper.logicalDeleteByIds(idList,
SourceStatus.TO_BE_ISSUED_DELETE.getCode());
+ log.info("success to delete stream source with id in {}", idList);
+ }
}
-}
\ No newline at end of file
+}
diff --git
a/inlong-manager/manager-web/src/main/resources/application-dev.properties
b/inlong-manager/manager-web/src/main/resources/application-dev.properties
index 5f35ddc8f7..9e77553ac5 100644
--- a/inlong-manager/manager-web/src/main/resources/application-dev.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-dev.properties
@@ -98,4 +98,7 @@ source.update.interval=60
source.cleansing.enabled=false
source.cleansing.interval=600
-
+# Select the InlongGroupIds whose latest modification time is within how many
hours, the default is 10 hours
+group.deleted.latest.hours=10
+# The maximum size when querying InlongGroupIds in batches, those
InlongGroupIds will be used to delete the related StreamSources.
+group.deleted.batchSize=100
diff --git
a/inlong-manager/manager-web/src/main/resources/application-prod.properties
b/inlong-manager/manager-web/src/main/resources/application-prod.properties
index e57a432945..c3bb35f3ce 100644
--- a/inlong-manager/manager-web/src/main/resources/application-prod.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-prod.properties
@@ -97,5 +97,7 @@ source.update.interval=60
source.cleansing.enabled=false
source.cleansing.interval=600
-# Group batch size to compromise in one period
-group.compromise.batchSize=100
+# Select the InlongGroupIds whose latest modification time is within how many
hours, the default is 10 hours
+group.deleted.latest.hours=10
+# The maximum size when querying InlongGroupIds in batches, those
InlongGroupIds will be used to delete the related StreamSources.
+group.deleted.batchSize=100
diff --git
a/inlong-manager/manager-web/src/main/resources/application-test.properties
b/inlong-manager/manager-web/src/main/resources/application-test.properties
index e0c0d88424..9e77553ac5 100644
--- a/inlong-manager/manager-web/src/main/resources/application-test.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-test.properties
@@ -97,3 +97,8 @@ source.update.interval=60
# If turned on, tasks in the incorrect state are periodically deleted
source.cleansing.enabled=false
source.cleansing.interval=600
+
+# Select the InlongGroupIds whose latest modification time is within how many
hours, the default is 10 hours
+group.deleted.latest.hours=10
+# The maximum size when querying InlongGroupIds in batches, those
InlongGroupIds will be used to delete the related StreamSources.
+group.deleted.batchSize=100
