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

Clean up the utility codes (#6732)

Browse Source
This commit is contained in:
kezhenxu94 2021-11-08 23:22:07 +08:00 committed by GitHub
parent 0f94577d26
commit 63602bf77c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
77 changed files with 638 additions and 2196 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/runner/AlertSender.java
View File

@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.alert.runner;
import org.apache.dolphinscheduler.alert.plugin.AlertPluginManager;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
@ -31,6 +30,8 @@ import org.apache.dolphinscheduler.spi.alert.AlertData;
import org.apache.dolphinscheduler.spi.alert.AlertInfo;
import org.apache.dolphinscheduler.spi.alert.AlertResult;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

3
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/AlertPluginInstanceServiceImpl.java
View File

@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.api.vo.AlertPluginInstanceVO;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
@ -33,6 +32,8 @@ import org.apache.dolphinscheduler.dao.mapper.AlertPluginInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import org.apache.dolphinscheduler.spi.params.PluginParamsTransfer;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;

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

@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.api.service.EnvironmentService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.SnowFlakeUtils;
import org.apache.dolphinscheduler.common.utils.SnowFlakeUtils.SnowFlakeException;
@ -35,6 +34,7 @@ import org.apache.dolphinscheduler.dao.mapper.EnvironmentMapper;
import org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections4.SetUtils;
import org.apache.commons.lang.StringUtils;

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

@ -40,7 +40,6 @@ import org.apache.dolphinscheduler.common.enums.RunMode;
import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.Command;
@ -58,6 +57,7 @@ import org.apache.dolphinscheduler.remote.processor.StateEventCallbackService;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;

5
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/LoggerServiceImpl.java
View File

@ -22,7 +22,6 @@ import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.LoggerService;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.ArrayUtils;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.service.log.LogClientService;
@ -41,6 +40,8 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.google.common.primitives.Bytes;
/**
* logger service impl
*/
@ -129,7 +130,7 @@ public class LoggerServiceImpl implements LoggerService {
taskInstance.getLogPath(),
host,
Constants.SYSTEM_LINE_SEPARATOR).getBytes(StandardCharsets.UTF_8);
return ArrayUtils.addAll(head,
return Bytes.concat(head,
logClient.getLogBytes(host, Constants.RPC_PORT, taskInstance.getLogPath()));
}

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

@ -41,7 +41,6 @@ import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.model.TaskNode;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.SnowFlakeUtils;
@ -73,6 +72,8 @@ import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.service.permission.PermissionCheck;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.commons.collections4.CollectionUtils;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;

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

@ -46,7 +46,6 @@ import org.apache.dolphinscheduler.common.graph.DAG;
import org.apache.dolphinscheduler.common.model.TaskNode;
import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
@ -68,6 +67,7 @@ import org.apache.dolphinscheduler.dao.mapper.TaskInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.io.BufferedReader;
@ -76,11 +76,13 @@ import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.springframework.beans.factory.annotation.Autowired;
@ -265,8 +267,15 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce
project.getCode(), processDefineCode, searchVal, executorId, statusArray, host, start, end);
List<ProcessInstance> processInstances = processInstanceList.getRecords();
List<Integer> userIds = CollectionUtils.transformToList(processInstances, ProcessInstance::getExecutorId);
Map<Integer, User> idToUserMap = CollectionUtils.collectionToMap(usersService.queryUser(userIds), User::getId);
List<Integer> userIds = Collections.emptyList();
if (CollectionUtils.isNotEmpty(processInstances)) {
userIds = processInstances.stream().map(ProcessInstance::getExecutorId).collect(Collectors.toList());
}
List<User> users = usersService.queryUser(userIds);
Map<Integer, User> idToUserMap = Collections.emptyMap();
if (CollectionUtils.isNotEmpty(users)) {
idToUserMap = users.stream().collect(Collectors.toMap(User::getId, Function.identity()));
}
for (ProcessInstance processInstance : processInstances) {
processInstance.setDuration(DateUtils.format2Duration(processInstance.getStartTime(), processInstance.getEndTime()));

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

@ -34,7 +34,6 @@ import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ProgramType;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
@ -53,6 +52,7 @@ import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.dao.utils.ResourceProcessDefinitionUtils;
import org.apache.commons.beanutils.BeanMap;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.io.IOException;
@ -80,6 +80,7 @@ import org.springframework.web.multipart.MultipartFile;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.google.common.io.Files;
/**
* resources service impl
@ -329,8 +330,8 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
if (!resource.isDirectory()) {
//get the origin file suffix
String originSuffix = FileUtils.suffix(originFullName);
String suffix = FileUtils.suffix(fullName);
String originSuffix = Files.getFileExtension(originFullName);
String suffix = Files.getFileExtension(fullName);
boolean suffixIsChanged = false;
if (StringUtils.isBlank(suffix) && StringUtils.isNotBlank(originSuffix)) {
suffixIsChanged = true;
@ -473,8 +474,8 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
}
// file suffix
String fileSuffix = FileUtils.suffix(file.getOriginalFilename());
String nameSuffix = FileUtils.suffix(name);
String fileSuffix = Files.getFileExtension(file.getOriginalFilename());
String nameSuffix = Files.getFileExtension(name);
// determine file suffix
if (!(StringUtils.isNotEmpty(fileSuffix) && fileSuffix.equalsIgnoreCase(nameSuffix))) {
@ -575,8 +576,8 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
*/
private boolean upload(User loginUser, String fullName, MultipartFile file, ResourceType type) {
// save to local
String fileSuffix = FileUtils.suffix(file.getOriginalFilename());
String nameSuffix = FileUtils.suffix(fullName);
String fileSuffix = Files.getFileExtension(file.getOriginalFilename());
String nameSuffix = Files.getFileExtension(fullName);
// determine file suffix
if (!(StringUtils.isNotEmpty(fileSuffix) && fileSuffix.equalsIgnoreCase(nameSuffix))) {
@ -598,11 +599,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
org.apache.dolphinscheduler.api.utils.FileUtils.copyFile(file, localFilename);
HadoopUtils.getInstance().copyLocalToHdfs(localFilename, hdfsFilename, true, true);
} catch (Exception e) {
try {
FileUtils.deleteFile(localFilename);
} catch (IOException ex) {
logger.error("delete local tmp file:{} error", localFilename, ex);
}
FileUtils.deleteFile(localFilename);
logger.error(e.getMessage(), e);
return false;
}
@ -839,7 +836,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
return result;
}
//check preview or not by file suffix
String nameSuffix = FileUtils.suffix(resource.getAlias());
String nameSuffix = Files.getFileExtension(resource.getAlias());
String resourceViewSuffixs = FileUtils.getResourceViewSuffixs();
if (StringUtils.isNotEmpty(resourceViewSuffixs)) {
List<String> strList = Arrays.asList(resourceViewSuffixs.split(","));
@ -1004,7 +1001,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
return result;
}
//check can edit by file suffix
String nameSuffix = FileUtils.suffix(resource.getAlias());
String nameSuffix = Files.getFileExtension(resource.getAlias());
String resourceViewSuffixs = FileUtils.getResourceViewSuffixs();
if (StringUtils.isNotEmpty(resourceViewSuffixs)) {
List<String> strList = Arrays.asList(resourceViewSuffixs.split(","));

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

@ -20,11 +20,11 @@ package org.apache.dolphinscheduler.api.service.impl;
import org.apache.dolphinscheduler.api.controller.BaseController;
import org.apache.dolphinscheduler.api.service.SessionService;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.Session;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.SessionMapper;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.Date;

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

@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.RegexUtils;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
@ -35,6 +34,7 @@ import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.Date;

3
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/UiPluginServiceImpl.java
View File

@ -21,10 +21,11 @@ import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.UiPluginService;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.PluginType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import org.apache.commons.collections.CollectionUtils;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

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

@ -29,7 +29,6 @@ import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.EncryptionUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
@ -55,6 +54,7 @@ import org.apache.dolphinscheduler.dao.mapper.UDFUserMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.dao.utils.ResourceProcessDefinitionUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.io.IOException;

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

@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.NodeType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.User;
@ -32,6 +31,7 @@ import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.WorkerGroupMapper;
import org.apache.dolphinscheduler.service.registry.RegistryClient;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;

44
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/FileUtils.java
View File

@ -16,22 +16,21 @@
*/
package org.apache.dolphinscheduler.api.utils;
import org.apache.commons.io.IOUtils;
import java.io.File;
import java.io.IOException;
import java.net.MalformedURLException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Path;
import java.nio.file.Paths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.web.multipart.MultipartFile;
import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.MalformedURLException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
/**
* file utils
*/
@ -44,18 +43,9 @@ public class FileUtils {
* @param file file
* @param destFilename destination file name
*/
public static void copyFile(MultipartFile file, String destFilename) {
try {
File destFile = new File(destFilename);
File destParentDir = new File(destFile.getParent());
if (!destParentDir.exists()) {
org.apache.commons.io.FileUtils.forceMkdir(destParentDir);
}
Files.copy(file.getInputStream(), Paths.get(destFilename));
org.apache.commons.io.FileUtils.copyFile(file.getResource().getFile(), new File(destFilename));
} catch (IOException e) {
logger.error("failed to copy file , {} is empty file", file.getOriginalFilename(), e);
}
@ -87,20 +77,12 @@ public class FileUtils {
* @return file content string
*/
public static String file2String(MultipartFile file) {
StringBuilder strBuilder = new StringBuilder();
try (InputStreamReader inputStreamReader = new InputStreamReader(file.getInputStream(), StandardCharsets.UTF_8)) {
BufferedReader streamReader = new BufferedReader(inputStreamReader);
String inputStr;
while ((inputStr = streamReader.readLine()) != null) {
strBuilder.append(inputStr);
}
try {
return IOUtils.toString(file.getInputStream(), StandardCharsets.UTF_8);
} catch (IOException e) {
logger.error("file convert to string failed: {}", file.getName());
}
return strBuilder.toString();
return "";
}
}

95
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/FourLetterWordMain.java
View File

@ -1,95 +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.dolphinscheduler.api.utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.Objects;
public class FourLetterWordMain {
private static final int DEFAULT_SOCKET_TIMEOUT = 5000;
protected static final Logger LOG = LoggerFactory.getLogger(FourLetterWordMain.class);
private FourLetterWordMain() {
throw new IllegalStateException("FourLetterWordMain class");
}
/**
* Send the 4letterword
* @param host the destination host
* @param port the destination port
* @param cmd the 4letterword
* @return server response
* @throws java.io.IOException io exceptions
*/
public static String send4LetterWord(String host, int port, String cmd)
throws IOException {
return send4LetterWord(host, port, cmd, DEFAULT_SOCKET_TIMEOUT);
}
/**
* Send the 4letterword
* @param host the destination host
* @param port the destination port
* @param cmd the 4letterword
* @param timeout in milliseconds, maximum time to wait while connecting/reading data
* @return server response
* @throws java.io.IOException io exceptions
*/
public static String send4LetterWord(String host, int port, String cmd, int timeout)
throws IOException {
Objects.requireNonNull(cmd, "cmd must not be null");
LOG.info("connecting to {} {}", host, port);
InetSocketAddress hostaddress= host != null ? new InetSocketAddress(host, port) :
new InetSocketAddress(InetAddress.getByName(null), port);
try (Socket sock = new Socket()) {
sock.setSoTimeout(timeout);
sock.connect(hostaddress, timeout);
OutputStream outstream = sock.getOutputStream();
outstream.write(cmd.getBytes());
outstream.flush();
// this replicates NC - close the output stream before reading
sock.shutdownOutput();
try (BufferedReader reader =
new BufferedReader(
new InputStreamReader(sock.getInputStream()))) {
StringBuilder sb = new StringBuilder();
String line;
while ((line = reader.readLine()) != null) {
sb.append(line + "\n");
}
return sb.toString();
}
} catch (SocketTimeoutException e) {
throw new IOException("Exception while executing four letter word: " + cmd, e);
}
}
}

18
dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/RegexUtils.java
View File

@ -17,7 +17,6 @@
package org.apache.dolphinscheduler.api.utils;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
/**
@ -25,28 +24,11 @@ import java.util.regex.Pattern;
*/
public class RegexUtils {
/**
* check number regex expression
*/
private static final String CHECK_NUMBER = "^-?\\d+(\\.\\d+)?$";
private static final String LINUX_USERNAME_PATTERN = "[a-z_][a-z\\d_]{0,30}";
private RegexUtils() {
}
/**
* check if the input is number
*
* @param str input
* @return
*/
public static boolean isNumeric(String str) {
Pattern pattern = Pattern.compile(CHECK_NUMBER);
Matcher isNum = pattern.matcher(str);
return isNum.matches();
}
/**
* check if the input is a valid linux username
* @param str input

3
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/AlertGroupServiceTest.java
View File

@ -26,12 +26,13 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.AlertGroupMapper;
import org.apache.dolphinscheduler.dao.vo.AlertGroupVo;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

3
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/EnvironmentServiceTest.java
View File

@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.Environment;
import org.apache.dolphinscheduler.dao.entity.EnvironmentWorkerGroupRelation;
import org.apache.dolphinscheduler.dao.entity.User;
@ -31,6 +30,8 @@ import org.apache.dolphinscheduler.dao.mapper.EnvironmentMapper;
import org.apache.dolphinscheduler.dao.mapper.EnvironmentWorkerGroupRelationMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

3
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/MonitorServiceTest.java
View File

@ -22,10 +22,11 @@ import org.apache.dolphinscheduler.api.service.impl.MonitorServiceImpl;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.DbType;
import org.apache.dolphinscheduler.common.model.Server;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.MonitorDBDao;
import org.apache.dolphinscheduler.dao.entity.MonitorRecord;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

3
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ProjectServiceTest.java
View File

@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.Project;
import org.apache.dolphinscheduler.dao.entity.ProjectUser;
@ -33,6 +32,8 @@ import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.ProjectUserMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;

3
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
View File

@ -23,12 +23,13 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.QueueMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

28
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java
View File

@ -22,9 +22,7 @@ import org.apache.dolphinscheduler.api.service.impl.ResourcesServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
@ -38,6 +36,9 @@ import org.apache.dolphinscheduler.dao.mapper.ResourceUserMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.commons.collections.CollectionUtils;
import java.io.IOException;
import java.util.ArrayList;
@ -62,13 +63,16 @@ import org.springframework.mock.web.MockMultipartFile;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.google.common.io.Files;
/**
* resources service test
*/
@RunWith(PowerMockRunner.class)
@PowerMockIgnore({"sun.security.*", "javax.net.*"})
@PrepareForTest({HadoopUtils.class, PropertyUtils.class, FileUtils.class, org.apache.dolphinscheduler.api.utils.FileUtils.class})
@PrepareForTest({HadoopUtils.class, PropertyUtils.class,
FileUtils.class, org.apache.dolphinscheduler.api.utils.FileUtils.class,
Files.class})
public class ResourcesServiceTest {
private static final Logger logger = LoggerFactory.getLogger(ResourcesServiceTest.class);
@ -99,9 +103,9 @@ public class ResourcesServiceTest {
@Before
public void setUp() {
PowerMockito.mockStatic(HadoopUtils.class);
PowerMockito.mockStatic(FileUtils.class);
PowerMockito.mockStatic(Files.class);
PowerMockito.mockStatic(org.apache.dolphinscheduler.api.utils.FileUtils.class);
try {
// new HadoopUtils
@ -132,15 +136,15 @@ public class ResourcesServiceTest {
//RESOURCE_SUFFIX_FORBID_CHANGE
mockMultipartFile = new MockMultipartFile("test.pdf", "test.pdf", "pdf", "test".getBytes());
PowerMockito.when(FileUtils.suffix("test.pdf")).thenReturn("pdf");
PowerMockito.when(FileUtils.suffix("ResourcesServiceTest.jar")).thenReturn("jar");
PowerMockito.when(Files.getFileExtension("test.pdf")).thenReturn("pdf");
PowerMockito.when(Files.getFileExtension("ResourcesServiceTest.jar")).thenReturn("jar");
result = resourcesService.createResource(user, "ResourcesServiceTest.jar", "ResourcesServiceTest", ResourceType.FILE, mockMultipartFile, -1, "/");
logger.info(result.toString());
Assert.assertEquals(Status.RESOURCE_SUFFIX_FORBID_CHANGE.getMsg(), result.getMsg());
//UDF_RESOURCE_SUFFIX_NOT_JAR
mockMultipartFile = new MockMultipartFile("ResourcesServiceTest.pdf", "ResourcesServiceTest.pdf", "pdf", "test".getBytes());
PowerMockito.when(FileUtils.suffix("ResourcesServiceTest.pdf")).thenReturn("pdf");
PowerMockito.when(Files.getFileExtension("ResourcesServiceTest.pdf")).thenReturn("pdf");
result = resourcesService.createResource(user, "ResourcesServiceTest.pdf", "ResourcesServiceTest", ResourceType.UDF, mockMultipartFile, -1, "/");
logger.info(result.toString());
Assert.assertEquals(Status.UDF_RESOURCE_SUFFIX_NOT_JAR.getMsg(), result.getMsg());
@ -268,10 +272,10 @@ public class ResourcesServiceTest {
resourcePage.setRecords(getResourceList());
Mockito.when(resourcesMapper.queryResourcePaging(Mockito.any(Page.class),
Mockito.eq(0), Mockito.eq(-1), Mockito.eq(0), Mockito.eq("test"), Mockito.any())).thenReturn(resourcePage);
Mockito.eq(0), Mockito.eq(-1), Mockito.eq(0), Mockito.eq("test"), Mockito.any())).thenReturn(resourcePage);
Result result = resourcesService.queryResourceListPaging(loginUser, -1, ResourceType.FILE, "test", 1, 10);
logger.info(result.toString());
Assert.assertEquals(Status.SUCCESS.getCode(), (int)result.getCode());
Assert.assertEquals(Status.SUCCESS.getCode(), (int) result.getCode());
PageInfo pageInfo = (PageInfo) result.getData();
Assert.assertTrue(CollectionUtils.isNotEmpty(pageInfo.getTotalList()));
@ -407,7 +411,7 @@ public class ResourcesServiceTest {
//USER_NOT_EXIST
PowerMockito.when(FileUtils.getResourceViewSuffixs()).thenReturn("jar");
PowerMockito.when(FileUtils.suffix("ResourcesServiceTest.jar")).thenReturn("jar");
PowerMockito.when(Files.getFileExtension("ResourcesServiceTest.jar")).thenReturn("jar");
result = resourcesService.readResource(1, 1, 10);
logger.info(result.toString());
Assert.assertTrue(Status.USER_NOT_EXIST.getCode() == result.getCode());
@ -508,7 +512,7 @@ public class ResourcesServiceTest {
//USER_NOT_EXIST
PowerMockito.when(FileUtils.getResourceViewSuffixs()).thenReturn("jar");
PowerMockito.when(FileUtils.suffix("ResourcesServiceTest.jar")).thenReturn("jar");
PowerMockito.when(Files.getFileExtension("ResourcesServiceTest.jar")).thenReturn("jar");
result = resourcesService.updateResourceContent(1, "content");
logger.info(result.toString());
Assert.assertTrue(Status.USER_NOT_EXIST.getCode() == result.getCode());
@ -717,4 +721,4 @@ public class ResourcesServiceTest {
resources.add(resource);
return resources;
}
}
}

3
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
View File

@ -23,7 +23,6 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.Tenant;
@ -33,6 +32,8 @@ import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

5
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/UdfFuncServiceTest.java
View File

@ -24,7 +24,6 @@ import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.UdfType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.Resource;
import org.apache.dolphinscheduler.dao.entity.UdfFunc;
@ -33,6 +32,8 @@ import org.apache.dolphinscheduler.dao.mapper.ResourceMapper;
import org.apache.dolphinscheduler.dao.mapper.UDFUserMapper;
import org.apache.dolphinscheduler.dao.mapper.UdfFuncMapper;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
@ -244,4 +245,4 @@ public class UdfFuncServiceTest {
udfFunc.setType(UdfType.HIVE);
return udfFunc;
}
}
}

7
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/UsersServiceTest.java
View File

@ -26,9 +26,7 @@ import org.apache.dolphinscheduler.api.service.impl.UsersServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.EncryptionUtils;
import org.apache.dolphinscheduler.dao.entity.AlertGroup;
import org.apache.dolphinscheduler.dao.entity.Project;
@ -45,6 +43,9 @@ import org.apache.dolphinscheduler.dao.mapper.ResourceUserMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UDFUserMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;
@ -680,4 +681,4 @@ public class UsersServiceTest {
return alertGroups;
}
}
}

7
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/FileUtilsTest.java
View File

@ -63,16 +63,17 @@ public class FileUtilsTest {
public void testCopyFile() throws IOException {
//Define dest file path
String src = rootPath + System.getProperty("file.separator") + "src.txt";
String destFilename = rootPath + System.getProperty("file.separator") + "data.txt";
logger.info("destFilename: "+destFilename);
//Define InputStream for MultipartFile
String data = "data text";
InputStream targetStream = new ByteArrayInputStream(data.getBytes());
org.apache.commons.io.FileUtils.writeStringToFile(new File(src), data);
//Use Mockito to mock MultipartFile
MultipartFile file = Mockito.mock(MultipartFile.class);
Mockito.when(file.getInputStream()).thenReturn(targetStream);
MultipartFile file = Mockito.mock(MultipartFile.class, Mockito.RETURNS_DEEP_STUBS);
Mockito.when(file.getResource().getFile()).thenReturn(new File(src));
//Invoke copyFile
FileUtils.copyFile(file,destFilename);

217
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/FourLetterWordMainTest.java
View File

@ -1,217 +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.dolphinscheduler.api.utils;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.when;
@RunWith(PowerMockRunner.class)
@PrepareForTest({FourLetterWordMain.class, Socket.class})
public class FourLetterWordMainTest {
private static final Logger logger =
LoggerFactory.getLogger(FourLetterWordMainTest.class);
private static final String NEW_LINE = "\n";
@InjectMocks
private FourLetterWordMain fourLetterWord;
@Mock
private Socket socket;
@Mock
private InetSocketAddress socketAddress;
private final String localHost = "127.0.0.1";
private final int zkPort = 2181;
private ByteArrayOutputStream byteArrayOutputStream;
private InputStream inputStream;
private String cmd;
private String testResult;
private String expectedStr;
@Before
public void setUp() {
// mock socket class
PowerMockito.mockStatic(Socket.class);
try {
PowerMockito.whenNew(Socket.class).withNoArguments()
.thenReturn(socket);
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* None mock test method, just to check zookeeper status.
* Comment @Before notation to run this test.
* Zookeeper status will be as:
* Zookeeper version: 3.4.11 ...
* Received: 6739707
* Sent: 6739773
* Connections: 20
* Outstanding: 0
* Zxid: 0x9ba
* Mode: standalone
* Node count: 263
*/
public void testCmd() {
// "192.168.64.11"
// final String zkHost = localHost;
final String zkHost = "192.168.64.11";
cmd = "srvr";
try {
// Change localhost to right zk host ip.
final String result = FourLetterWordMain
.send4LetterWord(zkHost, zkPort, cmd);
logger.info(cmd + ": " + result + "<<<");
} catch (Exception e) {
e.printStackTrace();
}
}
@Test
public void testEmptyCmd() {
cmd = "";
expectedStr = "";
testSend4LetterWord(cmd, expectedStr);
}
@Test
public void testNullCmd() {
cmd = null;
try {
testResult = FourLetterWordMain
.send4LetterWord(localHost, zkPort, cmd);
} catch (Exception e) {
testResult = e.getMessage();
}
logger.info("testNullCmd result: " + testResult);
assertEquals("cmd must not be null", testResult);
}
@Test
public void testNullSocketOutput() {
cmd = "test null socket output";
expectedStr = null;
testSend4LetterWord(cmd, expectedStr);
}
@Test
public void testOneLineOutput() {
cmd = "line 1";
// line end without \n
expectedStr = "line 1" + NEW_LINE;
testSend4LetterWord(cmd, expectedStr);
// line end with \n
expectedStr = "line 1\n" + NEW_LINE;
testSend4LetterWord(cmd, expectedStr);
}
@Test
public void testMultiline() {
cmd = "line 1 " + NEW_LINE +
"line 2 " + NEW_LINE +
"line 3 " + NEW_LINE;
expectedStr = cmd + NEW_LINE;
testSend4LetterWord(cmd, expectedStr);
expectedStr = NEW_LINE + NEW_LINE + NEW_LINE;
testSend4LetterWord(cmd, expectedStr);
}
@Test
public void testSocketTimeOut() {
cmd = "test socket time out";
try {
doThrow(new SocketTimeoutException())
.when(socket)
.connect(any(InetSocketAddress.class), Mockito.anyInt());
testResult = FourLetterWordMain
.send4LetterWord(localHost, zkPort, cmd);
} catch (Exception e) {
testResult = e.getMessage();
}
logger.info("testSocketTimeOut result: " + testResult);
assertEquals(
"Exception while executing four letter word: " + cmd,
testResult
);
}
/**
* Test FourLetterWordMain.send4LetterWord() with input cmd and output
* string.
* @param cmd
* @param expectedStr
*/
public void testSend4LetterWord(String cmd, String expectedStr) {
try {
final byte[] strBytes = cmd.getBytes();
byteArrayOutputStream = new ByteArrayOutputStream(strBytes.length);
byteArrayOutputStream.write(strBytes, 0, strBytes.length);
inputStream = new ByteArrayInputStream(expectedStr.getBytes());
when(socket.getOutputStream())
.thenReturn(byteArrayOutputStream);
when(socket.getInputStream()).thenReturn(inputStream);
final String result = FourLetterWordMain
.send4LetterWord(localHost, zkPort, cmd);
logger.info(
"testSend4LetterWord: " +
"cmd: " + cmd +
", expectedStr: " + expectedStr +
", result: " + result + "."
);
Assert.assertEquals(expectedStr, result);
} catch (Exception e) {
e.printStackTrace();
}
}
}

13
dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/RegexUtilsTest.java
View File

@ -25,17 +25,6 @@ import org.junit.Test;
*/
public class RegexUtilsTest {
@Test
public void testIsNumeric() {
String num1 = "123467854678";
boolean numeric = RegexUtils.isNumeric(num1);
Assert.assertTrue(numeric);
String num2 = "0.0.01";
boolean numeric2 = RegexUtils.isNumeric(num2);
Assert.assertFalse(numeric2);
}
@Test
public void testIsValidLinuxUserName() {
String name1 = "10000";
@ -72,4 +61,4 @@ public class RegexUtilsTest {
Assert.assertNull(result4);
}
}
}

6
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
View File

@ -18,9 +18,9 @@
package org.apache.dolphinscheduler.common;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import java.util.regex.Pattern;
@ -719,7 +719,7 @@ public final class Constants {
* application regex
*/
public static final String APPLICATION_REGEX = "application_\\d+_\\d+";
public static final String PID = OSUtils.isWindows() ? "handle" : "pid";
public static final String PID = SystemUtils.IS_OS_WINDOWS ? "handle" : "pid";
/**
* month_begin
*/
@ -751,7 +751,7 @@ public final class Constants {
public static final char LEFT_BRACE_CHAR = '(';
public static final char RIGHT_BRACE_CHAR = ')';
public static final String ADD_STRING = "+";
public static final String MULTIPLY_STRING = "*";
public static final String STAR = "*";
public static final String DIVISION_STRING = "/";
public static final String LEFT_BRACE_STRING = "(";
public static final char P = 'P';

830
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/graph/DAG.java
View File

@ -16,14 +16,22 @@
*/
package org.apache.dolphinscheduler.common.graph;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.*;
import java.util.AbstractMap;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* analysis of DAG
* Node: node
@ -31,487 +39,461 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
* EdgeInfo: edge description information
*/
public class DAG<Node, NodeInfo, EdgeInfo> {
private static final Logger logger = LoggerFactory.getLogger(DAG.class);
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private static final Logger logger = LoggerFactory.getLogger(DAG.class);
/**
* node map, key is node, value is node information
*/
private final Map<Node, NodeInfo> nodesMap;
private final ReadWriteLock lock = new ReentrantReadWriteLock();
/**
* edge map. key is node of origin;value is Map with key for destination node and value for edge
*/
private final Map<Node, Map<Node, EdgeInfo>> edgesMap;
/**
* node map, key is node, value is node information
*/
private volatile Map<Node, NodeInfo> nodesMap;
/**
* reversed edge set，key is node of destination, value is Map with key for origin node and value for edge
*/
private final Map<Node, Map<Node, EdgeInfo>> reverseEdgesMap;
/**
* edge map. key is node of origin;value is Map with key for destination node and value for edge
*/
private volatile Map<Node, Map<Node, EdgeInfo>> edgesMap;
/**
* reversed edge set，key is node of destination, value is Map with key for origin node and value for edge
*/
private volatile Map<Node, Map<Node, EdgeInfo>> reverseEdgesMap;
public DAG() {
nodesMap = new HashMap<>();
edgesMap = new HashMap<>();
reverseEdgesMap = new HashMap<>();
}
/**
* add node information
*
* @param node node
* @param nodeInfo node information
*/
public void addNode(Node node, NodeInfo nodeInfo) {
lock.writeLock().lock();
try{
nodesMap.put(node, nodeInfo);
}finally {
lock.writeLock().unlock();
public DAG() {
nodesMap = new HashMap<>();
edgesMap = new HashMap<>();
reverseEdgesMap = new HashMap<>();
}
}
/**
* add node information
*
* @param node node
* @param nodeInfo node information
*/
public void addNode(Node node, NodeInfo nodeInfo) {
lock.writeLock().lock();
try {
nodesMap.put(node, nodeInfo);
} finally {
lock.writeLock().unlock();
}
/**
* add edge
* @param fromNode node of origin
* @param toNode node of destination
* @return The result of adding an edge. returns false if the DAG result is a ring result
*/
public boolean addEdge(Node fromNode, Node toNode) {
return addEdge(fromNode, toNode, false);
}
/**
* add edge
* @param fromNode node of origin
* @param toNode node of destination
* @param createNode whether the node needs to be created if it does not exist
* @return The result of adding an edge. returns false if the DAG result is a ring result
*/
private boolean addEdge(Node fromNode, Node toNode, boolean createNode) {
return addEdge(fromNode, toNode, null, createNode);
}
/**
* add edge
*
* @param fromNode node of origin
* @param toNode node of destination
* @param edge edge description
* @param createNode whether the node needs to be created if it does not exist
* @return The result of adding an edge. returns false if the DAG result is a ring result
*/
public boolean addEdge(Node fromNode, Node toNode, EdgeInfo edge, boolean createNode) {
lock.writeLock().lock();
try{
// Whether an edge can be successfully added(fromNode -> toNode)
if (!isLegalAddEdge(fromNode, toNode, createNode)) {
logger.error("serious error: add edge({} -> {}) is invalid, cause cycle！", fromNode, toNode);
return false;
}
addNodeIfAbsent(fromNode, null);
addNodeIfAbsent(toNode, null);
addEdge(fromNode, toNode, edge, edgesMap);
addEdge(toNode, fromNode, edge, reverseEdgesMap);
return true;
}finally {
lock.writeLock().unlock();
}
}
/**
* whether this node is contained
*
* @param node node
* @return true if contains
*/
public boolean containsNode(Node node) {
lock.readLock().lock();
try{
return nodesMap.containsKey(node);
}finally {
lock.readLock().unlock();
}
}
/**
* whether this edge is contained
*
* @param fromNode node of origin
* @param toNode node of destination
* @return true if contains
*/
public boolean containsEdge(Node fromNode, Node toNode) {
lock.readLock().lock();
try{
Map<Node, EdgeInfo> endEdges = edgesMap.get(fromNode);
if (endEdges == null) {
return false;
}
return endEdges.containsKey(toNode);
}finally {
lock.readLock().unlock();
}
}
/**
* get node description
*
* @param node node
* @return node description
*/
public NodeInfo getNode(Node node) {
lock.readLock().lock();
try{
return nodesMap.get(node);
}finally {
lock.readLock().unlock();
}
}
/**
* Get the number of nodes
*
* @return the number of nodes
*/
public int getNodesCount() {
lock.readLock().lock();
try{
return nodesMap.size();
}finally {
lock.readLock().unlock();
}
}
/**
* Get the number of edges
*
* @return the number of edges
*/
public int getEdgesCount() {
lock.readLock().lock();
try{
int count = 0;
for (Map.Entry<Node, Map<Node, EdgeInfo>> entry : edgesMap.entrySet()) {
count += entry.getValue().size();
}
return count;
}finally {
lock.readLock().unlock();
}
}
/**
* get the start node of DAG
*
* @return the start node of DAG
*/
public Collection<Node> getBeginNode() {
lock.readLock().lock();
try{
return CollectionUtils.subtract(nodesMap.keySet(), reverseEdgesMap.keySet());
}finally {
lock.readLock().unlock();
/**
* add edge
*
* @param fromNode node of origin
* @param toNode node of destination
* @return The result of adding an edge. returns false if the DAG result is a ring result
*/
public boolean addEdge(Node fromNode, Node toNode) {
return addEdge(fromNode, toNode, false);
}
}
/**
* get the end node of DAG
*
* @return the end node of DAG
*/
public Collection<Node> getEndNode() {
lock.readLock().lock();
try{
return CollectionUtils.subtract(nodesMap.keySet(), edgesMap.keySet());
}finally {
lock.readLock().unlock();
/**
* add edge
*
* @param fromNode node of origin
* @param toNode node of destination
* @param createNode whether the node needs to be created if it does not exist
* @return The result of adding an edge. returns false if the DAG result is a ring result
*/
private boolean addEdge(Node fromNode, Node toNode, boolean createNode) {
return addEdge(fromNode, toNode, null, createNode);
}
}
/**
* add edge
*
* @param fromNode node of origin
* @param toNode node of destination
* @param edge edge description
* @param createNode whether the node needs to be created if it does not exist
* @return The result of adding an edge. returns false if the DAG result is a ring result
*/
public boolean addEdge(Node fromNode, Node toNode, EdgeInfo edge, boolean createNode) {
lock.writeLock().lock();
try {
// Whether an edge can be successfully added(fromNode -> toNode)
if (!isLegalAddEdge(fromNode, toNode, createNode)) {
logger.error("serious error: add edge({} -> {}) is invalid, cause cycle！", fromNode, toNode);
return false;
}
/**
* Gets all previous nodes of the node
*
* @param node node id to be calculated
* @return all previous nodes of the node
*/
public Set<Node> getPreviousNodes(Node node) {
lock.readLock().lock();
addNodeIfAbsent(fromNode, null);
addNodeIfAbsent(toNode, null);
addEdge(fromNode, toNode, edge, edgesMap);
addEdge(toNode, fromNode, edge, reverseEdgesMap);
return true;
} finally {
lock.writeLock().unlock();
}
try{
return getNeighborNodes(node, reverseEdgesMap);
}finally {
lock.readLock().unlock();
}
}
/**
* whether this node is contained
*
* @param node node
* @return true if contains
*/
public boolean containsNode(Node node) {
lock.readLock().lock();
/**
* Get all subsequent nodes of the node
*
* @param node node id to be calculated
* @return all subsequent nodes of the node
*/
public Set<Node> getSubsequentNodes(Node node) {
lock.readLock().lock();
try{
return getNeighborNodes(node, edgesMap);
}finally {
lock.readLock().unlock();
try {
return nodesMap.containsKey(node);
} finally {
lock.readLock().unlock();
}
}
}
/**
* whether this edge is contained
*
* @param fromNode node of origin
* @param toNode node of destination
* @return true if contains
*/
public boolean containsEdge(Node fromNode, Node toNode) {
lock.readLock().lock();
try {
Map<Node, EdgeInfo> endEdges = edgesMap.get(fromNode);
if (endEdges == null) {
return false;
}
/**
* Gets the degree of entry of the node
*
* @param node node id
* @return the degree of entry of the node
*/
public int getIndegree(Node node) {
lock.readLock().lock();
try{
return getPreviousNodes(node).size();
}finally {
lock.readLock().unlock();
return endEdges.containsKey(toNode);
} finally {
lock.readLock().unlock();
}
}
}
/**
* get node description
*
* @param node node
* @return node description
*/
public NodeInfo getNode(Node node) {
lock.readLock().lock();
/**
* whether the graph has a ring
*
* @return true if has cycle, else return false.
*/
public boolean hasCycle() {
lock.readLock().lock();
try{
return !topologicalSortImpl().getKey();
}finally {
lock.readLock().unlock();
try {
return nodesMap.get(node);
} finally {
lock.readLock().unlock();
}
}
}
/**
* Get the number of nodes
*
* @return the number of nodes
*/
public int getNodesCount() {
lock.readLock().lock();
/**
* Only DAG has a topological sort
* @return topologically sorted results, returns false if the DAG result is a ring result
* @throws Exception errors
*/
public List<Node> topologicalSort() throws Exception {
lock.readLock().lock();
try{
Map.Entry<Boolean, List<Node>> entry = topologicalSortImpl();
if (entry.getKey()) {
return entry.getValue();
}
throw new Exception("serious error: graph has cycle ! ");
}finally {
lock.readLock().unlock();
try {
return nodesMap.size();
} finally {
lock.readLock().unlock();
}
}
}
/**
* Get the number of edges
*
* @return the number of edges
*/
public int getEdgesCount() {
lock.readLock().lock();
try {
int count = 0;
/**
* if tho node does not exist,add this node
*
* @param node node
* @param nodeInfo node information
*/
private void addNodeIfAbsent(Node node, NodeInfo nodeInfo) {
if (!containsNode(node)) {
addNode(node, nodeInfo);
for (Map.Entry<Node, Map<Node, EdgeInfo>> entry : edgesMap.entrySet()) {
count += entry.getValue().size();
}
return count;
} finally {
lock.readLock().unlock();
}
}
}
/**
* get the start node of DAG
*
* @return the start node of DAG
*/
public Collection<Node> getBeginNode() {
lock.readLock().lock();
/**
* add edge
*
* @param fromNode node of origin
* @param toNode node of destination
* @param edge edge description
* @param edges edge set
*/
private void addEdge(Node fromNode, Node toNode, EdgeInfo edge, Map<Node, Map<Node, EdgeInfo>> edges) {
edges.putIfAbsent(fromNode, new HashMap<>());
Map<Node, EdgeInfo> toNodeEdges = edges.get(fromNode);
toNodeEdges.put(toNode, edge);
}
try {
return org.apache.commons.collections4.CollectionUtils.subtract(nodesMap.keySet(), reverseEdgesMap.keySet());
} finally {
lock.readLock().unlock();
}
}
/**
* Whether an edge can be successfully added(fromNode -> toNode)
* need to determine whether the DAG has cycle
*
* @param fromNode node of origin
* @param toNode node of destination
* @param createNode whether to create a node
* @return true if added
*/
private boolean isLegalAddEdge(Node fromNode, Node toNode, boolean createNode) {
if (fromNode.equals(toNode)) {
logger.error("edge fromNode({}) can't equals toNode({})", fromNode, toNode);
return false;
}
/**
* get the end node of DAG
*
* @return the end node of DAG
*/
public Collection<Node> getEndNode() {
lock.readLock().lock();
if (!createNode) {
if (!containsNode(fromNode) || !containsNode(toNode)){
logger.error("edge fromNode({}) or toNode({}) is not in vertices map", fromNode, toNode);
return false;
}
}
try {
return org.apache.commons.collections4.CollectionUtils.subtract(nodesMap.keySet(), edgesMap.keySet());
} finally {
lock.readLock().unlock();
}
// Whether an edge can be successfully added(fromNode -> toNode),need to determine whether the DAG has cycle!
int verticesCount = getNodesCount();
}
Queue<Node> queue = new LinkedList<>();
/**
* Gets all previous nodes of the node
*
* @param node node id to be calculated
* @return all previous nodes of the node
*/
public Set<Node> getPreviousNodes(Node node) {
lock.readLock().lock();
queue.add(toNode);
try {
return getNeighborNodes(node, reverseEdgesMap);
} finally {
lock.readLock().unlock();
}
}
// if DAG doesn't find fromNode, it's not has cycle!
while (!queue.isEmpty() && (--verticesCount > 0)) {
Node key = queue.poll();
/**
* Get all subsequent nodes of the node
*
* @param node node id to be calculated
* @return all subsequent nodes of the node
*/
public Set<Node> getSubsequentNodes(Node node) {
lock.readLock().lock();
for (Node subsequentNode : getSubsequentNodes(key)) {
if (subsequentNode.equals(fromNode)) {
return false;
}
try {
return getNeighborNodes(node, edgesMap);
} finally {
lock.readLock().unlock();
}
}
queue.add(subsequentNode);
}
}
/**
* Gets the degree of entry of the node
*
* @param node node id
* @return the degree of entry of the node
*/
public int getIndegree(Node node) {
lock.readLock().lock();
return true;
}
try {
return getPreviousNodes(node).size();
} finally {
lock.readLock().unlock();
}
}
/**
* whether the graph has a ring
*
* @return true if has cycle, else return false.
*/
public boolean hasCycle() {
lock.readLock().lock();
try {
return !topologicalSortImpl().getKey();
} finally {
lock.readLock().unlock();
}
}
/**
* Get all neighbor nodes of the node
*
* @param node Node id to be calculated
* @param edges neighbor edge information
* @return all neighbor nodes of the node
*/
private Set<Node> getNeighborNodes(Node node, final Map<Node, Map<Node, EdgeInfo>> edges) {
/**
* Only DAG has a topological sort
*
* @return topologically sorted results, returns false if the DAG result is a ring result
* @throws Exception errors
*/
public List<Node> topologicalSort() throws Exception {
lock.readLock().lock();
try {
Map.Entry<Boolean, List<Node>> entry = topologicalSortImpl();
if (entry.getKey()) {
return entry.getValue();
}
throw new Exception("serious error: graph has cycle ! ");
} finally {
lock.readLock().unlock();
}
}
/**
* if tho node does not exist,add this node
*
* @param node node
* @param nodeInfo node information
*/
private void addNodeIfAbsent(Node node, NodeInfo nodeInfo) {
if (!containsNode(node)) {
addNode(node, nodeInfo);
}
}
/**
* add edge
*
* @param fromNode node of origin
* @param toNode node of destination
* @param edge edge description
* @param edges edge set
*/
private void addEdge(Node fromNode, Node toNode, EdgeInfo edge, Map<Node, Map<Node, EdgeInfo>> edges) {
edges.putIfAbsent(fromNode, new HashMap<>());
Map<Node, EdgeInfo> toNodeEdges = edges.get(fromNode);
toNodeEdges.put(toNode, edge);
}
/**
* Whether an edge can be successfully added(fromNode -> toNode)
* need to determine whether the DAG has cycle
*
* @param fromNode node of origin
* @param toNode node of destination
* @param createNode whether to create a node
* @return true if added
*/
private boolean isLegalAddEdge(Node fromNode, Node toNode, boolean createNode) {
if (fromNode.equals(toNode)) {
logger.error("edge fromNode({}) can't equals toNode({})", fromNode, toNode);
return false;
}
if (!createNode) {
if (!containsNode(fromNode) || !containsNode(toNode)) {
logger.error("edge fromNode({}) or toNode({}) is not in vertices map", fromNode, toNode);
return false;
}
}
// Whether an edge can be successfully added(fromNode -> toNode),need to determine whether the DAG has cycle!
int verticesCount = getNodesCount();
Queue<Node> queue = new LinkedList<>();
queue.add(toNode);
// if DAG doesn't find fromNode, it's not has cycle!
while (!queue.isEmpty() && (--verticesCount > 0)) {
Node key = queue.poll();
for (Node subsequentNode : getSubsequentNodes(key)) {
if (subsequentNode.equals(fromNode)) {
return false;
}
queue.add(subsequentNode);
}
}
return true;
}
/**
* Get all neighbor nodes of the node
*
* @param node Node id to be calculated
* @param edges neighbor edge information
* @return all neighbor nodes of the node
*/
private Set<Node> getNeighborNodes(Node node, final Map<Node, Map<Node, EdgeInfo>> edges) {
final Map<Node, EdgeInfo> neighborEdges = edges.get(node);
if (neighborEdges == null) {
return Collections.emptySet();
}
return neighborEdges.keySet();
}
/**
* Determine whether there are ring and topological sorting results
*
* Directed acyclic graph (DAG) has topological ordering
* Breadth First Search：
* 1、Traversal of all the vertices in the graph, the degree of entry is 0 vertex into the queue
* 2、Poll a vertex in the queue to update its adjacency (minus 1) and queue the adjacency if it is 0 after minus 1
* 3、Do step 2 until the queue is empty
* If you cannot traverse all the nodes, it means that the current graph is not a directed acyclic graph.
* There is no topological sort.
*
*
* @return key Returns the state
* if success (acyclic) is true, failure (acyclic) is looped,
* and value (possibly one of the topological sort results)
*/
private Map.Entry<Boolean, List<Node>> topologicalSortImpl() {
// node queue with degree of entry 0
Queue<Node> zeroIndegreeNodeQueue = new LinkedList<>();
// save result
List<Node> topoResultList = new ArrayList<>();
// save the node whose degree is not 0
Map<Node, Integer> notZeroIndegreeNodeMap = new HashMap<>();
// Scan all the vertices and push vertexs with an entry degree of 0 to queue
for (Map.Entry<Node, NodeInfo> vertices : nodesMap.entrySet()) {
Node node = vertices.getKey();
int inDegree = getIndegree(node);
if (inDegree == 0) {
zeroIndegreeNodeQueue.add(node);
topoResultList.add(node);
} else {
notZeroIndegreeNodeMap.put(node, inDegree);
}
}
/**
* After scanning, there is no node with 0 degree of entry,
* indicating that there is a ring, and return directly
* Determine whether there are ring and topological sorting results
* <p>
* Directed acyclic graph (DAG) has topological ordering
* Breadth First Search：
* 1、Traversal of all the vertices in the graph, the degree of entry is 0 vertex into the queue
* 2、Poll a vertex in the queue to update its adjacency (minus 1) and queue the adjacency if it is 0 after minus 1
* 3、Do step 2 until the queue is empty
* If you cannot traverse all the nodes, it means that the current graph is not a directed acyclic graph.
* There is no topological sort.
*
* @return key Returns the state
* if success (acyclic) is true, failure (acyclic) is looped,
* and value (possibly one of the topological sort results)
*/
if(zeroIndegreeNodeQueue.isEmpty()){
return new AbstractMap.SimpleEntry(false, topoResultList);
}
private Map.Entry<Boolean, List<Node>> topologicalSortImpl() {
// node queue with degree of entry 0
Queue<Node> zeroIndegreeNodeQueue = new LinkedList<>();
// save result
List<Node> topoResultList = new ArrayList<>();
// save the node whose degree is not 0
Map<Node, Integer> notZeroIndegreeNodeMap = new HashMap<>();
// The topology algorithm is used to delete nodes with 0 degree of entry and its associated edges
while (!zeroIndegreeNodeQueue.isEmpty()) {
Node v = zeroIndegreeNodeQueue.poll();
// Get the neighbor node
Set<Node> subsequentNodes = getSubsequentNodes(v);
// Scan all the vertices and push vertexs with an entry degree of 0 to queue
for (Map.Entry<Node, NodeInfo> vertices : nodesMap.entrySet()) {
Node node = vertices.getKey();
int inDegree = getIndegree(node);
for (Node subsequentNode : subsequentNodes) {
Integer degree = notZeroIndegreeNodeMap.get(subsequentNode);
if(--degree == 0){
topoResultList.add(subsequentNode);
zeroIndegreeNodeQueue.add(subsequentNode);
notZeroIndegreeNodeMap.remove(subsequentNode);
}else{
notZeroIndegreeNodeMap.put(subsequentNode, degree);
if (inDegree == 0) {
zeroIndegreeNodeQueue.add(node);
topoResultList.add(node);
} else {
notZeroIndegreeNodeMap.put(node, inDegree);
}
}
}
/*
* After scanning, there is no node with 0 degree of entry,
* indicating that there is a ring, and return directly
*/
if (zeroIndegreeNodeQueue.isEmpty()) {
return new AbstractMap.SimpleEntry<>(false, topoResultList);
}
// The topology algorithm is used to delete nodes with 0 degree of entry and its associated edges
while (!zeroIndegreeNodeQueue.isEmpty()) {
Node v = zeroIndegreeNodeQueue.poll();
// Get the neighbor node
Set<Node> subsequentNodes = getSubsequentNodes(v);
for (Node subsequentNode : subsequentNodes) {
Integer degree = notZeroIndegreeNodeMap.get(subsequentNode);
if (--degree == 0) {
topoResultList.add(subsequentNode);
zeroIndegreeNodeQueue.add(subsequentNode);
notZeroIndegreeNodeMap.remove(subsequentNode);
} else {
notZeroIndegreeNodeMap.put(subsequentNode, degree);
}
}
}
// if notZeroIndegreeNodeMap is empty,there is no ring!
return new AbstractMap.SimpleEntry<>(notZeroIndegreeNodeMap.size() == 0, topoResultList);
}
// if notZeroIndegreeNodeMap is empty,there is no ring!
AbstractMap.SimpleEntry resultMap = new AbstractMap.SimpleEntry(notZeroIndegreeNodeMap.size() == 0 , topoResultList);
return resultMap;
}
}

2
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/AbstractParameters.java
View File

@ -19,9 +19,9 @@ package org.apache.dolphinscheduler.common.task;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;

2
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/sql/SqlParameters.java
View File

@ -21,9 +21,9 @@ import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.AbstractParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;

38
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/ArrayUtils.java
View File

@ -1,38 +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.dolphinscheduler.common.utils;
public class ArrayUtils {
public static byte[] clone(byte[] array) {
return array == null ? null : (byte[])((byte[])array.clone());
}
public static byte[] addAll(byte[] array1, byte[] array2) {
if (array1 == null) {
return clone(array2);
} else if (array2 == null) {
return clone(array1);
} else {
byte[] joinedArray = new byte[array1.length + array2.length];
System.arraycopy(array1, 0, joinedArray, 0, array1.length);
System.arraycopy(array2, 0, joinedArray, array1.length, array2.length);
return joinedArray;
}
}
}

223
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/CollectionUtils.java
View File

@ -18,18 +18,14 @@
package org.apache.dolphinscheduler.common.utils;
import org.apache.commons.beanutils.BeanMap;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* Provides utility methods and decorators for {@link Collection} instances.
@ -49,175 +45,6 @@ public class CollectionUtils {
throw new UnsupportedOperationException("Construct CollectionUtils");
}
/**
* The load factor used when none specified in constructor.
*/
static final float DEFAULT_LOAD_FACTOR = 0.75f;
/**
* Returns a new {@link Collection} containing <i>a</i> minus a subset of
* <i>b</i>. Only the elements of <i>b</i> that satisfy the predicate
* condition, <i>p</i> are subtracted from <i>a</i>.
*
* <p>The cardinality of each element <i>e</i> in the returned {@link Collection}
* that satisfies the predicate condition will be the cardinality of <i>e</i> in <i>a</i>
* minus the cardinality of <i>e</i> in <i>b</i>, or zero, whichever is greater.</p>
* <p>The cardinality of each element <i>e</i> in the returned {@link Collection} that does <b>not</b>
* satisfy the predicate condition will be equal to the cardinality of <i>e</i> in <i>a</i>.</p>
*
* @param a the collection to subtract from, must not be null
* @param b the collection to subtract, must not be null
* @param <T> T
* @return a new collection with the results
* @see Collection#removeAll
*/
public static <T> Collection<T> subtract(Set<T> a, Set<T> b) {
return org.apache.commons.collections4.CollectionUtils.subtract(a, b);
}
public static boolean isNotEmpty(Collection coll) {
return !isEmpty(coll);
}
public static boolean isEmpty(Collection coll) {
return coll == null || coll.isEmpty();
}
/**
* String to map
*
* @param str string
* @param separator separator
* @return string to map
*/
public static Map<String, String> stringToMap(String str, String separator) {
return stringToMap(str, separator, "");
}
/**
* String to map
*
* @param str string
* @param separator separator
* @param keyPrefix prefix
* @return string to map
*/
public static Map<String, String> stringToMap(String str, String separator, String keyPrefix) {
Map<String, String> emptyMap = new HashMap<>(0);
if (StringUtils.isEmpty(str)) {
return emptyMap;
}
if (StringUtils.isEmpty(separator)) {
return emptyMap;
}
String[] strings = str.split(separator);
int initialCapacity = (int)(strings.length / DEFAULT_LOAD_FACTOR) + 1;
Map<String, String> map = new HashMap<>(initialCapacity);
for (int i = 0; i < strings.length; i++) {
String[] strArray = strings[i].split("=");
if (strArray.length != 2) {
return emptyMap;
}
//strArray[0] KEY strArray[1] VALUE
if (StringUtils.isEmpty(keyPrefix)) {
map.put(strArray[0], strArray[1]);
} else {
map.put(keyPrefix + strArray[0], strArray[1]);
}
}
return map;
}
/**
* Transform item in collection
*
* @param collection origin collection
* @param transformFunc transform function
* @param <R> origin item type
* @param <T> target type
* @return transform list
*/
public static <R, T> List<T> transformToList(Collection<R> collection, Function<R, T> transformFunc) {
if (isEmpty(collection)) {
return new ArrayList<>();
}
return collection.stream().map(transformFunc).collect(Collectors.toList());
}
/**
* Collect collection to map
*
* @param collection origin collection
* @param keyTransformFunction key transform function
* @param <K> target k type
* @param <V> value
* @return map
*/
public static <K, V> Map<K, V> collectionToMap(Collection<V> collection, Function<V, K> keyTransformFunction) {
if (isEmpty(collection)) {
return new HashMap<>();
}
return collection.stream().collect(Collectors.toMap(keyTransformFunction, Function.identity()));
}
/**
* Helper class to easily access cardinality properties of two collections.
*
* @param <O> the element type
*/
private static class CardinalityHelper<O> {
/**
* Contains the cardinality for each object in collection A.
*/
final Map<O, Integer> cardinalityA;
/**
* Contains the cardinality for each object in collection B.
*/
final Map<O, Integer> cardinalityB;
/**
* Create a new CardinalityHelper for two collections.
*
* @param a the first collection
* @param b the second collection
*/
public CardinalityHelper(final Iterable<? extends O> a, final Iterable<? extends O> b) {
cardinalityA = CollectionUtils.getCardinalityMap(a);
cardinalityB = CollectionUtils.getCardinalityMap(b);
}
/**
* Returns the frequency of this object in collection A.
*
* @param obj the object
* @return the frequency of the object in collection A
*/
public int freqA(final Object obj) {
return getFreq(obj, cardinalityA);
}
/**
* Returns the frequency of this object in collection B.
*
* @param obj the object
* @return the frequency of the object in collection B
*/
public int freqB(final Object obj) {
return getFreq(obj, cardinalityB);
}
private int getFreq(final Object obj, final Map<?, Integer> freqMap) {
final Integer count = freqMap.get(obj);
if (count != null) {
return count;
}
return 0;
}
}
/**
* returns {@code true} iff the given {@link Collection}s contain
* exactly the same elements with exactly the same cardinalities.
@ -236,55 +63,7 @@ public class CollectionUtils {
return false;
}
return isEqualCollection(a, b);
}
/**
* Returns {@code true} iff the given {@link Collection}s contain
* exactly the same elements with exactly the same cardinalities.
* <p>
* That is, iff the cardinality of <i>e</i> in <i>a</i> is
* equal to the cardinality of <i>e</i> in <i>b</i>,
* for each element <i>e</i> in <i>a</i> or <i>b</i>.
*
* @param a the first collection, must not be null
* @param b the second collection, must not be null
* @return <code>true</code> iff the collections contain the same elements with the same cardinalities.
*/
public static boolean isEqualCollection(final Collection<?> a, final Collection<?> b) {
if (a.size() != b.size()) {
return false;
}
final CardinalityHelper<Object> helper = new CardinalityHelper<>(a, b);
if (helper.cardinalityA.size() != helper.cardinalityB.size()) {
return false;
}
for (final Object obj : helper.cardinalityA.keySet()) {
if (helper.freqA(obj) != helper.freqB(obj)) {
return false;
}
}
return true;
}
/**
* Returns a {@link Map} mapping each unique element in the given
* {@link Collection} to an {@link Integer} representing the number
* of occurrences of that element in the {@link Collection}.
* <p>
* Only those elements present in the collection will appear as
* keys in the map.
*
* @param <O> the type of object in the returned {@link Map}. This is a super type of O
* @param coll the collection to get the cardinality map for, must not be null
* @return the populated cardinality map
*/
public static <O> Map<O, Integer> getCardinalityMap(final Iterable<? extends O> coll) {
final Map<O, Integer> count = new HashMap<>();
for (final O obj : coll) {
count.put(obj, count.getOrDefault(obj, 0) + 1);
}
return count;
return org.apache.commons.collections.CollectionUtils.isEqualCollection(a, b);
}
/**

48
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/EnumUtils.java
View File

@ -1,48 +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.dolphinscheduler.common.utils;
public class EnumUtils {
private EnumUtils() {
throw new UnsupportedOperationException("Construct EnumUtils");
}
public static <E extends Enum<E>> E getEnum(final Class<E> enumClass, final String enumName) {
if (enumName == null) {
return null;
}
try {
return Enum.valueOf(enumClass, enumName);
} catch (final IllegalArgumentException ex) {
return null;
}
}
public static <E extends Enum<E>> boolean isValidEnum(final Class<E> enumClass, final String enumName) {
if (enumName == null) {
return false;
}
try {
Enum.valueOf(enumClass, enumName);
return true;
} catch (final IllegalArgumentException ex) {
return false;
}
}
}

57
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java
View File

@ -24,19 +24,13 @@ import static org.apache.dolphinscheduler.common.Constants.UTF_8;
import static org.apache.dolphinscheduler.common.Constants.YYYYMMDDHHMMSS;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.io.StringReader;
import java.nio.charset.StandardCharsets;
import java.util.Optional;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -50,30 +44,10 @@ public class FileUtils {
public static final String DATA_BASEDIR = PropertyUtils.getString(DATA_BASEDIR_PATH, "/tmp/dolphinscheduler");
public static final ThreadLocal<Logger> taskLoggerThreadLocal = new ThreadLocal<>();
private FileUtils() {
throw new UnsupportedOperationException("Construct FileUtils");
}
/**
* get file suffix
*
* @param filename file name
* @return file suffix
*/
public static String suffix(String filename) {
String fileSuffix = "";
if (!StringUtils.isEmpty(filename)) {
int lastIndex = filename.lastIndexOf('.');
if (lastIndex > 0) {
fileSuffix = filename.substring(lastIndex + 1);
}
}
return fileSuffix;
}
/**
* get download file absolute path and name
*
@ -153,8 +127,7 @@ public class FileUtils {
//create work dir
org.apache.commons.io.FileUtils.forceMkdir(execLocalPathFile);
String mkdirLog = "create dir success " + execLocalPath;
LoggerUtils.logInfo(Optional.ofNullable(logger), mkdirLog);
LoggerUtils.logInfo(Optional.ofNullable(taskLoggerThreadLocal.get()), mkdirLog);
logger.info(mkdirLog);
}
/**
@ -165,30 +138,16 @@ public class FileUtils {
* @return true if write success
*/
public static boolean writeContent2File(String content, String filePath) {
BufferedReader bufferedReader = null;
BufferedWriter bufferedWriter = null;
try {
File distFile = new File(filePath);
if (!distFile.getParentFile().exists() && !distFile.getParentFile().mkdirs()) {
FileUtils.logger.error("mkdir parent failed");
logger.error("mkdir parent failed");
return false;
}
bufferedReader = new BufferedReader(new StringReader(content));
bufferedWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(distFile), StandardCharsets.UTF_8));
char[] buf = new char[1024];
int len;
while ((len = bufferedReader.read(buf)) != -1) {
bufferedWriter.write(buf, 0, len);
}
bufferedWriter.flush();
bufferedReader.close();
bufferedWriter.close();
IOUtils.write(content, new FileOutputStream(filePath), StandardCharsets.UTF_8);
} catch (IOException e) {
FileUtils.logger.error(e.getMessage(), e);
logger.error(e.getMessage(), e);
return false;
} finally {
IOUtils.closeQuietly(bufferedWriter);
IOUtils.closeQuietly(bufferedReader);
}
return true;
}
@ -204,13 +163,9 @@ public class FileUtils {
* </ul>
*
* @param filename file name
* @throws IOException in case deletion is unsuccessful
*/
public static void deleteFile(String filename) throws IOException {
File file = new File(filename);
if (file.exists()) {
org.apache.commons.io.FileUtils.forceDelete(file);
}
public static void deleteFile(String filename) {
org.apache.commons.io.FileUtils.deleteQuietly(new File(filename));
}
/**

24
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/LoggerUtils.java
View File

@ -25,7 +25,6 @@ import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@ -124,25 +123,4 @@ public class LoggerUtils {
}
return "";
}
public static void logError(Optional<Logger> optionalLogger
, String error) {
optionalLogger.ifPresent((Logger logger) -> logger.error(error));
}
public static void logError(Optional<Logger> optionalLogger
, Throwable e) {
optionalLogger.ifPresent((Logger logger) -> logger.error(e.getMessage(), e));
}
public static void logError(Optional<Logger> optionalLogger
, String error, Throwable e) {
optionalLogger.ifPresent((Logger logger) -> logger.error(error, e));
}
public static void logInfo(Optional<Logger> optionalLogger
, String info) {
optionalLogger.ifPresent((Logger logger) -> logger.info(info));
}
}
}

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

@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.common.utils;
import org.apache.dolphinscheduler.common.shell.ShellExecutor;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import java.io.BufferedReader;
import java.io.FileInputStream;
@ -53,8 +54,6 @@ public class OSUtils {
private static final Logger logger = LoggerFactory.getLogger(OSUtils.class);
public static final ThreadLocal<Logger> taskLoggerThreadLocal = new ThreadLocal<>();
private static final SystemInfo SI = new SystemInfo();
public static final String TWO_DECIMAL = "0.00";
@ -64,7 +63,7 @@ public class OSUtils {
*/
public static final double NEGATIVE_ONE = -1;
private static HardwareAbstractionLayer hal = SI.getHardware();
private static final HardwareAbstractionLayer hal = SI.getHardware();
private OSUtils() {
throw new UnsupportedOperationException("Construct OSUtils");
@ -107,22 +106,6 @@ public class OSUtils {
return Double.parseDouble(df.format(availablePhysicalMemorySize));
}
/**
* get total physical memory size
* <p>
* Keep 2 decimal
*
* @return available Physical Memory Size, unit: G
*/
public static double totalPhysicalMemorySize() {
GlobalMemory memory = hal.getMemory();
double totalPhysicalMemorySize = memory.getTotal() / 1024.0 / 1024 / 1024;
DecimalFormat df = new DecimalFormat(TWO_DECIMAL);
df.setRoundingMode(RoundingMode.HALF_UP);
return Double.parseDouble(df.format(totalPhysicalMemorySize));
}
/**
* load average
*
@ -164,9 +147,9 @@ public class OSUtils {
public static List<String> getUserList() {
try {
if (isMacOS()) {
if (SystemUtils.IS_OS_MAC) {
return getUserListFromMac();
} else if (isWindows()) {
} else if (SystemUtils.IS_OS_WINDOWS) {
return getUserListFromWindows();
} else {
return getUserListFromLinux();
@ -262,14 +245,10 @@ public class OSUtils {
*/
public static void createUserIfAbsent(String userName) {
// if not exists this user, then create
taskLoggerThreadLocal.set(taskLoggerThreadLocal.get());
if (!getUserList().contains(userName)) {
boolean isSuccess = createUser(userName);
String infoLog = String.format("create user %s %s", userName, isSuccess ? "success" : "fail");
LoggerUtils.logInfo(Optional.ofNullable(logger), infoLog);
LoggerUtils.logInfo(Optional.ofNullable(taskLoggerThreadLocal.get()), infoLog);
logger.info("create user {} {}", userName, isSuccess ? "success" : "fail");
}
taskLoggerThreadLocal.remove();
}
/**
@ -283,21 +262,19 @@ public class OSUtils {
String userGroup = getGroup();
if (StringUtils.isEmpty(userGroup)) {
String errorLog = String.format("%s group does not exist for this operating system.", userGroup);
LoggerUtils.logError(Optional.ofNullable(logger), errorLog);
LoggerUtils.logError(Optional.ofNullable(taskLoggerThreadLocal.get()), errorLog);
logger.error(errorLog);
return false;
}
if (isMacOS()) {
if (SystemUtils.IS_OS_MAC) {
createMacUser(userName, userGroup);
} else if (isWindows()) {
} else if (SystemUtils.IS_OS_WINDOWS) {
createWindowsUser(userName, userGroup);
} else {
createLinuxUser(userName, userGroup);
}
return true;
} catch (Exception e) {
LoggerUtils.logError(Optional.ofNullable(logger), e);
LoggerUtils.logError(Optional.ofNullable(taskLoggerThreadLocal.get()), e);
logger.error(e.getMessage(), e);
}
return false;
@ -311,14 +288,9 @@ public class OSUtils {
* @throws IOException in case of an I/O error
*/
private static void createLinuxUser(String userName, String userGroup) throws IOException {
String infoLog1 = String.format("create linux os user : %s", userName);
LoggerUtils.logInfo(Optional.ofNullable(logger), infoLog1);
LoggerUtils.logInfo(Optional.ofNullable(taskLoggerThreadLocal.get()), infoLog1);
logger.info("create linux os user: {}", userName);
String cmd = String.format("sudo useradd -g %s %s", userGroup, userName);
String infoLog2 = String.format("execute cmd : %s", cmd);
LoggerUtils.logInfo(Optional.ofNullable(logger), infoLog2);
LoggerUtils.logInfo(Optional.ofNullable(taskLoggerThreadLocal.get()), infoLog2);
logger.info("execute cmd: {}", cmd);
exeCmd(cmd);
}
@ -330,23 +302,14 @@ public class OSUtils {
* @throws IOException in case of an I/O error
*/
private static void createMacUser(String userName, String userGroup) throws IOException {
Optional<Logger> optionalLogger = Optional.ofNullable(logger);
Optional<Logger> optionalTaskLogger = Optional.ofNullable(taskLoggerThreadLocal.get());
String infoLog1 = String.format("create mac os user : %s", userName);
LoggerUtils.logInfo(optionalLogger, infoLog1);
LoggerUtils.logInfo(optionalTaskLogger, infoLog1);
logger.info("create mac os user: {}", userName);
String createUserCmd = String.format("sudo sysadminctl -addUser %s -password %s", userName, userName);
String infoLog2 = String.format("create user command : %s", createUserCmd);
LoggerUtils.logInfo(optionalLogger, infoLog2);
LoggerUtils.logInfo(optionalTaskLogger, infoLog2);
logger.info("create user command: {}", createUserCmd);
exeCmd(createUserCmd);
String appendGroupCmd = String.format("sudo dseditgroup -o edit -a %s -t user %s", userName, userGroup);
String infoLog3 = String.format("append user to group : %s", appendGroupCmd);
LoggerUtils.logInfo(optionalLogger, infoLog3);
LoggerUtils.logInfo(optionalTaskLogger, infoLog3);
logger.info("append user to group: {}", appendGroupCmd);
exeCmd(appendGroupCmd);
}
@ -358,20 +321,14 @@ public class OSUtils {
* @throws IOException in case of an I/O error
*/
private static void createWindowsUser(String userName, String userGroup) throws IOException {
String infoLog1 = String.format("create windows os user : %s", userName);
LoggerUtils.logInfo(Optional.ofNullable(logger), infoLog1);
LoggerUtils.logInfo(Optional.ofNullable(taskLoggerThreadLocal.get()), infoLog1);
logger.info("create windows os user: {}", userName);
String userCreateCmd = String.format("net user \"%s\" /add", userName);
String infoLog2 = String.format("execute create user command : %s", userCreateCmd);
LoggerUtils.logInfo(Optional.ofNullable(logger), infoLog2);
LoggerUtils.logInfo(Optional.ofNullable(taskLoggerThreadLocal.get()), infoLog2);
logger.info("execute create user command: {}", userCreateCmd);
exeCmd(userCreateCmd);
String appendGroupCmd = String.format("net localgroup \"%s\" \"%s\" /add", userGroup, userName);
String infoLog3 = String.format("execute append user to group : %s", appendGroupCmd);
LoggerUtils.logInfo(Optional.ofNullable(logger), infoLog3);
LoggerUtils.logInfo(Optional.ofNullable(taskLoggerThreadLocal.get()), infoLog3);
logger.info("execute append user to group: {}", appendGroupCmd);
exeCmd(appendGroupCmd);
}
@ -382,7 +339,7 @@ public class OSUtils {
* @throws IOException errors
*/
public static String getGroup() throws IOException {
if (isWindows()) {
if (SystemUtils.IS_OS_WINDOWS) {
String currentProcUserName = System.getProperty("user.name");
String result = exeCmd(String.format("net user \"%s\"", currentProcUserName));
String line = result.split("\n")[22];
@ -454,33 +411,6 @@ public class OSUtils {
return Integer.parseInt(runtimeMXBean.getName().split("@")[0]);
}
/**
* whether is macOS
*
* @return true if mac
*/
public static boolean isMacOS() {
return getOSName().startsWith("Mac");
}
/**
* whether is windows
*
* @return true if windows
*/
public static boolean isWindows() {
return getOSName().startsWith("Windows");
}
/**
* get current OS name
*
* @return current OS name
*/
public static String getOSName() {
return System.getProperty("os.name");
}
/**
* check memory and cpu usage
*

94
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/ParameterUtils.java
View File

@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.common.utils;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.utils.placeholder.BusinessTimeUtils;
import org.apache.dolphinscheduler.common.utils.placeholder.PlaceholderUtils;
@ -27,7 +26,6 @@ import org.apache.dolphinscheduler.common.utils.placeholder.TimePlaceholderUtils
import org.apache.commons.lang.StringUtils;
import java.sql.PreparedStatement;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
@ -37,16 +35,10 @@ import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* parameter parse utils
*/
public class ParameterUtils {
private static final Logger logger = LoggerFactory.getLogger(ParameterUtils.class);
private static final String DATE_PARSE_PATTERN = "\\$\\[([^\\$\\]]+)]";
private static final String DATE_START_PATTERN = "^[0-9]";
@ -85,72 +77,6 @@ public class ParameterUtils {
return parameterString;
}
/**
* new
* convert parameters place holders
*
* @param parameterString parameter
* @param parameterMap parameter map
* @return convert parameters place holders
*/
public static String convertParameterPlaceholders2(String parameterString, Map<String, String> parameterMap) {
if (StringUtils.isEmpty(parameterString)) {
return parameterString;
}
//Get current time, schedule execute time
String cronTimeStr = parameterMap.get(Constants.PARAMETER_SHECDULE_TIME);
Date cronTime = null;
if (!StringUtils.isEmpty(cronTimeStr)) {
cronTime = DateUtils.parse(cronTimeStr, Constants.PARAMETER_FORMAT_TIME);
} else {
cronTime = new Date();
}
// replace variable ${} form,refers to the replacement of system variables and custom variables
if (!parameterMap.isEmpty()) {
parameterString = PlaceholderUtils.replacePlaceholders(parameterString, parameterMap, true);
}
// replace time $[...] form, eg. $[yyyyMMdd]
if (cronTime != null) {
return dateTemplateParse(parameterString, cronTime);
}
return parameterString;
}
/**
* set in parameter
*
* @param index index
* @param stmt preparedstatement
* @param dataType data type
* @param value value
* @throws Exception errors
*/
public static void setInParameter(int index, PreparedStatement stmt, DataType dataType, String value) throws Exception {
if (dataType.equals(DataType.VARCHAR)) {
stmt.setString(index, value);
} else if (dataType.equals(DataType.INTEGER)) {
stmt.setInt(index, Integer.parseInt(value));
} else if (dataType.equals(DataType.LONG)) {
stmt.setLong(index, Long.parseLong(value));
} else if (dataType.equals(DataType.FLOAT)) {
stmt.setFloat(index, Float.parseFloat(value));
} else if (dataType.equals(DataType.DOUBLE)) {
stmt.setDouble(index, Double.parseDouble(value));
} else if (dataType.equals(DataType.DATE)) {
stmt.setDate(index, java.sql.Date.valueOf(value));
} else if (dataType.equals(DataType.TIME)) {
stmt.setString(index, value);
} else if (dataType.equals(DataType.TIMESTAMP)) {
stmt.setTimestamp(index, java.sql.Timestamp.valueOf(value));
} else if (dataType.equals(DataType.BOOLEAN)) {
stmt.setBoolean(index, Boolean.parseBoolean(value));
}
}
/**
* curing user define parameters
*
@ -217,26 +143,6 @@ public class ParameterUtils {
return inputString;
}
/**
* $[yyyyMMdd] replace schedule time
*/
public static String replaceScheduleTime(String text, Date scheduleTime) {
Map<String, Property> paramsMap = new HashMap<>();
//if getScheduleTime null ,is current date
if (null == scheduleTime) {
scheduleTime = new Date();
}
String dateTime = org.apache.dolphinscheduler.common.utils.DateUtils.format(scheduleTime, Constants.PARAMETER_FORMAT_TIME);
Property p = new Property();
p.setValue(dateTime);
p.setProp(Constants.PARAMETER_SHECDULE_TIME);
paramsMap.put(Constants.PARAMETER_SHECDULE_TIME, p);
text = ParameterUtils.convertParameterPlaceholders2(text, convert(paramsMap));
return text;
}
/**
* format convert
*

34
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/RetryerUtils.java
View File

@ -23,20 +23,13 @@ import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.github.rholder.retry.RetryException;
import com.github.rholder.retry.Retryer;
import com.github.rholder.retry.RetryerBuilder;
import com.github.rholder.retry.StopStrategies;
import com.github.rholder.retry.WaitStrategies;
/**
* The Retryer util.
*/
public class RetryerUtils {
private static final Logger logger = LoggerFactory.getLogger(RetryerUtils.class);
private static Retryer<Boolean> defaultRetryerResultCheck;
private static Retryer<Boolean> defaultRetryerResultNoCheck;
@ -114,31 +107,4 @@ public class RetryerUtils {
public static Boolean retryCall(final Callable<Boolean> callable) throws ExecutionException, RetryException {
return retryCall(callable, true);
}
/**
* Retry call silent without exceptions thrown
*
* @param callable the callable
* @param checkResult whether check result
* @return if no exceptions ,it's result returned by callable ,else always false
*/
public static boolean retryCallSilent(final Callable<Boolean> callable, boolean checkResult) {
boolean result = false;
try {
result = getDefaultRetryer(checkResult).call(callable);
} catch (ExecutionException | RetryException e) {
logger.warn("Retry call {} failed {}", callable, e.getMessage(), e);
}
return result;
}
/**
* Retry call silent without exceptions thrown
*
* @param callable the callable
* @return if no exceptions ,it's result returned by callable ,else always false
*/
public static boolean retryCallSilent(final Callable<Boolean> callable) {
return retryCallSilent(callable, true);
}
}

46
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/SchemaUtils.java
View File

@ -23,8 +23,6 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@ -38,7 +36,7 @@ import org.slf4j.LoggerFactory;
public class SchemaUtils {
private static final Logger logger = LoggerFactory.getLogger(SchemaUtils.class);
private static Pattern p = Pattern.compile("\\s*|\t|\r|\n");
private static final Pattern p = Pattern.compile("\\s*|\t|\r|\n");
private SchemaUtils() {
throw new UnsupportedOperationException("Construct SchemaUtils");
@ -49,7 +47,6 @@ public class SchemaUtils {
*
* @return all schema list
*/
@SuppressWarnings("unchecked")
public static List<String> getAllSchemaList() {
List<String> schemaDirList = new ArrayList<>();
File[] schemaDirArr = FileUtils.getAllDir("sql/upgrade");
@ -61,28 +58,25 @@ public class SchemaUtils {
schemaDirList.add(file.getName());
}
Collections.sort(schemaDirList, new Comparator() {
@Override
public int compare(Object o1, Object o2) {
try {
String dir1 = String.valueOf(o1);
String dir2 = String.valueOf(o2);
String version1 = dir1.split("_")[0];
String version2 = dir2.split("_")[0];
if (version1.equals(version2)) {
return 0;
}
if (SchemaUtils.isAGreatVersion(version1, version2)) {
return 1;
}
return -1;
} catch (Exception e) {
logger.error(e.getMessage(), e);
throw new RuntimeException(e);
schemaDirList.sort((o1, o2) -> {
try {
String dir1 = String.valueOf(o1);
String dir2 = String.valueOf(o2);
String version1 = dir1.split("_")[0];
String version2 = dir2.split("_")[0];
if (version1.equals(version2)) {
return 0;
}
if (SchemaUtils.isAGreatVersion(version1, version2)) {
return 1;
}
return -1;
} catch (Exception e) {
logger.error(e.getMessage(), e);
throw new RuntimeException(e);
}
});
@ -124,7 +118,7 @@ public class SchemaUtils {
public static String getSoftVersion() {
String softVersion;
try {
softVersion = FileUtils.readFile2Str(new FileInputStream(new File("sql/soft_version")));
softVersion = FileUtils.readFile2Str(new FileInputStream("sql/soft_version"));
softVersion = replaceBlank(softVersion);
} catch (FileNotFoundException e) {
logger.error(e.getMessage(), e);

45
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/SensitiveLogUtils.java
View File

@ -1,45 +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.dolphinscheduler.common.utils;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.commons.lang.StringUtils;
/**
* sensitive log Util
*/
public class SensitiveLogUtils {
private SensitiveLogUtils() {
throw new UnsupportedOperationException("Construct SensitiveLogUtils");
}
/**
* @param dataSourcePwd data source password
* @return String
*/
public static String maskDataSourcePwd(String dataSourcePwd) {
if (!StringUtils.isEmpty(dataSourcePwd)) {
dataSourcePwd = Constants.PASSWORD_DEFAULT;
}
return dataSourcePwd;
}
}

82
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/placeholder/PropertyPlaceholderHelper.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.dolphinscheduler.common.utils.placeholder;
import static java.util.Objects.requireNonNull;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@ -36,7 +37,7 @@ public class PropertyPlaceholderHelper {
private static final Log logger = LogFactory.getLog(PropertyPlaceholderHelper.class);
private static final Map<String, String> wellKnownSimplePrefixes = new HashMap<String, String>(4);
private static final Map<String, String> wellKnownSimplePrefixes = new HashMap<>(4);
static {
wellKnownSimplePrefixes.put("}", "{");
@ -58,16 +59,7 @@ public class PropertyPlaceholderHelper {
/**
* Creates a new {@code PropertyPlaceholderHelper} that uses the supplied prefix and suffix.
* Unresolvable placeholders are ignored.
* @param placeholderPrefix the prefix that denotes the start of a placeholder
* @param placeholderSuffix the suffix that denotes the end of a placeholder
*/
public PropertyPlaceholderHelper(String placeholderPrefix, String placeholderSuffix) {
this(placeholderPrefix, placeholderSuffix, null, true);
}
/**
* Creates a new {@code PropertyPlaceholderHelper} that uses the supplied prefix and suffix.
*
* @param placeholderPrefix the prefix that denotes the start of a placeholder
* @param placeholderSuffix the suffix that denotes the end of a placeholder
* @param valueSeparator the separating character between the placeholder variable
@ -78,15 +70,14 @@ public class PropertyPlaceholderHelper {
public PropertyPlaceholderHelper(String placeholderPrefix, String placeholderSuffix,
String valueSeparator, boolean ignoreUnresolvablePlaceholders) {
notNull(placeholderPrefix, "'placeholderPrefix' must not be null");
notNull(placeholderSuffix, "'placeholderSuffix' must not be null");
requireNonNull((Object) placeholderPrefix, "'placeholderPrefix' must not be null");
requireNonNull((Object) placeholderSuffix, "'placeholderSuffix' must not be null");
this.placeholderPrefix = placeholderPrefix;
this.placeholderSuffix = placeholderSuffix;
String simplePrefixForSuffix = wellKnownSimplePrefixes.get(this.placeholderSuffix);
if (simplePrefixForSuffix != null && this.placeholderPrefix.endsWith(simplePrefixForSuffix)) {
this.simplePrefix = simplePrefixForSuffix;
}
else {
} else {
this.simplePrefix = this.placeholderPrefix;
}
this.valueSeparator = valueSeparator;
@ -94,37 +85,21 @@ public class PropertyPlaceholderHelper {
}
/**
* Replaces all placeholders of format {@code ${name}} with the corresponding
* property from the supplied {@link Properties}.
* @param value the value containing the placeholders to be replaced
* @param properties the {@code Properties} to use for replacement
* @return the supplied value with placeholders replaced inline
*/
public String replacePlaceholders(String value, final Properties properties) {
notNull(properties, "'properties' must not be null");
return replacePlaceholders(value, new PlaceholderResolver() {
@Override
public String resolvePlaceholder(String placeholderName) {
return properties.getProperty(placeholderName);
}
});
}
/**
* Replaces all placeholders of format {@code ${name}} with the value returned
* from the supplied {@link PlaceholderResolver}.
*
* @param value the value containing the placeholders to be replaced
* @param placeholderResolver the {@code PlaceholderResolver} to use for replacement
* @return the supplied value with placeholders replaced inline
*/
public String replacePlaceholders(String value, PlaceholderResolver placeholderResolver) {
notNull(value, "'value' must not be null");
requireNonNull((Object) value, "'value' must not be null");
return parseStringValue(value, placeholderResolver, new HashSet<String>());
}
protected String parseStringValue(
String value, PlaceholderResolver placeholderResolver, Set<String> visitedPlaceholders) {
String value, PlaceholderResolver placeholderResolver, Set<String> visitedPlaceholders) {
StringBuilder result = new StringBuilder(value);
@ -136,7 +111,7 @@ public class PropertyPlaceholderHelper {
String originalPlaceholder = placeholder;
if (!visitedPlaceholders.add(originalPlaceholder)) {
throw new IllegalArgumentException(
"Circular placeholder reference '" + originalPlaceholder + "' in property definitions");
"Circular placeholder reference '" + originalPlaceholder + "' in property definitions");
}
// Recursive invocation, parsing placeholders contained in the placeholder key.
placeholder = parseStringValue(placeholder, placeholderResolver, visitedPlaceholders);
@ -162,18 +137,15 @@ public class PropertyPlaceholderHelper {
logger.trace("Resolved placeholder '" + placeholder + "'");
}
startIndex = result.indexOf(this.placeholderPrefix, startIndex + propVal.length());
}
else if (this.ignoreUnresolvablePlaceholders) {
} else if (this.ignoreUnresolvablePlaceholders) {
// Proceed with unprocessed value.
startIndex = result.indexOf(this.placeholderPrefix, endIndex + this.placeholderSuffix.length());
}
else {
} else {
throw new IllegalArgumentException("Could not resolve placeholder '" +
placeholder + "'" + " in value \"" + value + "\"");
placeholder + "'" + " in value \"" + value + "\"");
}
visitedPlaceholders.remove(originalPlaceholder);
}
else {
} else {
startIndex = -1;
}
}
@ -189,16 +161,13 @@ public class PropertyPlaceholderHelper {
if (withinNestedPlaceholder > 0) {
withinNestedPlaceholder--;
index = index + this.placeholderSuffix.length();
}
else {
} else {
return index;
}
}
else if (substringMatch(buf, index, this.simplePrefix)) {
} else if (substringMatch(buf, index, this.simplePrefix)) {
withinNestedPlaceholder++;
index = index + this.simplePrefix.length();
}
else {
} else {
index++;
}
}
@ -213,6 +182,7 @@ public class PropertyPlaceholderHelper {
/**
* Resolve the supplied placeholder name to the replacement value.
*
* @param placeholderName the name of the placeholder to resolve
* @return the replacement value, or {@code null} if no replacement is to be made
*/
@ -222,6 +192,7 @@ public class PropertyPlaceholderHelper {
/**
* Test whether the given string matches the given substring
* at the given index.
*
* @param str the original string (or StringBuilder)
* @param index the index in the original string to start matching against
* @param substring the substring to match at the given index
@ -236,20 +207,5 @@ public class PropertyPlaceholderHelper {
}
return true;
}
/**
* Assert that an object is not {@code null}.
* <pre class="code">Assert.notNull(clazz, "The class must not be null");</pre>
* @param object the object to check
* @param message the exception message to use if the assertion fails
* @throws IllegalArgumentException if the object is {@code null}
*/
public static void notNull(Object object, String message) {
if (object == null) {
throw new IllegalArgumentException(message);
}
}
}

6
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/placeholder/TimePlaceholderUtils.java
View File

@ -24,7 +24,7 @@ import static org.apache.dolphinscheduler.common.Constants.DIVISION_STRING;
import static org.apache.dolphinscheduler.common.Constants.LEFT_BRACE_CHAR;
import static org.apache.dolphinscheduler.common.Constants.LEFT_BRACE_STRING;
import static org.apache.dolphinscheduler.common.Constants.MULTIPLY_CHAR;
import static org.apache.dolphinscheduler.common.Constants.MULTIPLY_STRING;
import static org.apache.dolphinscheduler.common.Constants.STAR;
import static org.apache.dolphinscheduler.common.Constants.N;
import static org.apache.dolphinscheduler.common.Constants.P;
import static org.apache.dolphinscheduler.common.Constants.RIGHT_BRACE_CHAR;
@ -266,9 +266,9 @@ public class TimePlaceholderUtils {
* @return true or false
*/
private static boolean compare(String peek, String cur) {
if (MULTIPLY_STRING.equals(peek) && (DIVISION_STRING.equals(cur) || MULTIPLY_STRING.equals(cur) || ADD_STRING.equals(cur) || SUBTRACT_STRING.equals(cur))) {
if (STAR.equals(peek) && (DIVISION_STRING.equals(cur) || STAR.equals(cur) || ADD_STRING.equals(cur) || SUBTRACT_STRING.equals(cur))) {
return true;
} else if (DIVISION_STRING.equals(peek) && (DIVISION_STRING.equals(cur) || MULTIPLY_STRING.equals(cur) || ADD_STRING.equals(cur) || SUBTRACT_STRING.equals(cur))) {
} else if (DIVISION_STRING.equals(peek) && (DIVISION_STRING.equals(cur) || STAR.equals(cur) || ADD_STRING.equals(cur) || SUBTRACT_STRING.equals(cur))) {
return true;
} else if (ADD_STRING.equals(peek) && (ADD_STRING.equals(cur) || SUBTRACT_STRING.equals(cur))) {
return true;

5
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/ConstantsTest.java
View File

@ -16,7 +16,8 @@
*/
package org.apache.dolphinscheduler.common;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.commons.lang.SystemUtils;
import org.junit.Assert;
import org.junit.Test;
@ -30,7 +31,7 @@ public class ConstantsTest {
*/
@Test
public void testPID() {
if (OSUtils.isWindows()) {
if (SystemUtils.IS_OS_WINDOWS) {
Assert.assertEquals(Constants.PID, "handle");
} else {
Assert.assertEquals(Constants.PID, "pid");

10
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/os/OSUtilsTest.java
View File

@ -39,16 +39,6 @@ public class OSUtilsTest {
Assert.assertTrue(memoryUsage >= 0.0);
}
@Test
public void physicalMemorySize() {
double availablePhysicalMemorySize = OSUtils.availablePhysicalMemorySize();
double totalPhysicalMemorySize = OSUtils.totalPhysicalMemorySize();
logger.info("availablePhysicalMemorySize : {}", availablePhysicalMemorySize);
logger.info("totalPhysicalMemorySize : {}", totalPhysicalMemorySize);
Assert.assertTrue(availablePhysicalMemorySize >= 0.0);
Assert.assertTrue(totalPhysicalMemorySize >= 0.0);
}
@Test
public void loadAverage() {
double loadAverage = OSUtils.loadAverage();

4
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/task/FlinkParametersTest.java
View File

@ -18,7 +18,9 @@ package org.apache.dolphinscheduler.common.task;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.flink.FlinkParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.commons.collections.CollectionUtils;
import org.junit.Assert;
import org.junit.Test;

3
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/task/SparkParametersTest.java
View File

@ -19,7 +19,8 @@ package org.apache.dolphinscheduler.common.task;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.spark.SparkParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.LinkedList;
import java.util.List;

3
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/task/SqlParametersTest.java
View File

@ -23,7 +23,8 @@ import org.apache.dolphinscheduler.common.enums.DataType;
import org.apache.dolphinscheduler.common.enums.Direct;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.List;

161
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/CollectionUtilsTest.java
View File

@ -1,161 +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.dolphinscheduler.common.utils;
import org.apache.dolphinscheduler.common.Constants;
import org.junit.Assert;
import org.junit.Test;
import java.util.*;
public class CollectionUtilsTest {
@Test
public void equalLists() {
Assert.assertTrue(CollectionUtils.equalLists(null,null));
Assert.assertTrue(CollectionUtils.equalLists(new ArrayList<Integer>(),new ArrayList<Integer>()));
List<Integer> a = new ArrayList<Integer>();
a.add(1);
a.add(2);
List<Integer> b = new ArrayList<Integer>();
b.add(1);
b.add(2);
Assert.assertTrue(CollectionUtils.equalLists(a, b));
a.add(1);
Assert.assertFalse(CollectionUtils.equalLists(a, b));
b.add(2);
Assert.assertFalse(CollectionUtils.equalLists(a, b));
a.add(2);
b.add(1);
a.add(4);
b.add(2);
Assert.assertFalse(CollectionUtils.equalLists(a, b));
Assert.assertFalse(CollectionUtils.equalLists(null, new ArrayList<Integer>()));
Assert.assertFalse(CollectionUtils.equalLists(new ArrayList<Integer>(), null));
}
@Test
public void subtract() {
Set<Integer> a = new HashSet<Integer>();
a.add(1);
a.add(2);
a.add(3);
Set<Integer> b = new HashSet<Integer>();
b.add(0);
b.add(2);
b.add(4);
Assert.assertArrayEquals(new Integer[]{1,3},CollectionUtils.subtract(a,b).toArray());
}
@Test
public void stringToMap() {
Map<String, String> a = CollectionUtils.stringToMap("a=b;c=d;", ";");
Assert.assertNotNull(a);
Assert.assertTrue(a.size() == 2);
a = CollectionUtils.stringToMap(null, ";");
Assert.assertTrue(a.isEmpty());
a = CollectionUtils.stringToMap("", ";");
Assert.assertTrue(a.isEmpty());
a = CollectionUtils.stringToMap("a=b;c=d", "");
Assert.assertTrue(a.isEmpty());
a = CollectionUtils.stringToMap("a=b;c=d", null);
Assert.assertTrue(a.isEmpty());
a = CollectionUtils.stringToMap("a=b;c=d;e=f", ";");
Assert.assertEquals(3, a.size());
a = CollectionUtils.stringToMap("a;b=f", ";");
Assert.assertTrue(a.isEmpty());
a = CollectionUtils.stringToMap("a=b;c=d;e=f;", ";", "test");
Assert.assertEquals(3, a.size());
Assert.assertNotNull(a.get("testa"));
}
@Test
public void getListByExclusion() {
Assert.assertNotNull(CollectionUtils.getListByExclusion(null, null));
List<Integer> originList = new ArrayList<>();
originList.add(1);
originList.add(2);
List<Map<String, Object>> ret = CollectionUtils.getListByExclusion(originList, null);
Assert.assertEquals(2, ret.size());
ret = CollectionUtils.getListByExclusion(originList, new HashSet<>());
Assert.assertEquals(2, ret.size());
Assert.assertFalse(ret.get(0).isEmpty());
Set<String> exclusion = new HashSet<>();
exclusion.add(Constants.CLASS);
ret = CollectionUtils.getListByExclusion(originList, exclusion);
Assert.assertEquals(2, ret.size());
Assert.assertTrue(ret.get(0).isEmpty());
}
@Test
public void isNotEmpty() {
List<Integer> list = new ArrayList<>();
Assert.assertFalse(CollectionUtils.isNotEmpty(list));
Assert.assertFalse(CollectionUtils.isNotEmpty(null));
}
@Test
public void isEmpty(){
List<Integer> list = new ArrayList<>();
Assert.assertTrue(CollectionUtils.isEmpty(list));
Assert.assertTrue(CollectionUtils.isEmpty(null));
list.add(1);
Assert.assertFalse(CollectionUtils.isEmpty(list));
}
@Test
public void isEqualCollection() {
List<Integer> a = new ArrayList<>();
a.add(1);
List<Integer> b = new ArrayList<>();
b.add(1);
Assert.assertTrue(CollectionUtils.isEqualCollection(a,b));
b.add(2);
Assert.assertFalse(CollectionUtils.isEqualCollection(a,b));
}
@Test
public void getCardinalityMap(){
List<Integer> a = new ArrayList<>();
a.add(1);
a.add(2);
a.add(2);
a.add(3);
a.add(3);
a.add(3);
Map<Integer,Integer> cardinalityMap = CollectionUtils.getCardinalityMap(a);
Assert.assertEquals(3, cardinalityMap.size());
Assert.assertEquals(1, cardinalityMap.get(1).intValue());
Assert.assertEquals(2, cardinalityMap.get(2).intValue());
Assert.assertEquals(3, cardinalityMap.get(3).intValue());
}
@Test
public void transformToList() {
List<String> stringList = new ArrayList<>();
stringList.add("1");
List<Integer> integers = CollectionUtils.transformToList(stringList, String::length);
Assert.assertFalse(integers.isEmpty());
}
@Test
public void collectionToMap() {
List<String> stringList = new ArrayList<>();
stringList.add("1");
Map<Integer, String> lengthStringMap = CollectionUtils.collectionToMap(stringList, String::length);
Assert.assertFalse(lengthStringMap.isEmpty());
}
}

8
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/FileUtilsTest.java
View File

@ -35,14 +35,6 @@ import org.powermock.modules.junit4.PowerMockRunner;
@PrepareForTest(DateUtils.class)
public class FileUtilsTest {
@Test
public void suffix() {
Assert.assertEquals("java", FileUtils.suffix("ninfor.java"));
Assert.assertEquals("", FileUtils.suffix(null));
Assert.assertEquals("", FileUtils.suffix(""));
Assert.assertEquals("", FileUtils.suffix("ninfor-java"));
}
@Test
public void testGetDownloadFilename() {
PowerMockito.mockStatic(DateUtils.class);

85
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/LoggerUtilsTest.java
View File

@ -1,85 +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.dolphinscheduler.common.utils;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.List;
import java.util.Optional;
import org.junit.Assert;
import org.junit.Test;
import org.junit.Test.None;
import org.junit.runner.RunWith;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@RunWith(PowerMockRunner.class)
@PrepareForTest({LoggerUtils.class})
public class LoggerUtilsTest {
private Logger logger = LoggerFactory.getLogger(LoggerUtilsTest.class);
@Test
public void buildTaskId() {
String taskId = LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX, 798L,1,4084, 15210);
Assert.assertEquals(" - [taskAppId=TASK-798_1-4084-15210]", taskId);
}
@Test
public void getAppIds() {
List<String> appIdList = LoggerUtils.getAppIds("Running job: application_1_1", logger);
Assert.assertEquals("application_1_1", appIdList.get(0));
}
@Test
public void testReadWholeFileContent() throws Exception {
BufferedReader bufferedReader = PowerMockito.mock(BufferedReader.class);
PowerMockito.whenNew(BufferedReader.class).withAnyArguments().thenReturn(bufferedReader);
PowerMockito.when(bufferedReader.readLine()).thenReturn("").thenReturn(null);
FileInputStream fileInputStream = PowerMockito.mock(FileInputStream.class);
PowerMockito.whenNew(FileInputStream.class).withAnyArguments().thenReturn(fileInputStream);
InputStreamReader inputStreamReader = PowerMockito.mock(InputStreamReader.class);
PowerMockito.whenNew(InputStreamReader.class).withAnyArguments().thenReturn(inputStreamReader);
String log = LoggerUtils.readWholeFileContent("/tmp/log");
Assert.assertNotNull(log);
PowerMockito.when(bufferedReader.readLine()).thenThrow(new IOException());
log = LoggerUtils.readWholeFileContent("/tmp/log");
Assert.assertNotNull(log);
}
@Test(expected = None.class)
public void testLogError() {
Optional<Logger> loggerOptional = Optional.of(this.logger);
LoggerUtils.logError(loggerOptional, "error message");
LoggerUtils.logError(loggerOptional, new RuntimeException("error message"));
LoggerUtils.logError(loggerOptional, "error message", new RuntimeException("runtime exception"));
LoggerUtils.logInfo(loggerOptional, "info message");
}
}

8
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/OSUtilsTest.java
View File

@ -16,6 +16,8 @@
*/
package org.apache.dolphinscheduler.common.utils;
import org.apache.commons.lang.SystemUtils;
import java.io.IOException;
import java.util.List;
@ -36,11 +38,9 @@ public class OSUtilsTest {
@Test
public void testOSMetric() {
if (!OSUtils.isWindows()) {
if (!SystemUtils.IS_OS_WINDOWS) {
double availablePhysicalMemorySize = OSUtils.availablePhysicalMemorySize();
Assert.assertTrue(availablePhysicalMemorySize >= 0.0d);
double totalPhysicalMemorySize = OSUtils.totalPhysicalMemorySize();
Assert.assertTrue(totalPhysicalMemorySize >= 0.0d);
double loadAverage = OSUtils.loadAverage();
logger.info("loadAverage {}", loadAverage);
double memoryUsage = OSUtils.memoryUsage();
@ -87,7 +87,7 @@ public class OSUtilsTest {
@Test
public void exeCmd() {
if (OSUtils.isMacOS() || !OSUtils.isWindows()) {
if (SystemUtils.IS_OS_MAC || !SystemUtils.IS_OS_WINDOWS) {
try {
String result = OSUtils.exeCmd("echo helloWorld");
Assert.assertEquals("helloWorld\n",result);

109
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/RetryerUtilsTest.java
View File

@ -213,113 +213,4 @@ public class RetryerUtilsTest {
testRetryExceptionWithPara(true);
testRetryExceptionWithPara(false);
}
@Test
public void testRetrySilent() {
try {
for (int execTarget = 1; execTarget <= 3; execTarget++) {
int finalExecTarget = execTarget;
int[] execTime = {0};
boolean result = RetryerUtils.retryCallSilent(() -> {
execTime[0]++;
return execTime[0] == finalExecTarget;
});
Assert.assertEquals(finalExecTarget, execTime[0]);
Assert.assertTrue(result);
}
} catch (Exception e) {
Assert.fail("Unexpected exception " + e.getMessage());
}
int[] execTime = {0};
try {
boolean result = RetryerUtils.retryCallSilent(() -> {
execTime[0]++;
return execTime[0] == 4;
});
Assert.assertFalse(result);
} catch (Exception e) {
Assert.fail("Unexpected exception " + e.getMessage());
}
}
@Test
public void testRetrySilentWithPara() {
try {
for (int execTarget = 1; execTarget <= 3; execTarget++) {
int finalExecTarget = execTarget;
int[] execTime = {0};
boolean result = RetryerUtils.retryCallSilent(() -> {
execTime[0]++;
return execTime[0] == finalExecTarget;
}, true);
Assert.assertEquals(finalExecTarget, execTime[0]);
Assert.assertTrue(result);
}
} catch (Exception e) {
Assert.fail("Unexpected exception " + e.getMessage());
}
int[] execTime = {0};
try {
boolean result = RetryerUtils.retryCallSilent(() -> {
execTime[0]++;
return execTime[0] == 4;
}, true);
Assert.assertFalse(result);
} catch (Exception e) {
Assert.fail("Unexpected exception " + e.getMessage());
}
}
@Test
public void testRetrySilentNoCheckResult(){
try {
for (int execTarget = 1; execTarget <= 5; execTarget++) {
int[] execTime = {0};
boolean result = RetryerUtils.retryCallSilent(() -> {
execTime[0]++;
return execTime[0] > 1;
}, false);
Assert.assertEquals(1, execTime[0]);
Assert.assertFalse(result);
}
} catch (Exception e) {
Assert.fail("Unexpected exception " + e.getMessage());
}
}
private void testRetrySilentExceptionWithPara(boolean checkResult) {
try {
for (int execTarget = 1; execTarget <= 3; execTarget++) {
int finalExecTarget = execTarget;
int[] execTime = {0};
boolean result = RetryerUtils.retryCallSilent(() -> {
execTime[0]++;
if (execTime[0] != finalExecTarget) {
throw new IllegalArgumentException(String.valueOf(execTime[0]));
}
return true;
}, checkResult);
Assert.assertEquals(finalExecTarget, execTime[0]);
Assert.assertTrue(result);
}
} catch (Exception e) {
Assert.fail("Unexpected exception " + e.getMessage());
}
int[] execTime = {0};
try {
boolean result = RetryerUtils.retryCallSilent(() -> {
execTime[0]++;
if (execTime[0] != 4) {
throw new IllegalArgumentException(String.valueOf(execTime[0]));
}
return true;
}, checkResult);
Assert.assertFalse(result);
} catch (Exception e) {
Assert.fail("Unexpected exception " + e.getMessage());
}
}
@Test
public void testRetrySilentException() {
testRetrySilentExceptionWithPara(true);
testRetrySilentExceptionWithPara(false);
}
}

3
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/SchemaUtilsTest.java
View File

@ -105,7 +105,8 @@ public class SchemaUtilsTest {
List<String> real = SchemaUtils.getAllSchemaList();
List<String> expect = Arrays.asList("1.0.1_schema", "1.0.2_schema",
"1.1.0_schema", "1.2.0_schema");
Assert.assertTrue(CollectionUtils.isEqualCollection(real, expect));
boolean result = org.apache.commons.collections.CollectionUtils.isEqualCollection(real, expect);
Assert.assertTrue(result);
//normal
files = new File[0];

37
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/SensitiveLogUtilsTest.java
View File

@ -1,37 +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.dolphinscheduler.common.utils;
import org.apache.dolphinscheduler.common.Constants;
import org.junit.Assert;
import org.junit.Test;
public class SensitiveLogUtilsTest {
@Test
public void testMaskDataSourcePwd() {
String password = "123456";
String emptyPassword = "";
Assert.assertEquals(Constants.PASSWORD_DEFAULT, SensitiveLogUtils.maskDataSourcePwd(password));
Assert.assertEquals("", SensitiveLogUtils.maskDataSourcePwd(emptyPassword));
}
}

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

@ -19,11 +19,12 @@ package org.apache.dolphinscheduler.dao;
import static java.util.Objects.requireNonNull;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.datasource.ConnectionFactory;
import org.apache.dolphinscheduler.dao.entity.PluginDefine;
import org.apache.dolphinscheduler.dao.mapper.PluginDefineMapper;
import org.apache.commons.collections.CollectionUtils;
import java.util.List;
import org.slf4j.Logger;

2
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/upgrade/UpgradeDao.java
View File

@ -25,7 +25,6 @@ import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.common.enums.TimeoutFlag;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.TaskTimeoutParameter;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.ConnectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.SchemaUtils;
@ -38,6 +37,7 @@ import org.apache.dolphinscheduler.dao.entity.ProcessDefinitionLog;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelationLog;
import org.apache.dolphinscheduler.dao.entity.TaskDefinitionLog;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.io.File;

6
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/DagHelper.java
View File

@ -25,11 +25,12 @@ import org.apache.dolphinscheduler.common.process.ProcessDag;
import org.apache.dolphinscheduler.common.task.conditions.ConditionsParameters;
import org.apache.dolphinscheduler.common.task.switchtask.SwitchParameters;
import org.apache.dolphinscheduler.common.task.switchtask.SwitchResultVo;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessTaskRelation;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
@ -87,8 +88,7 @@ public class DagHelper {
List<TaskNode> destFlowNodeList = new ArrayList<>();
List<String> startNodeList = startNodeNameList;
if (taskDependType != TaskDependType.TASK_POST
&& CollectionUtils.isEmpty(startNodeList)) {
if (taskDependType != TaskDependType.TASK_POST && CollectionUtils.isEmpty(startNodeList)) {
logger.error("start node list is empty! cannot continue run the process ");
return destFlowNodeList;
}

2
dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/ResourceProcessDefinitionUtils.java
View File

@ -16,7 +16,7 @@
*/
package org.apache.dolphinscheduler.dao.utils;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.Arrays;
import java.util.HashMap;

3
dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/mapper/ResourceMapperTest.java
View File

@ -25,13 +25,14 @@ import static org.junit.Assert.assertThat;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.Resource;
import org.apache.dolphinscheduler.dao.entity.ResourcesUser;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.commons.collections.CollectionUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;

11
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/log/SensitiveDataConverter.java
View File

@ -17,17 +17,18 @@
package org.apache.dolphinscheduler.server.log;
import ch.qos.logback.classic.pattern.MessageConverter;
import ch.qos.logback.classic.spi.ILoggingEvent;
import static org.apache.dolphinscheduler.common.Constants.STAR;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.SensitiveLogUtils;
import org.apache.commons.lang.StringUtils;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import ch.qos.logback.classic.pattern.MessageConverter;
import ch.qos.logback.classic.spi.ILoggingEvent;
/**
* sensitive data log converter
*/
@ -69,7 +70,7 @@ public class SensitiveDataConverter extends MessageConverter {
*
* @param logMsg original log
*/
private String passwordHandler(Pattern pwdPattern, String logMsg) {
static String passwordHandler(Pattern pwdPattern, String logMsg) {
Matcher matcher = pwdPattern.matcher(logMsg);
@ -79,7 +80,7 @@ public class SensitiveDataConverter extends MessageConverter {
String password = matcher.group();
String maskPassword = SensitiveLogUtils.maskDataSourcePwd(password);
String maskPassword = StringUtils.repeat(STAR, StringUtils.length(password));
matcher.appendReplacement(sb, maskPassword);
}

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/CommonHostManager.java
View File

@ -18,7 +18,6 @@
package org.apache.dolphinscheduler.server.master.dispatch.host;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.HeartBeat;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.server.master.dispatch.context.ExecutionContext;
@ -26,6 +25,7 @@ import org.apache.dolphinscheduler.server.master.dispatch.enums.ExecutorType;
import org.apache.dolphinscheduler.server.master.dispatch.host.assign.HostWorker;
import org.apache.dolphinscheduler.server.master.registry.ServerNodeManager;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;

3
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/LowerWeightHostManager.java
View File

@ -18,7 +18,6 @@
package org.apache.dolphinscheduler.server.master.dispatch.host;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.HeartBeat;
import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.remote.utils.NamedThreadFactory;
@ -28,6 +27,8 @@ import org.apache.dolphinscheduler.server.master.dispatch.host.assign.HostWorker
import org.apache.dolphinscheduler.server.master.dispatch.host.assign.LowerWeightRoundRobin;
import org.apache.dolphinscheduler.spi.utils.StringUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/dispatch/host/assign/AbstractSelector.java
View File

@ -16,7 +16,7 @@
*/
package org.apache.dolphinscheduler.server.master.dispatch.host.assign;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.commons.collections.CollectionUtils;
import java.util.Collection;

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/WorkflowExecuteThread.java
View File

@ -43,7 +43,6 @@ import org.apache.dolphinscheduler.common.model.TaskNodeRelation;
import org.apache.dolphinscheduler.common.process.ProcessDag;
import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.NetUtils;
@ -69,6 +68,7 @@ import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import org.apache.dolphinscheduler.service.queue.PeerTaskInstancePriorityQueue;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;

12
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/BaseTaskProcessor.java
View File

@ -29,8 +29,6 @@ import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
import org.apache.dolphinscheduler.common.task.sqoop.SqoopParameters;
import org.apache.dolphinscheduler.common.task.sqoop.sources.SourceMysqlParameter;
import org.apache.dolphinscheduler.common.task.sqoop.targets.TargetMysqlParameter;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.EnumUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.TaskParametersUtils;
@ -51,6 +49,7 @@ import org.apache.dolphinscheduler.spi.task.request.SQLTaskExecutionContext;
import org.apache.dolphinscheduler.spi.task.request.SqoopTaskExecutionContext;
import org.apache.dolphinscheduler.spi.task.request.UdfFuncRequest;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.HashMap;
@ -63,6 +62,9 @@ import java.util.stream.Stream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Enums;
import com.google.common.base.Strings;
public abstract class BaseTaskProcessor implements ITaskProcessor {
protected Logger logger = LoggerFactory.getLogger(getClass());
@ -300,8 +302,8 @@ public abstract class BaseTaskProcessor implements ITaskProcessor {
sqlTaskExecutionContext.setDefaultFS(HadoopUtils.getInstance().getDefaultFS());
// whether udf type
boolean udfTypeFlag = EnumUtils.isValidEnum(UdfType.class, sqlParameters.getType())
&& !StringUtils.isEmpty(sqlParameters.getUdfs());
boolean udfTypeFlag = Enums.getIfPresent(UdfType.class, Strings.nullToEmpty(sqlParameters.getType())).isPresent()
&& !StringUtils.isEmpty(sqlParameters.getUdfs());
if (udfTypeFlag) {
String[] udfFunIds = sqlParameters.getUdfs().split(",");
@ -371,4 +373,4 @@ public abstract class BaseTaskProcessor implements ITaskProcessor {
return resourcesMap;
}
}
}

3
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/monitor/AbstractMonitor.java
View File

@ -16,8 +16,7 @@
*/
package org.apache.dolphinscheduler.server.monitor;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;

30
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/ArgsUtils.java
View File

@ -1,30 +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.dolphinscheduler.server.utils;
public class ArgsUtils {
private ArgsUtils() throws IllegalStateException {
throw new IllegalStateException("Utility class");
}
public static String escape(String arg) {
return arg.replace(" ", "\\ ").replace("\"", "\\\"").replace("'", "\\'");
}
}

5
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/utils/ProcessUtils.java
View File

@ -19,7 +19,6 @@ package org.apache.dolphinscheduler.server.utils;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.CommonUtils;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
@ -30,7 +29,9 @@ import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.service.log.LogClientService;
import org.apache.dolphinscheduler.service.queue.entity.TaskExecutionContext;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import java.io.File;
import java.nio.charset.StandardCharsets;
@ -153,7 +154,7 @@ public class ProcessUtils {
List<String> pidList = new ArrayList<>();
Matcher mat = null;
// pstree pid get sub pids
if (OSUtils.isMacOS()) {
if (SystemUtils.IS_OS_MAC) {
String pids = OSUtils.exeCmd(String.format("%s -sp %d", Constants.PSTREE, processId));
if (null != pids) {
mat = MACPATTERN.matcher(pids);

4
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java
View File

@ -148,11 +148,9 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
OSUtils.createUserIfAbsent(taskExecutionContext.getTenantCode());
}
} catch (Throwable ex) {
String errorLog = String.format("create execLocalPath : %s", execLocalPath);
LoggerUtils.logError(Optional.of(logger), errorLog, ex);
logger.error("create execLocalPath: {}", execLocalPath, ex);
TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
}
FileUtils.taskLoggerThreadLocal.remove();
taskCallbackService.addRemoteChannel(taskExecutionContext.getTaskInstanceId(),
new NettyRemoteChannel(channel, command.getOpaque()));

116
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/log/SensitiveDataConverterTest.java
View File

@ -17,23 +17,16 @@
package org.apache.dolphinscheduler.server.log;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.utils.SensitiveLogUtils;
import static org.apache.dolphinscheduler.server.log.SensitiveDataConverter.passwordHandler;
import org.apache.dolphinscheduler.common.Constants;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Marker;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.classic.spi.IThrowableProxy;
import ch.qos.logback.classic.spi.LoggerContextVO;
public class SensitiveDataConverterTest {
@ -58,86 +51,7 @@ public class SensitiveDataConverterTest {
@Test
public void convert() {
SensitiveDataConverter sensitiveDataConverter = new SensitiveDataConverter();
String result = sensitiveDataConverter.convert(new ILoggingEvent() {
@Override
public String getThreadName() {
return null;
}
@Override
public Level getLevel() {
return Level.INFO;
}
@Override
public String getMessage() {
return null;
}
@Override
public Object[] getArgumentArray() {
return new Object[0];
}
@Override
public String getFormattedMessage() {
return logMsg;
}
@Override
public String getLoggerName() {
return null;
}
@Override
public LoggerContextVO getLoggerContextVO() {
return null;
}
@Override
public IThrowableProxy getThrowableProxy() {
return null;
}
@Override
public StackTraceElement[] getCallerData() {
return new StackTraceElement[0];
}
@Override
public boolean hasCallerData() {
return false;
}
@Override
public Marker getMarker() {
return null;
}
@Override
public Map<String, String> getMDCPropertyMap() {
return null;
}
@Override
public Map<String, String> getMdc() {
return null;
}
@Override
public long getTimeStamp() {
return 0;
}
@Override
public void prepareForDeferredProcessing() {
}
});
Assert.assertNotEquals(maskLogMsg, passwordHandler(pwdPattern, logMsg));
}
/**
@ -153,28 +67,4 @@ public class SensitiveDataConverterTest {
}
/**
* password regex test
*
* @param logMsg original log
*/
private static String passwordHandler(Pattern pattern, String logMsg) {
Matcher matcher = pattern.matcher(logMsg);
StringBuffer sb = new StringBuffer(logMsg.length());
while (matcher.find()) {
String password = matcher.group();
String maskPassword = SensitiveLogUtils.maskDataSourcePwd(password);
matcher.appendReplacement(sb, maskPassword);
}
matcher.appendTail(sb);
return sb.toString();
}
}

10
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/utils/ProcessUtilsTest.java
View File

@ -25,6 +25,8 @@ import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.commons.lang.SystemUtils;
import java.util.ArrayList;
import java.util.List;
@ -36,11 +38,12 @@ import org.mockito.MockitoAnnotations;
import org.powermock.api.mockito.PowerMockito;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import org.powermock.reflect.Whitebox;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@RunWith(PowerMockRunner.class)
@PrepareForTest({System.class, OSUtils.class, HadoopUtils.class, PropertyUtils.class})
@PrepareForTest({System.class, OSUtils.class, HadoopUtils.class, PropertyUtils.class, SystemUtils.class})
public class ProcessUtilsTest {
private static final Logger logger = LoggerFactory.getLogger(ProcessUtils.class);
@ -53,11 +56,8 @@ public class ProcessUtilsTest {
@Test
public void getPidsStr() throws Exception {
int processId = 1;
String pidList = ProcessUtils.getPidsStr(processId);
Assert.assertNotEquals("The child process of process 1 should not be empty", pidList, "");
PowerMockito.mockStatic(OSUtils.class);
when(OSUtils.isMacOS()).thenReturn(true);
Whitebox.setInternalState(SystemUtils.class, "IS_OS_MAC", true);
when(OSUtils.exeCmd(String.format("%s -p %d", Constants.PSTREE, processId))).thenReturn(null);
String pidListMac = ProcessUtils.getPidsStr(processId);
Assert.assertEquals("", pidListMac);

7
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/log/LogClientService.java
View File

@ -149,7 +149,6 @@ public class LogClientService implements AutoCloseable {
public byte[] getLogBytes(String host, int port, String path) {
logger.info("log path {}", path);
GetLogBytesRequestCommand request = new GetLogBytesRequestCommand(path);
byte[] result = null;
final Host address = new Host(host, port);
try {
Command command = request.convert2Command();
@ -157,14 +156,14 @@ public class LogClientService implements AutoCloseable {
if (response != null) {
GetLogBytesResponseCommand getLog = JSONUtils.parseObject(
response.getBody(), GetLogBytesResponseCommand.class);
return getLog.getData();
return getLog.getData() == null ? new byte[0] : getLog.getData();
}
} catch (Exception e) {
logger.error("get log size error", e);
} finally {
this.client.closeChannel(address);
}
return result;
return new byte[0];
}
/**
@ -199,4 +198,4 @@ public class LogClientService implements AutoCloseable {
public boolean isRunning() {
return isRunning;
}
}
}

3
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/permission/PermissionCheck.java
View File

@ -20,11 +20,12 @@ package org.apache.dolphinscheduler.service.permission;
import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.service.exceptions.ServiceException;
import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.commons.collections.CollectionUtils;
import java.util.List;
import org.slf4j.Logger;

5
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/process/ProcessService.java
View File

@ -51,7 +51,6 @@ import org.apache.dolphinscheduler.common.process.ResourceInfo;
import org.apache.dolphinscheduler.common.task.AbstractParameters;
import org.apache.dolphinscheduler.common.task.TaskTimeoutParameter;
import org.apache.dolphinscheduler.common.task.subprocess.SubProcessParameters;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.ParameterUtils;
@ -106,6 +105,7 @@ import org.apache.dolphinscheduler.remote.utils.Host;
import org.apache.dolphinscheduler.service.log.LogClientService;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import java.util.ArrayList;
@ -2403,7 +2403,8 @@ public class ProcessService {
if (!processTaskRelationList.isEmpty()) {
Set<Integer> processTaskRelationSet = processTaskRelationList.stream().map(ProcessTaskRelation::hashCode).collect(toSet());
Set<Integer> taskRelationSet = taskRelationList.stream().map(ProcessTaskRelationLog::hashCode).collect(toSet());
if (CollectionUtils.isEqualCollection(processTaskRelationSet, taskRelationSet)) {
boolean result = CollectionUtils.isEqualCollection(processTaskRelationSet, taskRelationSet);
if (result) {
return Constants.EXIT_CODE_SUCCESS;
}
processTaskRelationMapper.deleteByCode(projectCode, processDefinitionCode);

3
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/quartz/cron/CronUtils.java
View File

@ -28,10 +28,11 @@ import static com.cronutils.model.CronType.QUARTZ;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CycleEnum;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.Schedule;
import org.apache.commons.collections.CollectionUtils;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Calendar;