[cherry-pick-2.0.3]Cherry pick to 2.0.3 (#8041)

* [cherry-pick-2.0.3]Correct typo in English README #7695

* Update NOTICE year (#7784)

* [BUG] Repair JDBC connection of Oracle (#7883)

* [cherry-pick-2.0.3][Fix-7538] [server] Fix when there is a forbidden node in dag, the execution flow is abnormal #7613

* [cherry-pick-2.0.3][Fix-7825] Remedy the value of create time and update time to be current time when importing a process json file. #7828

* [FIX-7732][fix] fix column 'is_directory' of table `t_ds_resources` type error in PG database (#7898)

Fix column 'is_directory' of table t_ds_resources type error in PG database
This closes #7732

Co-authored-by: Brennan Fox <brnnnfx@users.noreply.github.com>
Co-authored-by: Kirs <acm_master@163.com>
Co-authored-by: X&Z <980813351@qq.com>
Co-authored-by: caishunfeng <534328519@qq.com>
Co-authored-by: 天仇 <532066967@qq.com>
This commit is contained in:
wind 2022-01-14 14:24:49 +08:00 committed by GitHub
parent 600ee5e77c
commit 1f0aeabb92
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 238 additions and 228 deletions

2
NOTICE
View File

@ -1,5 +1,5 @@
Apache DolphinScheduler Apache DolphinScheduler
Copyright 2019-2021 The Apache Software Foundation Copyright 2019-2022 The Apache Software Foundation
This product includes software developed at This product includes software developed at
The Apache Software Foundation (http://www.apache.org/). The Apache Software Foundation (http://www.apache.org/).

View File

@ -41,7 +41,7 @@ Its main objectives are as follows:
## What's in DolphinScheduler ## What's in DolphinScheduler
Stability | Accessibility | Features | Scalability | Stability | Accessibility | Features | Scalability |
-- | -- | -- | -- --------- | ------------- | -------- | ------------|
Decentralized multi-master and multi-worker | Visualization of workflow key information, such as task status, task type, retry times, task operation machine information, visual variables, and so on at a glance.  |  Support pause, recover operation | Support customized task types Decentralized multi-master and multi-worker | Visualization of workflow key information, such as task status, task type, retry times, task operation machine information, visual variables, and so on at a glance.  |  Support pause, recover operation | Support customized task types
support HA | Visualization of all workflow operations, dragging tasks to draw DAGs, configuring data sources and resources. At the same time, for third-party systems, provide API mode operations. | Users on DolphinScheduler can achieve many-to-one or one-to-one mapping relationship through tenants and Hadoop users, which is very important for scheduling large data jobs. | The scheduler supports distributed scheduling, and the overall scheduling capability will increase linearly with the scale of the cluster. Master and Worker support dynamic adjustment. support HA | Visualization of all workflow operations, dragging tasks to draw DAGs, configuring data sources and resources. At the same time, for third-party systems, provide API mode operations. | Users on DolphinScheduler can achieve many-to-one or one-to-one mapping relationship through tenants and Hadoop users, which is very important for scheduling large data jobs. | The scheduler supports distributed scheduling, and the overall scheduling capability will increase linearly with the scale of the cluster. Master and Worker support dynamic adjustment.
Overload processing: By using the task queue mechanism, the number of schedulable tasks on a single machine can be flexibly configured. Machine jam can be avoided with high tolerance to numbers of tasks cached in task queue. | One-click deployment | Support traditional shell tasks, and big data platform task scheduling: MR, Spark, SQL (MySQL, PostgreSQL, hive, spark SQL), Python, Procedure, Sub_Process | | Overload processing: By using the task queue mechanism, the number of schedulable tasks on a single machine can be flexibly configured. Machine jam can be avoided with high tolerance to numbers of tasks cached in task queue. | One-click deployment | Support traditional shell tasks, and big data platform task scheduling: MR, Spark, SQL (MySQL, PostgreSQL, hive, spark SQL), Python, Procedure, Sub_Process | |
@ -59,11 +59,11 @@ Overload processing: By using the task queue mechanism, the number of schedulabl
## QuickStart in Docker ## QuickStart in Docker
Please referer the official website document: [QuickStart in Docker](https://dolphinscheduler.apache.org/en-us/docs/latest/user_doc/docker-deployment.html) Please refer the official website document: [QuickStart in Docker](https://dolphinscheduler.apache.org/en-us/docs/latest/user_doc/guide/installation/docker.html)
## QuickStart in Kubernetes ## QuickStart in Kubernetes
Please referer the official website document: [QuickStart in Kubernetes](https://dolphinscheduler.apache.org/en-us/docs/latest/user_doc/kubernetes-deployment.html) Please refer to the official website document: [QuickStart in Kubernetes](https://dolphinscheduler.apache.org/en-us/docs/latest/user_doc/guide/installation/kubernetes.html)
## How to Build ## How to Build
@ -95,8 +95,8 @@ You are very welcome to communicate with the developers and users of Dolphin Sch
2. Follow the [Twitter account of DolphinScheduler](https://twitter.com/dolphinschedule) and get the latest news on time. 2. Follow the [Twitter account of DolphinScheduler](https://twitter.com/dolphinschedule) and get the latest news on time.
### Contributor over time ### Contributor over time
[![Contributor over time](https://contributor-graph-api.apiseven.com/contributors-svg?chart=contributorOverTime&repo=apache/dolphinscheduler)](https://www.apiseven.com/en/contributor-graph?chart=contributorOverTime&repo=apache/dolphinscheduler) [![Contributor over time](https://contributor-graph-api.apiseven.com/contributors-svg?chart=contributorOverTime&repo=apache/dolphinscheduler)](https://www.apiseven.com/en/contributor-graph?chart=contributorOverTime&repo=apache/dolphinscheduler)
## How to Contribute ## How to Contribute

View File

@ -951,6 +951,8 @@ public class ProcessDefinitionServiceImpl extends BaseServiceImpl implements Pro
} }
processDefinition.setLocations(newArrayNode.toString()); processDefinition.setLocations(newArrayNode.toString());
} }
processDefinition.setCreateTime(new Date());
processDefinition.setUpdateTime(new Date());
Map<String, Object> createDagResult = createDagDefine(loginUser, taskRelationLogList, processDefinition, Lists.newArrayList()); Map<String, Object> createDagResult = createDagDefine(loginUser, taskRelationLogList, processDefinition, Lists.newArrayList());
if (Status.SUCCESS.equals(createDagResult.get(Constants.STATUS))) { if (Status.SUCCESS.equals(createDagResult.get(Constants.STATUS))) {
putMsg(createDagResult, Status.SUCCESS); putMsg(createDagResult, Status.SUCCESS);

View File

@ -28,219 +28,208 @@ import com.fasterxml.jackson.annotation.JsonFormat;
@TableName("t_ds_resources") @TableName("t_ds_resources")
public class Resource { public class Resource {
/** /**
* id * id
*/ */
@TableId(value="id", type=IdType.AUTO) @TableId(value = "id", type = IdType.AUTO)
private int id; private int id;
/** /**
* parent id * parent id
*/ */
private int pid; private int pid;
/** /**
* resource alias * resource alias
*/ */
private String alias; private String alias;
/** /**
* full name * full name
*/ */
private String fullName; private String fullName;
/** /**
* is directory * is directory
*/ */
private boolean isDirectory=false; private boolean isDirectory = false;
/** /**
* description * description
*/ */
private String description; private String description;
/** /**
* file alias * file alias
*/ */
private String fileName; private String fileName;
/** /**
* user id * user id
*/ */
private int userId; private int userId;
/** /**
* resource type * resource type
*/ */
private ResourceType type; private ResourceType type;
/** /**
* resource size * resource size
*/ */
private long size; private long size;
/** /**
* create time * create time
*/ */
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8") @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date createTime; private Date createTime;
/** /**
* update time * update time
*/ */
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone="GMT+8") @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
private Date updateTime; private Date updateTime;
public Resource() { public Resource() {
} }
public Resource(int id, String alias, String fileName, String description, int userId, public Resource(int id, String alias, String fileName, String description, int userId,
ResourceType type, long size, ResourceType type, long size,
Date createTime, Date updateTime) { Date createTime, Date updateTime) {
this.id = id; this.id = id;
this.alias = alias; this.alias = alias;
this.fileName = fileName; this.fileName = fileName;
this.description = description; this.description = description;
this.userId = userId; this.userId = userId;
this.type = type; this.type = type;
this.size = size; this.size = size;
this.createTime = createTime; this.createTime = createTime;
this.updateTime = updateTime; this.updateTime = updateTime;
} }
public Resource(int id, int pid, String alias, String fullName, boolean isDirectory) { public Resource(int id, int pid, String alias, String fullName, boolean isDirectory) {
this.id = id; this.id = id;
this.pid = pid; this.pid = pid;
this.alias = alias; this.alias = alias;
this.fullName = fullName; this.fullName = fullName;
this.isDirectory = isDirectory; this.isDirectory = isDirectory;
} }
/*public Resource(String alias, String fileName, String description, int userId, ResourceType type, long size, Date createTime, Date updateTime) { public Resource(int pid, String alias, String fullName, boolean isDirectory, String description, String fileName, int userId, ResourceType type, long size, Date createTime, Date updateTime) {
this.alias = alias; this.pid = pid;
this.fileName = fileName; this.alias = alias;
this.description = description; this.fullName = fullName;
this.userId = userId; this.isDirectory = isDirectory;
this.type = type; this.description = description;
this.size = size; this.fileName = fileName;
this.createTime = createTime; this.userId = userId;
this.updateTime = updateTime; this.type = type;
}*/ this.size = size;
this.createTime = createTime;
this.updateTime = updateTime;
}
public Resource(int pid, String alias, String fullName, boolean isDirectory, String description, String fileName, int userId, ResourceType type, long size, Date createTime, Date updateTime) { public int getId() {
this.pid = pid; return id;
this.alias = alias; }
this.fullName = fullName;
this.isDirectory = isDirectory;
this.description = description;
this.fileName = fileName;
this.userId = userId;
this.type = type;
this.size = size;
this.createTime = createTime;
this.updateTime = updateTime;
}
public int getId() { public void setId(int id) {
return id; this.id = id;
} }
public void setId(int id) { public String getAlias() {
this.id = id; return alias;
} }
public String getAlias() { public void setAlias(String alias) {
return alias; this.alias = alias;
} }
public void setAlias(String alias) { public int getPid() {
this.alias = alias; return pid;
} }
public int getPid() { public void setPid(int pid) {
return pid; this.pid = pid;
} }
public void setPid(int pid) { public String getFullName() {
this.pid = pid; return fullName;
} }
public String getFullName() { public void setFullName(String fullName) {
return fullName; this.fullName = fullName;
} }
public void setFullName(String fullName) { public boolean isDirectory() {
this.fullName = fullName; return isDirectory;
} }
public boolean isDirectory() { public void setDirectory(boolean directory) {
return isDirectory; isDirectory = directory;
} }
public void setDirectory(boolean directory) { public String getFileName() {
isDirectory = directory; return fileName;
} }
public String getFileName() { public void setFileName(String fileName) {
return fileName; this.fileName = fileName;
} }
public void setFileName(String fileName) { public String getDescription() {
this.fileName = fileName; return description;
} }
public String getDescription() { public void setDescription(String description) {
return description; this.description = description;
} }
public void setDescription(String description) { public int getUserId() {
this.description = description; return userId;
} }
public int getUserId() { public void setUserId(int userId) {
return userId; this.userId = userId;
} }
public void setUserId(int userId) {
this.userId = userId;
}
public ResourceType getType() { public ResourceType getType() {
return type; return type;
} }
public void setType(ResourceType type) { public void setType(ResourceType type) {
this.type = type; this.type = type;
} }
public long getSize() { public long getSize() {
return size; return size;
} }
public void setSize(long size) { public void setSize(long size) {
this.size = size; this.size = size;
} }
public Date getCreateTime() { public Date getCreateTime() {
return createTime; return createTime;
} }
public void setCreateTime(Date createTime) { public void setCreateTime(Date createTime) {
this.createTime = createTime; this.createTime = createTime;
} }
public Date getUpdateTime() { public Date getUpdateTime() {
return updateTime; return updateTime;
} }
public void setUpdateTime(Date updateTime) { public void setUpdateTime(Date updateTime) {
this.updateTime = updateTime; this.updateTime = updateTime;
} }
@Override @Override
public String toString() { public String toString() {
return "Resource{" + return "Resource{" +
"id=" + id + "id=" + id +
", pid=" + pid + ", pid=" + pid +
", alias='" + alias + '\'' + ", alias='" + alias + '\'' +
@ -254,30 +243,30 @@ public class Resource {
", createTime=" + createTime + ", createTime=" + createTime +
", updateTime=" + updateTime + ", updateTime=" + updateTime +
'}'; '}';
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
} }
Resource resource = (Resource) o; @Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
Resource resource = (Resource) o;
if (id != resource.id) {
return false;
}
return alias.equals(resource.alias);
if (id != resource.id) {
return false;
} }
return alias.equals(resource.alias);
} @Override
public int hashCode() {
@Override int result = id;
public int hashCode() { result = 31 * result + alias.hashCode();
int result = id; return result;
result = 31 * result + alias.hashCode(); }
return result;
}
} }

View File

@ -640,7 +640,7 @@ CREATE TABLE t_ds_resources (
update_time timestamp DEFAULT NULL , update_time timestamp DEFAULT NULL ,
pid int, pid int,
full_name varchar(64), full_name varchar(64),
is_directory int, is_directory boolean DEFAULT FALSE,
PRIMARY KEY (id), PRIMARY KEY (id),
CONSTRAINT t_ds_resources_un UNIQUE (full_name, type) CONSTRAINT t_ds_resources_un UNIQUE (full_name, type)
) ; ) ;

View File

@ -65,14 +65,16 @@ public class OracleDatasourceProcessor extends AbstractDatasourceProcessor {
public BaseConnectionParam createConnectionParams(BaseDataSourceParamDTO datasourceParam) { public BaseConnectionParam createConnectionParams(BaseDataSourceParamDTO datasourceParam) {
OracleDatasourceParamDTO oracleParam = (OracleDatasourceParamDTO) datasourceParam; OracleDatasourceParamDTO oracleParam = (OracleDatasourceParamDTO) datasourceParam;
String address; String address;
String jdbcUrl;
if (DbConnectType.ORACLE_SID.equals(oracleParam.getConnectType())) { if (DbConnectType.ORACLE_SID.equals(oracleParam.getConnectType())) {
address = String.format("%s%s:%s", address = String.format("%s%s:%s",
Constants.JDBC_ORACLE_SID, oracleParam.getHost(), oracleParam.getPort()); Constants.JDBC_ORACLE_SID, oracleParam.getHost(), oracleParam.getPort());
jdbcUrl = address + ":" + oracleParam.getDatabase();
} else { } else {
address = String.format("%s%s:%s", address = String.format("%s%s:%s",
Constants.JDBC_ORACLE_SERVICE_NAME, oracleParam.getHost(), oracleParam.getPort()); Constants.JDBC_ORACLE_SERVICE_NAME, oracleParam.getHost(), oracleParam.getPort());
jdbcUrl = address + "/" + oracleParam.getDatabase();
} }
String jdbcUrl = address + "/" + oracleParam.getDatabase();
OracleConnectionParam oracleConnectionParam = new OracleConnectionParam(); OracleConnectionParam oracleConnectionParam = new OracleConnectionParam();
oracleConnectionParam.setUser(oracleParam.getUserName()); oracleConnectionParam.setUser(oracleParam.getUserName());

View File

@ -61,13 +61,13 @@ public class OracleDatasourceProcessorTest {
.createConnectionParams(oracleDatasourceParamDTO); .createConnectionParams(oracleDatasourceParamDTO);
Assert.assertNotNull(connectionParams); Assert.assertNotNull(connectionParams);
Assert.assertEquals("jdbc:oracle:thin:@localhost:3308", connectionParams.getAddress()); Assert.assertEquals("jdbc:oracle:thin:@localhost:3308", connectionParams.getAddress());
Assert.assertEquals("jdbc:oracle:thin:@localhost:3308/default", connectionParams.getJdbcUrl()); Assert.assertEquals("jdbc:oracle:thin:@localhost:3308:default", connectionParams.getJdbcUrl());
} }
@Test @Test
public void testCreateConnectionParams2() { public void testCreateConnectionParams2() {
String connectionJson = "{\"user\":\"root\",\"password\":\"123456\",\"address\":\"jdbc:oracle:thin:@localhost:3308\"" String connectionJson = "{\"user\":\"root\",\"password\":\"123456\",\"address\":\"jdbc:oracle:thin:@localhost:3308\""
+ ",\"database\":\"default\",\"jdbcUrl\":\"jdbc:oracle:thin:@localhost:3308/default\",\"connectType\":\"ORACLE_SID\"}"; + ",\"database\":\"default\",\"jdbcUrl\":\"jdbc:oracle:thin:@localhost:3308:default\",\"connectType\":\"ORACLE_SID\"}";
OracleConnectionParam connectionParams = (OracleConnectionParam) oracleDatasourceProcessor OracleConnectionParam connectionParams = (OracleConnectionParam) oracleDatasourceProcessor
.createConnectionParams(connectionJson); .createConnectionParams(connectionJson);
Assert.assertNotNull(connectionParams); Assert.assertNotNull(connectionParams);
@ -82,9 +82,9 @@ public class OracleDatasourceProcessorTest {
@Test @Test
public void testGetJdbcUrl() { public void testGetJdbcUrl() {
OracleConnectionParam oracleConnectionParam = new OracleConnectionParam(); OracleConnectionParam oracleConnectionParam = new OracleConnectionParam();
oracleConnectionParam.setJdbcUrl("jdbc:oracle:thin:@localhost:3308/default"); oracleConnectionParam.setJdbcUrl("jdbc:oracle:thin:@localhost:3308:default");
oracleConnectionParam.setOther("other=other"); oracleConnectionParam.setOther("other=other");
Assert.assertEquals("jdbc:oracle:thin:@localhost:3308/default?other=other", Assert.assertEquals("jdbc:oracle:thin:@localhost:3308:default?other=other",
oracleDatasourceProcessor.getJdbcUrl(oracleConnectionParam)); oracleDatasourceProcessor.getJdbcUrl(oracleConnectionParam));
} }
@ -97,4 +97,4 @@ public class OracleDatasourceProcessorTest {
public void testGetValidationQuery() { public void testGetValidationQuery() {
Assert.assertEquals(Constants.ORACLE_VALIDATION_QUERY, oracleDatasourceProcessor.getValidationQuery()); Assert.assertEquals(Constants.ORACLE_VALIDATION_QUERY, oracleDatasourceProcessor.getValidationQuery());
} }
} }

View File

@ -1,5 +1,5 @@
Apache DolphinScheduler Apache DolphinScheduler
Copyright 2019-2021 The Apache Software Foundation Copyright 2019-2022 The Apache Software Foundation
This product includes software developed at This product includes software developed at
The Apache Software Foundation (http://www.apache.org/). The Apache Software Foundation (http://www.apache.org/).

View File

@ -920,33 +920,50 @@ public class WorkflowExecuteThread implements Runnable {
return DependResult.SUCCESS; return DependResult.SUCCESS;
} }
TaskNode taskNode = dag.getNode(taskCode); TaskNode taskNode = dag.getNode(taskCode);
List<String> depCodeList = taskNode.getDepList(); List<String> indirectDepCodeList = new ArrayList<>();
for (String depsNode : depCodeList) { setIndirectDepList(taskCode, indirectDepCodeList);
if (!dag.containsNode(depsNode) for (String depsNode : indirectDepCodeList) {
|| forbiddenTaskList.containsKey(depsNode) if (dag.containsNode(depsNode) && !skipTaskNodeList.containsKey(depsNode)) {
|| skipTaskNodeList.containsKey(depsNode)) { // dependencies must be fully completed
continue; if (!completeTaskList.containsKey(depsNode)) {
} return DependResult.WAITING;
// dependencies must be fully completed }
if (!completeTaskList.containsKey(depsNode)) { ExecutionStatus depTaskState = completeTaskList.get(depsNode).getState();
return DependResult.WAITING; if (depTaskState.typeIsPause() || depTaskState.typeIsCancel()) {
} return DependResult.NON_EXEC;
ExecutionStatus depTaskState = completeTaskList.get(depsNode).getState(); }
if (depTaskState.typeIsPause() || depTaskState.typeIsCancel()) { // ignore task state if current task is condition
return DependResult.NON_EXEC; if (taskNode.isConditionsTask()) {
} continue;
// ignore task state if current task is condition }
if (taskNode.isConditionsTask()) { if (!dependTaskSuccess(depsNode, taskCode)) {
continue; return DependResult.FAILED;
} }
if (!dependTaskSuccess(depsNode, taskCode)) {
return DependResult.FAILED;
} }
} }
logger.info("taskCode: {} completeDependTaskList: {}", taskCode, Arrays.toString(completeTaskList.keySet().toArray())); logger.info("taskCode: {} completeDependTaskList: {}", taskCode, Arrays.toString(completeTaskList.keySet().toArray()));
return DependResult.SUCCESS; return DependResult.SUCCESS;
} }
/**
* This function is specially used to handle the dependency situation where the parent node is a prohibited node.
* When the parent node is a forbidden node, the dependency relationship should continue to be traced
*
* @param taskCode taskCode
* @param indirectDepCodeList All indirectly dependent nodes
*/
private void setIndirectDepList(String taskCode, List<String> indirectDepCodeList) {
TaskNode taskNode = dag.getNode(taskCode);
List<String> depCodeList = taskNode.getDepList();
for (String depsNode : depCodeList) {
if (forbiddenTaskList.containsKey(depsNode)) {
setIndirectDepList(depsNode, indirectDepCodeList);
} else {
indirectDepCodeList.add(depsNode);
}
}
}
/** /**
* depend node is completed, but here need check the condition task branch is the next node * depend node is completed, but here need check the condition task branch is the next node
*/ */