test/DolphinScheduler
1
0
Fork 0
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git synced 2024-12-04 05:09:48 +08:00
Code Issues Actions 10 Packages Projects Releases Wiki Activity

Fix insertOrUpdate plugin may failed due to concurrent operation (#11471) (#11996)

Browse Source 
(cherry picked from commit 864a90820d)
This commit is contained in:
Wenjun Ruan 2022-09-17 08:59:45 +08:00 committed by GitHub
parent 5e42f52bdf
commit 8eadf5e5aa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 32 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/ApiApplicationServer.java
View File

@ -21,7 +21,6 @@ import org.apache.dolphinscheduler.common.enums.PluginType;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.plugin.task.api.TaskChannelFactory;
import org.apache.dolphinscheduler.plugin.task.api.TaskPluginException;
import org.apache.dolphinscheduler.service.task.TaskPluginManager;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
@ -68,10 +67,7 @@ public class ApiApplicationServer {
String paramsJson = PluginParamsTransfer.transferParamsToJson(params);
PluginDefine pluginDefine = new PluginDefine(taskPluginName, PluginType.TASK.getDesc(), paramsJson);
int count = pluginDao.addOrUpdatePluginDefine(pluginDefine);
if (count <= 0) {
throw new TaskPluginException("Failed to update task plugin: " + taskPluginName);
}
pluginDao.addOrUpdatePluginDefine(pluginDefine);
}
}
}

38
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/PluginDao.java
View File

@ -21,12 +21,20 @@ import static java.util.Objects.requireNonNull;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import org.apache.dolphinscheduler.plugin.task.api.TaskPluginException;
import java.util.Objects;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class PluginDao {
@Autowired
private PluginDefineMapper pluginDefineMapper;
@ -43,20 +51,36 @@ public class PluginDao {
* add or update plugin define
*
* @param pluginDefine new pluginDefine
* @return plugin id
*/
public int addOrUpdatePluginDefine(PluginDefine pluginDefine) {
requireNonNull(pluginDefine, "pluginDefine is null");
public int addOrUpdatePluginDefine(@NonNull PluginDefine pluginDefine) {
requireNonNull(pluginDefine.getPluginName(), "pluginName is null");
requireNonNull(pluginDefine.getPluginType(), "pluginType is null");
PluginDefine currPluginDefine = pluginDefineMapper.queryByNameAndType(pluginDefine.getPluginName(), pluginDefine.getPluginType());
PluginDefine currPluginDefine =
pluginDefineMapper.queryByNameAndType(pluginDefine.getPluginName(), pluginDefine.getPluginType());
if (currPluginDefine == null) {
if (pluginDefineMapper.insert(pluginDefine) == 1 && pluginDefine.getId() > 0) {
return pluginDefine.getId();
try {
if (pluginDefineMapper.insert(pluginDefine) == 1 && pluginDefine.getId() > 0) {
return pluginDefine.getId();
}
throw new TaskPluginException(
String.format("Failed to insert plugin definition, pluginName: %s, pluginType: %s",
pluginDefine.getPluginName(), pluginDefine.getPluginType()));
} catch (TaskPluginException ex) {
throw ex;
} catch (Exception ex) {
log.error("Insert plugin definition error, there may already exist a plugin", ex);
currPluginDefine = pluginDefineMapper.queryByNameAndType(pluginDefine.getPluginName(),
pluginDefine.getPluginType());
if (currPluginDefine == null) {
throw new TaskPluginException(
String.format("Failed to insert plugin definition, pluginName: %s, pluginType: %s",
pluginDefine.getPluginName(), pluginDefine.getPluginType()));
}
}
throw new IllegalStateException("Failed to insert plugin definition");
}
if (!currPluginDefine.getPluginParams().equals(pluginDefine.getPluginParams())) {
if (!Objects.equals(currPluginDefine.getPluginParams(), pluginDefine.getPluginParams())) {
currPluginDefine.setUpdateTime(pluginDefine.getUpdateTime());
currPluginDefine.setPluginParams(pluginDefine.getPluginParams());
pluginDefineMapper.updateById(currPluginDefine);