diff --git a/dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/Constants.java b/dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/Constants.java
index 55c0f9ffbf601f0dd0e068d70cee49e76a03a2a2..94d95b3c26de46ca2bd521578956d29ae37ba620 100644
--- a/dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/Constants.java
+++ b/dolphinscheduler-alert/src/main/java/org/apache/dolphinscheduler/alert/utils/Constants.java
@@ -20,7 +20,9 @@ package org.apache.dolphinscheduler.alert.utils;
* constants
*/
public class Constants {
-
+ private Constants() {
+ throw new IllegalStateException("Constants class");
+ }
/**
* alert properties path
*/
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java
index 29a16447e18f4afcf870f985c8ea6cd0d8c90874..6438e206f8ed1e1bcb55e4b95a810276e1b8398b 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java
@@ -119,7 +119,7 @@ public class ResourcesService extends BaseService {
putMsg(result, Status.UDF_RESOURCE_SUFFIX_NOT_JAR);
return result;
}
- if (file.getSize() > Constants.maxFileSize) {
+ if (file.getSize() > Constants.MAX_FILE_SIZE) {
logger.error("file size is too large: {}", file.getOriginalFilename());
putMsg(result, Status.RESOURCE_SIZE_EXCEED_LIMIT);
return result;
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/CheckUtils.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/CheckUtils.java
index 7099378b1d3fedc6bc051c76c3d127ce254a3e62..c5c702404da2f46c4e76e09f986a2bb3aec47497 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/CheckUtils.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/utils/CheckUtils.java
@@ -35,7 +35,9 @@ import java.util.regex.Pattern;
*/
public class CheckUtils {
-
+ private CheckUtils() {
+ throw new IllegalStateException("CheckUtils class");
+ }
/**
* check username
*
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/CheckUtilsTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/CheckUtilsTest.java
index b0b4319fb4d6d3957f93ec77c20dd0af71723e74..24a0ed31d65b54b91cd27416ab9fcd8f9f524567 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/CheckUtilsTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/utils/CheckUtilsTest.java
@@ -17,8 +17,26 @@
package org.apache.dolphinscheduler.api.utils;
+import org.apache.commons.lang.StringUtils;
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.ProgramType;
+import org.apache.dolphinscheduler.common.enums.TaskType;
+import org.apache.dolphinscheduler.common.process.ResourceInfo;
+import org.apache.dolphinscheduler.common.task.AbstractParameters;
+import org.apache.dolphinscheduler.common.task.datax.DataxParameters;
+import org.apache.dolphinscheduler.common.task.dependent.DependentParameters;
+import org.apache.dolphinscheduler.common.task.flink.FlinkParameters;
+import org.apache.dolphinscheduler.common.task.http.HttpParameters;
+import org.apache.dolphinscheduler.common.task.mr.MapreduceParameters;
+import org.apache.dolphinscheduler.common.task.procedure.ProcedureParameters;
+import org.apache.dolphinscheduler.common.task.python.PythonParameters;
+import org.apache.dolphinscheduler.common.task.shell.ShellParameters;
+import org.apache.dolphinscheduler.common.task.spark.SparkParameters;
+import org.apache.dolphinscheduler.common.task.sql.SqlParameters;
+import org.apache.dolphinscheduler.common.task.subprocess.SubProcessParameters;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.common.utils.TaskParametersUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -78,6 +96,14 @@ public class CheckUtilsTest {
}
+ @Test
+ public void testCheckOtherParams() {
+ assertFalse(CheckUtils.checkOtherParams(null));
+ assertFalse(CheckUtils.checkOtherParams(""));
+ assertTrue(CheckUtils.checkOtherParams("xxx"));
+ assertFalse(CheckUtils.checkOtherParams("{}"));
+ assertFalse(CheckUtils.checkOtherParams("{\"key1\":111}"));
+ }
/**
* check passwd
*/
@@ -106,5 +132,90 @@ public class CheckUtilsTest {
assertTrue(CheckUtils.checkPhone("17362537263"));
}
+ @Test
+ public void testCheckTaskNodeParameters() {
+
+ assertFalse(CheckUtils.checkTaskNodeParameters(null,null));
+ assertFalse(CheckUtils.checkTaskNodeParameters(null,"unKnown"));
+ assertFalse(CheckUtils.checkTaskNodeParameters("unKnown","unKnown"));
+ assertFalse(CheckUtils.checkTaskNodeParameters("unKnown",null));
+
+ // sub SubProcessParameters
+ SubProcessParameters subProcessParameters = new SubProcessParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(subProcessParameters), TaskType.SUB_PROCESS.toString()));
+ subProcessParameters.setProcessDefinitionId(1234);
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(subProcessParameters), TaskType.SUB_PROCESS.toString()));
+
+ // ShellParameters
+ ShellParameters shellParameters = new ShellParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(shellParameters), TaskType.SHELL.toString()));
+ shellParameters.setRawScript("");
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(shellParameters), TaskType.SHELL.toString()));
+ shellParameters.setRawScript("sss");
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(shellParameters), TaskType.SHELL.toString()));
+
+ // ProcedureParameters
+ ProcedureParameters procedureParameters = new ProcedureParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(procedureParameters), TaskType.PROCEDURE.toString()));
+ procedureParameters.setDatasource(1);
+ procedureParameters.setType("xx");
+ procedureParameters.setMethod("yy");
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(procedureParameters), TaskType.PROCEDURE.toString()));
+
+ // SqlParameters
+ SqlParameters sqlParameters = new SqlParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(sqlParameters), TaskType.SQL.toString()));
+ sqlParameters.setDatasource(1);
+ sqlParameters.setType("xx");
+ sqlParameters.setSql("yy");
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(sqlParameters), TaskType.SQL.toString()));
+
+ // MapreduceParameters
+ MapreduceParameters mapreduceParameters = new MapreduceParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(mapreduceParameters), TaskType.MR.toString()));
+ mapreduceParameters.setMainJar(new ResourceInfo());
+ mapreduceParameters.setProgramType(ProgramType.JAVA);
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(mapreduceParameters), TaskType.MR.toString()));
+
+ // SparkParameters
+ SparkParameters sparkParameters = new SparkParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(sparkParameters), TaskType.SPARK.toString()));
+ sparkParameters.setMainJar(new ResourceInfo());
+ sparkParameters.setProgramType(ProgramType.SCALA);
+ sparkParameters.setSparkVersion("1.1.1");
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(sparkParameters), TaskType.SPARK.toString()));
+
+ // PythonParameters
+ PythonParameters pythonParameters = new PythonParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(pythonParameters), TaskType.PYTHON.toString()));
+ pythonParameters.setRawScript("ss");
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(pythonParameters), TaskType.PYTHON.toString()));
+
+ // DependentParameters
+ DependentParameters dependentParameters = new DependentParameters();
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(dependentParameters), TaskType.DEPENDENT.toString()));
+
+ // FlinkParameters
+ FlinkParameters flinkParameters = new FlinkParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(flinkParameters), TaskType.FLINK.toString()));
+ flinkParameters.setMainJar(new ResourceInfo());
+ flinkParameters.setProgramType(ProgramType.JAVA);
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(flinkParameters), TaskType.FLINK.toString()));
+
+ // HTTP
+ HttpParameters httpParameters = new HttpParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(httpParameters), TaskType.HTTP.toString()));
+ httpParameters.setUrl("httpUrl");
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(httpParameters), TaskType.HTTP.toString()));
+
+ // DataxParameters
+ DataxParameters dataxParameters = new DataxParameters();
+ assertFalse(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(dataxParameters), TaskType.DATAX.toString()));
+ dataxParameters.setDataSource(111);
+ dataxParameters.setDataTarget(333);
+ dataxParameters.setSql("sql");
+ dataxParameters.setTargetTable("tar");
+ assertTrue(CheckUtils.checkTaskNodeParameters(JSONUtils.toJsonString(dataxParameters), TaskType.DATAX.toString()));
+ }
}
\ No newline at end of file
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
index 9cb1a5821e2e9fcb035bc453e6045399ded37111..73125f49266521c17cbcf6f7a8fe83b1562157db 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
@@ -25,7 +25,9 @@ import java.util.regex.Pattern;
* Constants
*/
public final class Constants {
-
+ private Constants() {
+ throw new IllegalStateException("Constants class");
+ }
/**
* common properties path
*/
@@ -124,49 +126,41 @@ public final class Constants {
/**
* MasterServer directory registered in zookeeper
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_MASTERS = "zookeeper.dolphinscheduler.masters";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_MASTERS = "/masters";
/**
* WorkerServer directory registered in zookeeper
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_WORKERS = "zookeeper.dolphinscheduler.workers";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_WORKERS = "/workers";
/**
* all servers directory registered in zookeeper
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_DEAD_SERVERS = "zookeeper.dolphinscheduler.dead.servers";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_DEAD_SERVERS = "/dead-servers";
/**
* MasterServer lock directory registered in zookeeper
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_MASTERS = "zookeeper.dolphinscheduler.lock.masters";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_MASTERS = "/lock/masters";
/**
* WorkerServer lock directory registered in zookeeper
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_WORKERS = "zookeeper.dolphinscheduler.lock.workers";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_WORKERS = "/lock/workers";
/**
* MasterServer failover directory registered in zookeeper
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_MASTERS = "zookeeper.dolphinscheduler.lock.failover.masters";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_MASTERS = "/lock/failover/masters";
/**
* WorkerServer failover directory registered in zookeeper
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_WORKERS = "zookeeper.dolphinscheduler.lock.failover.workers";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_WORKERS = "/lock/failover/workers";
/**
* MasterServer startup failover runing and fault tolerance process
*/
- //public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_STARTUP_MASTERS = "zookeeper.dolphinscheduler.lock.failover.startup.masters";
public static final String ZOOKEEPER_DOLPHINSCHEDULER_LOCK_FAILOVER_STARTUP_MASTERS = "/lock/failover/startup-masters";
/**
@@ -354,87 +348,87 @@ public final class Constants {
/**
* heartbeat threads number
*/
- public static final int defaulWorkerHeartbeatThreadNum = 1;
+ public static final int DEFAUL_WORKER_HEARTBEAT_THREAD_NUM = 1;
/**
* heartbeat interval
*/
- public static final int defaultWorkerHeartbeatInterval = 60;
+ public static final int DEFAULT_WORKER_HEARTBEAT_INTERVAL = 60;
/**
* worker fetch task number
*/
- public static final int defaultWorkerFetchTaskNum = 1;
+ public static final int DEFAULT_WORKER_FETCH_TASK_NUM = 1;
/**
* worker execute threads number
*/
- public static final int defaultWorkerExecThreadNum = 10;
+ public static final int DEFAULT_WORKER_EXEC_THREAD_NUM = 10;
/**
* master cpu load
*/
- public static final int defaultMasterCpuLoad = Runtime.getRuntime().availableProcessors() * 2;
+ public static final int DEFAULT_MASTER_CPU_LOAD = Runtime.getRuntime().availableProcessors() * 2;
/**
* master reserved memory
*/
- public static final double defaultMasterReservedMemory = OSUtils.totalMemorySize() / 10;
+ public static final double DEFAULT_MASTER_RESERVED_MEMORY = OSUtils.totalMemorySize() / 10;
/**
* worker cpu load
*/
- public static final int defaultWorkerCpuLoad = Runtime.getRuntime().availableProcessors() * 2;
+ public static final int DEFAULT_WORKER_CPU_LOAD = Runtime.getRuntime().availableProcessors() * 2;
/**
* worker reserved memory
*/
- public static final double defaultWorkerReservedMemory = OSUtils.totalMemorySize() / 10;
+ public static final double DEFAULT_WORKER_RESERVED_MEMORY = OSUtils.totalMemorySize() / 10;
/**
* master execute threads number
*/
- public static final int defaultMasterExecThreadNum = 100;
+ public static final int DEFAULT_MASTER_EXEC_THREAD_NUM = 100;
/**
* default master concurrent task execute num
*/
- public static final int defaultMasterTaskExecNum = 20;
+ public static final int DEFAULT_MASTER_TASK_EXEC_NUM = 20;
/**
* default log cache rows num,output when reach the number
*/
- public static final int defaultLogRowsNum = 4 * 16;
+ public static final int DEFAULT_LOG_ROWS_NUM = 4 * 16;
/**
* log flush interval,output when reach the interval
*/
- public static final int defaultLogFlushInterval = 1000;
+ public static final int DEFAULT_LOG_FLUSH_INTERVAL = 1000;
/**
* default master heartbeat thread number
*/
- public static final int defaulMasterHeartbeatThreadNum = 1;
+ public static final int DEFAULT_MASTER_HEARTBEAT_THREAD_NUM = 1;
/**
* default master heartbeat interval
*/
- public static final int defaultMasterHeartbeatInterval = 60;
+ public static final int DEFAULT_MASTER_HEARTBEAT_INTERVAL = 60;
/**
* default master commit retry times
*/
- public static final int defaultMasterCommitRetryTimes = 5;
+ public static final int DEFAULT_MASTER_COMMIT_RETRY_TIMES = 5;
/**
* default master commit retry interval
*/
- public static final int defaultMasterCommitRetryInterval = 3000;
+ public static final int DEFAULT_MASTER_COMMIT_RETRY_INTERVAL = 3000;
/**
* time unit secong to minutes
@@ -474,9 +468,9 @@ public final class Constants {
public static final String THREAD_NAME_MASTER_SERVER = "Master-Server";
public static final String THREAD_NAME_WORKER_SERVER = "Worker-Server";
- public static String TASK_RECORD_TABLE_HIVE_LOG = "eamp_hive_log_hd";
+ public static final String TASK_RECORD_TABLE_HIVE_LOG = "eamp_hive_log_hd";
- public static String TASK_RECORD_TABLE_HISTORY_HIVE_LOG = "eamp_hive_hist_log_hd";
+ public static final String TASK_RECORD_TABLE_HISTORY_HIVE_LOG = "eamp_hive_hist_log_hd";
/**
@@ -874,7 +868,7 @@ public final class Constants {
public static final String FLINK_JOB_MANAGE_MEM = "-yjm";
public static final String FLINK_TASK_MANAGE_MEM = "-ytm";
- public static final String FLINK_detach = "-d";
+ public static final String FLINK_DETACH = "-d";
public static final String FLINK_MAIN_CLASS = "-c";
@@ -989,7 +983,7 @@ public final class Constants {
* session timeout
*/
public static final int SESSION_TIME_OUT = 7200;
- public static final int maxFileSize = 1024 * 1024 * 1024;
+ public static final int MAX_FILE_SIZE = 1024 * 1024 * 1024;
public static final String UDF = "UDF";
public static final String CLASS = "class";
public static final String RECEIVERS = "receivers";
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/job/db/ClickHouseDataSource.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/job/db/ClickHouseDataSource.java
index fe76497ff87f913afd00338dd59f19b10b2e27e5..48550c31cc7113fab9ae58844ce2a4ec8bede5e0 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/job/db/ClickHouseDataSource.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/job/db/ClickHouseDataSource.java
@@ -38,7 +38,7 @@ public class ClickHouseDataSource extends BaseDataSource {
@Override
public String getJdbcUrl() {
String jdbcUrl = getAddress();
- if (jdbcUrl.lastIndexOf("/") != (jdbcUrl.length() - 1)) {
+ if (jdbcUrl.lastIndexOf('/') != (jdbcUrl.length() - 1)) {
jdbcUrl += "/";
}
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/subprocess/SubProcessParameters.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/subprocess/SubProcessParameters.java
index 21e3ce224840909342c42f95ca6c98f0c4cad7f0..c7784de8ddc997d056c651ac40df1c3e2d9eb624 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/subprocess/SubProcessParameters.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/task/subprocess/SubProcessParameters.java
@@ -38,7 +38,7 @@ public class SubProcessParameters extends AbstractParameters {
@Override
public boolean checkParameters() {
- return this.processDefinitionId != 0;
+ return this.processDefinitionId != null && this.processDefinitionId != 0;
}
@Override
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/CollectionUtils.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/CollectionUtils.java
index 9c02111c3695c33710378ced75051e423cc9821d..22c58640ccd314f6bbbe6d050fa2009560a997a8 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/CollectionUtils.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/CollectionUtils.java
@@ -37,6 +37,9 @@ import java.util.*;
*/
public class CollectionUtils {
+ private CollectionUtils() {
+ throw new IllegalStateException("CollectionUtils class");
+ }
/**
* Returns a new {@link Collection} containing a minus a subset of
* b. Only the elements of b that satisfy the predicate
@@ -139,26 +142,6 @@ public class CollectionUtils {
cardinalityB = CollectionUtils.getCardinalityMap(b);
}
- /**
- * Returns the maximum frequency of an object.
- *
- * @param obj the object
- * @return the maximum frequency of the object
- */
- private int max(final Object obj) {
- return Math.max(freqA(obj), freqB(obj));
- }
-
- /**
- * Returns the minimum frequency of an object.
- *
- * @param obj the object
- * @return the minimum frequency of the object
- */
- private int min(final Object obj) {
- return Math.min(freqA(obj), freqB(obj));
- }
-
/**
* Returns the frequency of this object in collection A.
*
@@ -225,7 +208,7 @@ public class CollectionUtils {
if (a.size() != b.size()) {
return false;
}
- final CardinalityHelper