github
/
DolphinScheduler
mirror of https://gitee.com/dolphinscheduler/DolphinScheduler.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Browse Source

cherry-pick (#12608) 

[improvement-#12416] Use CollectionUtils.isEmpty judge the collection isEmpty to avoid NPE
#12449

Co-authored-by: fuchanghai <33984497+fuchanghai@users.noreply.github.com>
3.1.1-release
Kerwin 2 years ago committed by GitHub
parent
94ef02e9de
commit
2c93dc6041
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 74 additions and 42 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      dolphinscheduler-alert/dolphinscheduler-alert-plugins/dolphinscheduler-alert-feishu/src/main/java/org/apache/dolphinscheduler/plugin/alert/feishu/FeiShuSender.java
  2. 4
      dolphinscheduler-alert/dolphinscheduler-alert-plugins/dolphinscheduler-alert-webexteams/src/main/java/org/apache/dolphinscheduler/plugin/alert/webexteams/WebexTeamsAlertChannel.java
  3. 2
      dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api.test/utils/JSONUtils.java
  4. 20
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/configuration/TaskTypeConfiguration.java
  5. 4
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/permission/ResourcePermissionCheckServiceImpl.java
  6. 2
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
  7. 2
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java
  8. 25
      dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
  9. 4
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/JSONUtils.java
  10. 4
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/NetUtils.java
  11. 2
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/OSUtils.java
  12. 4
      dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/PropertyUtils.java
  13. 6
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringGlobalParams.java
  14. 10
      dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
  15. 2
      dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parser/TimePlaceholderUtils.java
  16. 8
      dolphinscheduler-task-plugin/dolphinscheduler-task-switch/src/main/java/org/apache/dolphinscheduler/plugin/task/switchtask/SwitchParameters.java
  17. 14
      dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java

3
dolphinscheduler-alert/dolphinscheduler-alert-plugins/dolphinscheduler-alert-feishu/src/main/java/org/apache/dolphinscheduler/plugin/alert/feishu/FeiShuSender.java
Unescape View File

@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.alert.api.AlertResult;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.commons.codec.binary.StringUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.client.methods.CloseableHttpResponse;
@ -106,7 +107,7 @@ public final class FeiShuSender {
if (alertData.getContent() != null) {
List<Map> list = JSONUtils.toList(alertData.getContent(), Map.class);
if (list.isEmpty()) {
if (CollectionUtils.isEmpty(list)) {
return alertData.getTitle() + alertData.getContent();
}

4
dolphinscheduler-alert/dolphinscheduler-alert-plugins/dolphinscheduler-alert-webexteams/src/main/java/org/apache/dolphinscheduler/plugin/alert/webexteams/WebexTeamsAlertChannel.java
Unescape View File

@ -22,6 +22,8 @@ import org.apache.dolphinscheduler.alert.api.AlertData;
import org.apache.dolphinscheduler.alert.api.AlertInfo;
import org.apache.dolphinscheduler.alert.api.AlertResult;
import org.apache.commons.collections.MapUtils;
import java.util.Map;
public final class WebexTeamsAlertChannel implements AlertChannel {
@ -29,7 +31,7 @@ public final class WebexTeamsAlertChannel implements AlertChannel {
public AlertResult process(AlertInfo alertInfo) {
AlertData alertData = alertInfo.getAlertData();
Map<String, String> alertParams = alertInfo.getAlertParams();
if (alertParams == null || alertParams.size() == 0) {
if (MapUtils.isEmpty(alertParams)) {
return new AlertResult("false", "WebexTeams alert params is empty");
}

2
dolphinscheduler-api-test/dolphinscheduler-api-test-case/src/test/java/org/apache/dolphinscheduler/api.test/utils/JSONUtils.java
Unescape View File

@ -331,7 +331,7 @@ public class JSONUtils {
public static ObjectNode parseObject(String text) {
try {
if (text.isEmpty()) {
if (StringUtils.isEmpty(text)) {
return parseObject(text, ObjectNode.class);
} else {
return (ObjectNode) objectMapper.readTree(text);

20
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/configuration/TaskTypeConfiguration.java
Unescape View File

@ -25,6 +25,8 @@ import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
@ -61,16 +63,16 @@ public class TaskTypeConfiguration {
public Set<FavTaskDto> getDefaultTaskTypes() {
Set<FavTaskDto> defaultTaskTypes = new HashSet<>();
if (defaultTaskTypes.size() <= 0) {
universal.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_UNIVERSAL)));
cloud.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_CLOUD)));
logic.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_LOGIC)));
dataIntegration.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_DATA_INTEGRATION)));
dataQuality.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_DATA_QUALITY)));
machineLearning.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_MACHINE_LEARNING)));
other.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_OTHER)));
if (CollectionUtils.isNotEmpty(defaultTaskTypes)) {
return defaultTaskTypes;
}
universal.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_UNIVERSAL)));
cloud.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_CLOUD)));
logic.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_LOGIC)));
dataIntegration.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_DATA_INTEGRATION)));
dataQuality.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_DATA_QUALITY)));
machineLearning.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_MACHINE_LEARNING)));
other.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_OTHER)));
return defaultTaskTypes;
}

