Merge branch 'dev' into dev (682fd7be) · Commits · SUMMER2020 / students / proj-2002251

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ExecutorService.java

+10 −1

Original line number	Diff line number	Diff line
		@@ -21,6 +21,7 @@ import org.apache.dolphinscheduler.api.enums.ExecuteType;
		import org.apache.dolphinscheduler.api.enums.Status;
		import org.apache.dolphinscheduler.common.Constants;
		import org.apache.dolphinscheduler.common.enums.*;
		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;
		@@ -59,7 +60,7 @@ public class ExecutorService extends BaseService{
		private ProcessDefinitionMapper processDefinitionMapper;

		@Autowired
		private ProcessDefinitionService processDefinitionService;
		private MonitorService monitorService;


		@Autowired
		@@ -123,6 +124,14 @@ public class ExecutorService extends BaseService{
		return result;
		}

		// check master server exists
		List<Server> masterServers = monitorService.getServerListFromZK(true);


		if (masterServers.size() == 0) {
		putMsg(result, Status.MASTER_NOT_EXISTS);
		return result;
		}
		/**
		* create command
		*/

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/SchedulerService.java

+1 −0

Original line number	Diff line number	Diff line
		@@ -365,6 +365,7 @@ public class SchedulerService extends BaseService {

		if (masterServers.size() == 0) {
		putMsg(result, Status.MASTER_NOT_EXISTS);
		return result;
		}

		// set status

dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ExecutorService2Test.java

+38 −5

Original line number	Diff line number	Diff line
		@@ -22,6 +22,7 @@ import org.apache.dolphinscheduler.common.enums.CommandType;
		import org.apache.dolphinscheduler.common.enums.Priority;
		import org.apache.dolphinscheduler.common.enums.ReleaseState;
		import org.apache.dolphinscheduler.common.enums.RunMode;
		import org.apache.dolphinscheduler.common.model.Server;
		import org.apache.dolphinscheduler.dao.entity.*;
		import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
		import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
		@@ -63,6 +64,9 @@ public class ExecutorService2Test {
		@Mock
		private ProjectService projectService;

		@Mock
		private MonitorService monitorService;

		private int processDefinitionId = 1;

		private int tenantId = 1;
		@@ -102,6 +106,7 @@ public class ExecutorService2Test {
		Mockito.when(processDefinitionMapper.selectById(processDefinitionId)).thenReturn(processDefinition);
		Mockito.when(processService.getTenantForProcess(tenantId, userId)).thenReturn(new Tenant());
		Mockito.when(processService.createCommand(any(Command.class))).thenReturn(1);
		Mockito.when(monitorService.getServerListFromZK(true)).thenReturn(getMasterServersList());
		}

		/**
		@@ -121,7 +126,6 @@ public class ExecutorService2Test {
		Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
		verify(processService, times(1)).createCommand(any(Command.class));
		}catch (Exception e){
		Assert.assertTrue(false);
		}
		}

		@@ -142,7 +146,6 @@ public class ExecutorService2Test {
		Assert.assertEquals(Status.START_PROCESS_INSTANCE_ERROR, result.get(Constants.STATUS));
		verify(processService, times(0)).createCommand(any(Command.class));
		}catch (Exception e){
		Assert.assertTrue(false);
		}
		}

		@@ -163,7 +166,6 @@ public class ExecutorService2Test {
		Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
		verify(processService, times(1)).createCommand(any(Command.class));
		}catch (Exception e){
		Assert.assertTrue(false);
		}
		}

		@@ -184,7 +186,6 @@ public class ExecutorService2Test {
		Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
		verify(processService, times(31)).createCommand(any(Command.class));
		}catch (Exception e){
		Assert.assertTrue(false);
		}
		}

		@@ -205,10 +206,42 @@ public class ExecutorService2Test {
		Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
		verify(processService, times(15)).createCommand(any(Command.class));
		}catch (Exception e){
		Assert.assertTrue(false);
		}
		}


		@Test
		public void testNoMsterServers() throws ParseException{
		Mockito.when(monitorService.getServerListFromZK(true)).thenReturn(new ArrayList<Server>());

		Map<String, Object> result = executorService.execProcessInstance(loginUser, projectName,
		processDefinitionId, cronTime, CommandType.COMPLEMENT_DATA,
		null, null,
		null, null, 0,
		"", "", RunMode.RUN_MODE_PARALLEL,
		Priority.LOW, Constants.DEFAULT_WORKER_GROUP, 110);
		Assert.assertEquals(result.get(Constants.STATUS),Status.MASTER_NOT_EXISTS);

		}

		private List<Server> getMasterServersList(){
		List<Server> masterServerList = new ArrayList<>();
		Server masterServer1 = new Server();
		masterServer1.setId(1);
		masterServer1.setHost("192.168.220.188");
		masterServer1.setPort(1121);
		masterServerList.add(masterServer1);

		Server masterServer2 = new Server();
		masterServer2.setId(2);
		masterServer2.setHost("192.168.220.189");
		masterServer2.setPort(1122);
		masterServerList.add(masterServer2);

		return masterServerList;

		}

		private List<Schedule> zeroSchedulerList(){
		return Collections.EMPTY_LIST;
		}

script/env/dolphinscheduler_env.sh

+2 −2

Original line number	Diff line number	Diff line
		@@ -23,6 +23,6 @@ export PYTHON_HOME=/opt/soft/python
		export JAVA_HOME=/opt/soft/java
		export HIVE_HOME=/opt/soft/hive
		export FLINK_HOME=/opt/soft/flink
		export DATAX_HOME=/opt/soft/dataX
		export DATAX_HOME=/opt/soft/datax/bin/datax.py

		export PATH=$HADOOP_HOME/bin:$SPARK_HOME1/bin:$SPARK_HOME2/bin:$PYTHON_HOME:$JAVA_HOME/bin:$HIVE_HOME/bin:$PATH:$FLINK_HOME/bin:$DATAX_HOME/bin:$PATH
		export PATH=$HADOOP_HOME/bin:$SPARK_HOME1/bin:$SPARK_HOME2/bin:$PYTHON_HOME:$JAVA_HOME/bin:$HIVE_HOME/bin:$PATH:$FLINK_HOME/bin:$DATAX_HOME:$PATH