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 ac201cdeb0 [INLONG-8586][Manager] Stop Stream Source which is still
running after group is stopped (#8587)
ac201cdeb0 is described below
commit ac201cdeb070ef8d85f961056e6001582546bae3
Author: kipshi <[email protected]>
AuthorDate: Wed Jul 26 13:05:09 2023 +0800
[INLONG-8586][Manager] Stop Stream Source which is still running after
group is stopped (#8587)
Co-authored-by: kipshi <[email protected]>
---
.../dao/mapper/InlongGroupEntityMapper.java | 13 ++-
.../resources/mappers/InlongGroupEntityMapper.xml | 20 +++-
.../service/group/coordinator/Coordinator.java | 31 +++++++
.../manager/service/task/DataCleansingTask.java | 2 +-
.../service/task/DelGroupCoordinatorTask.java | 101 +++++++++++++++++++++
.../src/main/resources/application-prod.properties | 3 +
6 files changed, 165 insertions(+), 5 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 82af114c67..7d5b7bf324 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,7 +76,18 @@ public interface InlongGroupEntityMapper {
* @param limit max item count
* @return all matched group ids
*/
- List<String> selectDeletedGroupIds(@Param("timeBefore") Date timeBefore,
@Param("limit") Integer limit);
+ List<String> selectDeletedGroupIdsWithTimeBefore(@Param("timeBefore") Date
timeBefore,
+ @Param("limit") Integer limit);
+
+ /**
+ * Select all groups which are logical deleted after the specified last
modify time
+ *
+ * @param timeAfter the latest modify time after which to select
+ * @param limit max item count
+ * @return
+ */
+ List<String> selectDeletedGroupIdsWithTimeAfter(@Param("timeAfter") Date
timeAfter,
+ @Param("limit") Integer limit);
int updateByPrimaryKey(InlongGroupEntity record);
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 6b763e475d..600638c721 100644
---
a/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
+++
b/inlong-manager/manager-dao/src/main/resources/mappers/InlongGroupEntityMapper.xml
@@ -390,20 +390,34 @@
and is_deleted = 0
</where>
</select>
- <select id="selectDeletedGroupIds" resultType="java.lang.String">
+ <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">
- tenant = #{tenant,jdbcType=VARCHAR}
+ and tenant = #{tenant,jdbcType=VARCHAR}
</if>
- and modify_time <= #{timeBefore, jdbcType=TIMESTAMP}
</where>
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>
+ group by inlong_group_id
+ having min(is_deleted) > 0
+ limit #{limit, jdbcType=INTEGER}
+ </select>
<update id="updateByPrimaryKey"
parameterType="org.apache.inlong.manager.dao.entity.InlongGroupEntity">
<bind name="_isInlongService" value="LoginUser.InlongService"/>
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
new file mode 100644
index 0000000000..95ba2fc8e8
--- /dev/null
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/coordinator/Coordinator.java
@@ -0,0 +1,31 @@
+/*
+ * 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/task/DataCleansingTask.java
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DataCleansingTask.java
index ce652f2042..871700c5b3 100644
---
a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DataCleansingTask.java
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DataCleansingTask.java
@@ -119,7 +119,7 @@ public class DataCleansingTask extends TimerTask implements
InitializingBean {
calendar.add(Calendar.DAY_OF_MONTH, -before);
Date daysBefore = calendar.getTime();
- List<String> groupIds =
groupMapper.selectDeletedGroupIds(daysBefore, batchSize);
+ List<String> groupIds =
groupMapper.selectDeletedGroupIdsWithTimeBefore(daysBefore, batchSize);
if (CollectionUtils.isEmpty(groupIds)) {
return;
}
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/DelGroupCoordinatorTask.java
new file mode 100644
index 0000000000..dcbd21e84d
--- /dev/null
+++
b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/task/DelGroupCoordinatorTask.java
@@ -0,0 +1,101 @@
+/*
+ * 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.task;
+
+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;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.temporal.ChronoUnit;
+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.TimeUnit;
+
+/**
+ * Stop all stream source which is running when group is deleted
+ */
+@Slf4j
+@Service
+public class DelGroupCoordinatorTask extends TimerTask implements Coordinator,
InitializingBean {
+
+ private static final int INITIAL_DELAY = 300;
+ private static final int INTERVAL = 1800;
+
+ @Value("${group.compromise.batchSize:100}")
+ private Integer batchSize;
+ @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);
+ }
+
+ @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());
+ List<String> groupIds =
groupMapper.selectDeletedGroupIdsWithTimeAfter(modifyTime, batchSize);
+ if (CollectionUtils.isEmpty(groupIds)) {
+ return;
+ }
+ for (String groupId : groupIds) {
+ coordinate(groupId);
+ }
+ }
+
+ @Override
+ public void coordinate(String inlongGroupId) {
+ List<StreamSourceEntity> sourceList =
sourceMapper.selectByRelatedId(inlongGroupId, null, null);
+ if (CollectionUtils.isEmpty(sourceList)) {
+ return;
+ }
+ 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);
+ }
+ }
+ }
+}
\ No newline at end of file
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 62245389ce..e57a432945 100644
--- a/inlong-manager/manager-web/src/main/resources/application-prod.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-prod.properties
@@ -96,3 +96,6 @@ source.update.interval=60
# If turned on, tasks in the incorrect state are periodically deleted
source.cleansing.enabled=false
source.cleansing.interval=600
+
+# Group batch size to compromise in one period
+group.compromise.batchSize=100