4
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/permission/ResourcePermissionCheckServiceImpl.java
Unescape View File

@ -109,7 +109,9 @@ public class ResourcePermissionCheckServiceImpl implements ResourcePermissionChe
Set<?> originResSet = new HashSet<>(Arrays.asList(needChecks));
Set<?> ownResSets = RESOURCE_LIST_MAP.get(authorizationType).listAuthorizedResource(userId, logger);
originResSet.removeAll(ownResSets);
return originResSet.isEmpty();
if (CollectionUtils.isNotEmpty(originResSet))
logger.warn("User does not have resource permission on associated resources, userId:{}", userId);
return CollectionUtils.isEmpty(originResSet);
}
return true;
}

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessDefinitionServiceImpl.java
Unescape View File

@ -773,7 +773,7 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
return result;
}
/**
/**
* Process definition want to delete whether used in other task, should throw exception when have be used.
*
* This function avoid delete process definition already dependencies by other tasks by accident.

2
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ProcessInstanceServiceImpl.java
Unescape View File

@ -811,7 +811,7 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce
ganttDto.setTaskNames(nodeList);
List<Task> taskList = new ArrayList<>();
if (!nodeList.isEmpty()) {
if (CollectionUtils.isNotEmpty(nodeList)) {
List<Long> taskCodes = nodeList.stream().map(Long::parseLong).collect(Collectors.toList());
List<TaskInstance> taskInstances = taskInstanceMapper.queryByProcessInstanceIdsAndTaskCodes(
Collections.singletonList(processInstanceId), taskCodes);

25
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TaskDefinitionServiceImpl.java
Unescape View File

@ -58,6 +58,7 @@ import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.task.TaskPluginManager;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import java.util.ArrayList;
@ -135,8 +136,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
}
List<TaskDefinitionLog> taskDefinitionLogs = JSONUtils.toList(taskDefinitionJson, TaskDefinitionLog.class);
if (taskDefinitionLogs.isEmpty()) {
logger.error("taskDefinitionJson invalid: {}", taskDefinitionJson);
if (CollectionUtils.isEmpty(taskDefinitionLogs)) {
logger.warn("Parameter taskDefinitionJson is invalid.");
putMsg(result, Status.DATA_IS_NOT_VALID, taskDefinitionJson);
return result;
}
@ -232,7 +233,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
// upstreamTaskCodes - queryUpStreamTaskCodes
Set<Long> diffCode = upstreamTaskCodes.stream().filter(code -> !queryUpStreamTaskCodes.contains(code))
.collect(Collectors.toSet());
if (!diffCode.isEmpty()) {
if (CollectionUtils.isNotEmpty(diffCode)) {
String taskCodes = StringUtils.join(diffCode, Constants.COMMA);
logger.error("Some task definitions with parameter upstreamCodes do not exist, taskDefinitionCodes:{}.",
taskCodes);
@ -351,8 +352,8 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
return result;
}
List<ProcessTaskRelation> processTaskRelationList =
processTaskRelationMapper.queryDownstreamByTaskCode(taskCode);
if (!processTaskRelationList.isEmpty()) {
processTaskRelationMapper.queryDownstreamByTaskCode(taskDefinition.getCode());
if (CollectionUtils.isNotEmpty(processTaskRelationList)) {
Set<Long> postTaskCodes = processTaskRelationList
.stream()
.map(ProcessTaskRelation::getPostTaskCode)
@ -445,7 +446,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
}
List<ProcessTaskRelation> taskRelationList =
processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
if (!taskRelationList.isEmpty()) {
if (CollectionUtils.isNotEmpty(taskRelationList)) {
logger.info(
"Task definition has upstream tasks, start handle them after update task, taskDefinitionCode:{}.",
taskCode);
@ -573,13 +574,13 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
}
}
Map<Long, TaskDefinition> queryUpStreamTaskCodeMap;
if (!upstreamTaskCodes.isEmpty()) {
if (CollectionUtils.isNotEmpty(upstreamTaskCodes)) {
List<TaskDefinition> upstreamTaskDefinitionList = taskDefinitionMapper.queryByCodeList(upstreamTaskCodes);
queryUpStreamTaskCodeMap = upstreamTaskDefinitionList.stream()
.collect(Collectors.toMap(TaskDefinition::getCode, taskDefinition -> taskDefinition));
// upstreamTaskCodes - queryUpStreamTaskCodeMap.keySet
upstreamTaskCodes.removeAll(queryUpStreamTaskCodeMap.keySet());
if (!upstreamTaskCodes.isEmpty()) {
if (CollectionUtils.isNotEmpty(upstreamTaskCodes)) {
String notExistTaskCodes = StringUtils.join(upstreamTaskCodes, Constants.COMMA);
logger.error("Some task definitions in parameter upstreamTaskCodes do not exist, notExistTaskCodes:{}.",
notExistTaskCodes);
@ -589,7 +590,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
} else {
queryUpStreamTaskCodeMap = new HashMap<>();
}
if (!upstreamTaskRelations.isEmpty()) {
if (CollectionUtils.isNotEmpty(upstreamTaskCodes)) {
ProcessTaskRelation taskRelation = upstreamTaskRelations.get(0);
List<ProcessTaskRelation> processTaskRelations =
processTaskRelationMapper.queryByProcessCode(projectCode, taskRelation.getProcessDefinitionCode());
@ -613,7 +614,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
taskRelation.setPreTaskVersion(queryUpStreamTask.getValue().getVersion());
processTaskRelationList.add(taskRelation);
}
if (queryUpStreamTaskCodeMap.isEmpty() && !processTaskRelationList.isEmpty()) {
if (MapUtils.isEmpty(queryUpStreamTaskCodeMap) && CollectionUtils.isNotEmpty(processTaskRelationList)) {
processTaskRelationList.add(processTaskRelationList.get(0));
}
updateDag(loginUser, result, taskRelation.getProcessDefinitionCode(), processTaskRelations,
@ -666,7 +667,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
if (switchVersion > 0) {
List<ProcessTaskRelation> taskRelationList =
processTaskRelationMapper.queryUpstreamByCode(projectCode, taskCode);
if (!taskRelationList.isEmpty()) {
if (CollectionUtils.isNotEmpty(taskRelationList)) {
logger.info(
"Task definition has upstream tasks, start handle them after switch task, taskDefinitionCode:{}.",
taskCode);
@ -795,7 +796,7 @@ public class TaskDefinitionServiceImpl extends BaseServiceImpl implements TaskDe
taskDefinitionMapper.queryDefineListPaging(page, projectCode, searchWorkflowName,
searchTaskName, taskType, taskExecuteType);
List<TaskMainInfo> records = taskMainInfoIPage.getRecords();
if (!records.isEmpty()) {
if (CollectionUtils.isNotEmpty(records)) {
Map<Long, TaskMainInfo> taskMainInfoMap = new HashMap<>();
for (TaskMainInfo info : records) {
taskMainInfoMap.compute(info.getTaskCode(), (k, v) -> {

4
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/JSONUtils.java
Unescape View File

@ -24,6 +24,8 @@ import static com.fasterxml.jackson.databind.MapperFeature.REQUIRE_SETTERS_FOR_G
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.dolphinscheduler.common.constants.DateConstants.YYYY_MM_DD_HH_MM_SS;
import org.apache.commons.lang3.StringUtils;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
@ -346,7 +348,7 @@ public class JSONUtils {
public static ObjectNode parseObject(String text) {
try {
if (text.isEmpty()) {
if (StringUtils.isEmpty(text)) {
return parseObject(text, ObjectNode.class);
} else {
return (ObjectNode) objectMapper.readTree(text);

4
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/NetUtils.java
Unescape View File

@ -21,6 +21,8 @@ import static java.util.Collections.emptyList;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.commons.collections.CollectionUtils;
import java.io.IOException;
import java.net.Inet6Address;
import java.net.InetAddress;
@ -264,7 +266,7 @@ public class NetUtils {
}
private static NetworkInterface findAddress(List<NetworkInterface> validNetworkInterfaces) {
if (validNetworkInterfaces.isEmpty()) {
if (CollectionUtils.isEmpty(validNetworkInterfaces)) {
return null;
}
String networkPriority = PropertyUtils.getString(Constants.DOLPHIN_SCHEDULER_NETWORK_PRIORITY_STRATEGY, NETWORK_PRIORITY_DEFAULT);

2
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/OSUtils.java
Unescape View File

@ -241,7 +241,7 @@ public class OSUtils {
int startPos = 0;
int endPos = lines.length - 2;
for (int i = 0; i < lines.length; i++) {
if (lines[i].isEmpty()) {
if (StringUtils.isEmpty(lines[i])) {
continue;
}

4
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/PropertyUtils.java
Unescape View File

@ -22,6 +22,8 @@ import static org.apache.dolphinscheduler.common.constants.Constants.COMMON_PROP
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.ResUploadType;
import org.apache.commons.collections.CollectionUtils;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
@ -280,7 +282,7 @@ public class PropertyUtils {
return null;
}
Set<Object> keys = properties.keySet();
if (keys.isEmpty()) {
if (CollectionUtils.isEmpty(keys)) {
return null;
}
Map<String, String> propertiesMap = new HashMap<>();

6
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/expand/CuringGlobalParams.java
Unescape View File

@ -149,7 +149,7 @@ public class CuringGlobalParams implements CuringParamsService {
parameters.setVarPool(taskInstance.getVarPool());
Map<String, Property> varParams = parameters.getVarPoolMap();
if (globalParams.isEmpty() && localParams.isEmpty() && varParams.isEmpty()) {
if (MapUtils.isEmpty(globalParams) && MapUtils.isEmpty(localParams) && MapUtils.isEmpty(varParams)) {
return null;
}
// if it is a complement,
@ -168,10 +168,10 @@ public class CuringGlobalParams implements CuringParamsService {
}
params.put(PARAMETER_TASK_INSTANCE_ID, Integer.toString(taskInstance.getId()));
if (varParams.size() != 0) {
if (MapUtils.isNotEmpty(varParams)) {
globalParams.putAll(varParams);
}
if (localParams.size() != 0) {
if (MapUtils.isNotEmpty(localParams)) {
globalParams.putAll(localParams);
}

10
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessServiceImpl.java
Unescape View File

@ -146,6 +146,8 @@ import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Nullable;
import org.apache.commons.collections.MapUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
@ -729,7 +731,7 @@ public class ProcessServiceImpl implements ProcessService {
queryReleaseSchedulerListByProcessDefinitionCode(command.getProcessDefinitionCode());
List<Date> complementDateList = CronUtils.getSelfFireDateList(start, end, schedules);
if (complementDateList.size() > 0) {
if (CollectionUtils.isNotEmpty(complementDateList)) {
scheduleTime = complementDateList.get(0);
} else {
logger.error("set scheduler time error: complement date list is empty, command: {}",
@ -823,7 +825,7 @@ public class ProcessServiceImpl implements ProcessService {
// set start param into global params
Map<String, String> globalMap = processDefinition.getGlobalParamMap();
List<Property> globalParamList = processDefinition.getGlobalParamList();
if (startParamMap.size() > 0 && globalMap != null) {
if (MapUtils.isNotEmpty(startParamMap) && globalMap != null) {
// start param to overwrite global param
for (Map.Entry<String, String> param : globalMap.entrySet()) {
String val = startParamMap.get(param.getKey());
@ -1137,7 +1139,7 @@ public class ProcessServiceImpl implements ProcessService {
complementDate = CronUtils.getSelfScheduleDateList(cmdParam);
}
if (complementDate.size() > 0 && Flag.NO == processInstance.getIsSubProcess()) {
if (CollectionUtils.isNotEmpty(complementDate) && Flag.NO == processInstance.getIsSubProcess()) {
processInstance.setScheduleTime(complementDate.get(0));
}
@ -1451,7 +1453,7 @@ public class ProcessServiceImpl implements ProcessService {
}
processMapStr = JSONUtils.toJsonString(cmdParam);
}
if (fatherParams.size() != 0) {
if (MapUtils.isNotEmpty(fatherParams)) {
cmdParam.put(CMD_PARAM_FATHER_PARAMS, JSONUtils.toJsonString(fatherParams));
processMapStr = JSONUtils.toJsonString(cmdParam);
}

2
dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parser/TimePlaceholderUtils.java
Unescape View File

@ -254,7 +254,7 @@ public class TimePlaceholderUtils {
if (Character.isDigit(expression.charAt(i))) {
num = num + expression.charAt(i);
} else {
if (!num.isEmpty()) {
if (!StringUtils.isEmpty(num)) {
result.add(num);
}
result.add(expression.charAt(i) + "");

8
dolphinscheduler-task-plugin/dolphinscheduler-task-switch/src/main/java/org/apache/dolphinscheduler/plugin/task/switchtask/SwitchParameters.java
Unescape View File

@ -24,6 +24,8 @@ import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.List;
public class SwitchParameters extends AbstractParameters {
@ -88,8 +90,10 @@ public class SwitchParameters extends AbstractParameters {
}
//validate conditions must have next node
List<SwitchCondition> conditions = this.switchResult.getDependTaskList();
if (conditions != null && conditions.size() != 0) {
if (conditions.stream().anyMatch(e -> (StringUtils.isNotEmpty(e.getCondition()) && e.getNextNode() == null))) {
if (CollectionUtils.isNotEmpty(conditions)) {
if (conditions.stream()
.anyMatch(e -> (StringUtils.isNotEmpty(e.getCondition()) && e.getNextNode() == null))) {
return false;
}
}

14
dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/message/MessageRetryRunner.java
Unescape View File

@ -25,6 +25,18 @@ import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
import org.apache.dolphinscheduler.remote.command.BaseCommand;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.service.utils.LoggerUtils;
import org.apache.commons.collections.MapUtils;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.PostConstruct;
import lombok.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@ -99,7 +111,7 @@ public class MessageRetryRunner extends BaseDaemonThread {
public void run() {
while (!ServerLifeCycleManager.isStopped()) {
try {
if (needToRetryMessages.isEmpty()) {
if (MapUtils.isEmpty(needToRetryMessages)) {
Thread.sleep(MESSAGE_RETRY_WINDOW);
}

Write Preview
Loading…
Cancel
Save