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

conflict reslove

Browse Source
This commit is contained in:
qiaozhanwei 2020-04-01 18:29:47 +08:00
parent ca096e3652
commit 4bda3cd92c
3 changed files with 13 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java
View File

@ -21,6 +21,7 @@ import com.alibaba.fastjson.JSONObject;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.model.TaskNode;
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.task.TaskTimeoutParameter;
import org.apache.dolphinscheduler.common.utils.*;
@ -35,6 +36,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
/**
@ -211,9 +213,11 @@ public class TaskExecuteThread implements Runnable {
AbstractParameters baseParam = TaskParametersUtils.getParameters(taskNode.getType(), taskNode.getParams());
if (baseParam != null) {
List<String> projectResourceFiles = baseParam.getResourceFilesList();
List<ResourceInfo> projectResourceFiles = baseParam.getResourceFilesList();
if (projectResourceFiles != null) {
projectFiles.addAll(projectResourceFiles);
Stream<String> resourceInfotream = projectResourceFiles.stream().map(resourceInfo -> resourceInfo.getRes());
projectFiles.addAll(resourceInfotream.collect(Collectors.toList()));
}
}
@ -236,7 +240,7 @@ public class TaskExecuteThread implements Runnable {
if (!resFile.exists()) {
try {
// query the tenant code of the resource according to the name of the resource
String resHdfsPath = HadoopUtils.getHdfsFilename(tenantCode, resource);
String resHdfsPath = HadoopUtils.getHdfsResourceFileName(tenantCode, resource);
logger.info("get resource file from hdfs :{}", resHdfsPath);
HadoopUtils.getInstance().copyHdfsToLocal(resHdfsPath, execLocalPath + File.separator + resource, false, true);

10
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/zk/ZKMasterClient.java
View File

@ -78,7 +78,7 @@ public class ZKMasterClient extends AbstractZKClient {
}
}catch (Exception e){
logger.error("master start up exception",e);
logger.error("master start up exception",e);
}finally {
releaseMutex(mutex);
}
@ -97,13 +97,13 @@ public class ZKMasterClient extends AbstractZKClient {
*/
@Override
protected void dataChanged(CuratorFramework client, TreeCacheEvent event, String path) {
if(path.startsWith(getZNodeParentPath(ZKNodeType.MASTER)+Constants.SINGLE_SLASH)){ //monitor master
//monitor master
if(path.startsWith(getZNodeParentPath(ZKNodeType.MASTER)+Constants.SINGLE_SLASH)){
handleMasterEvent(event,path);
}else if(path.startsWith(getZNodeParentPath(ZKNodeType.WORKER)+Constants.SINGLE_SLASH)){ //monitor worker
}else if(path.startsWith(getZNodeParentPath(ZKNodeType.WORKER)+Constants.SINGLE_SLASH)){
//monitor worker
handleWorkerEvent(event,path);
}
//other path event, ignore
}
/**

2
dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/zk/AbstractZKClient.java
View File

@ -126,7 +126,7 @@ public abstract class AbstractZKClient extends ZookeeperCachedOperator {
int i = 0;
for (Map.Entry<String, String> entry : masterMap.entrySet()) {
Server masterServer = ResInfo.parseHeartbeatForZKInfo(entry.getValue());
masterServer.setZkDirectory( parentPath + "/"+ entry.getKey());
masterServer.setZkDirectory(parentPath + "/"+ entry.getKey());
masterServer.setId(i);
i ++;
masterServers.add(masterServer);