Master变更之后将原先执行中的任务设置为失败状态

@@ -6,3 +6,4 @@ job/.classpath
 job/.project
 web/.classpath
 web/.project
+build/to_remote/war/yoho-kisjob-web.war

@@ -62,6 +62,13 @@ public class NodeUtil {
 		return localIpAddress;
 	}
 
+	public static boolean isCurrentNode(String nodeIdentifier) {
+		if (nodeIdentifier == null) {
+			return false;
+		}
+		return nodeIdentifier.equals(getCurrentNodeIp());
+	}
+
 	private static boolean isPublicIpAddress(final InetAddress ipAddress) {
 		return !ipAddress.isSiteLocalAddress() && !ipAddress.isLoopbackAddress() && !isV6IpAddress(ipAddress);
 	}

@@ -467,6 +467,14 @@ public class DefaultRegistryCenter implements RegistryCenter {
 					String jobName = map.get("jobName");
 					JobMeta jobMeta = jobMetaService.queryByKey(nodeGroup, jobName);
 					Long jobInstanceId = Long.valueOf(zkHelper.get(event.getData().getPath()));
+					KisJobInstance jobInstance = jobInstanceService.queryJobInstById(jobInstanceId);
+					if (jobInstance == null || jobInstance.getInstanceStatus().intValue() != State.Running.getCode()) {
+						// 如果实例不存在或者不是running状态的 就不通知调度中心去执行
+						log.info(
+								"The job instance is not running and skip execution groups. jobName: {}, jobInstanceId: {}",
+								jobName, jobInstanceId);
+						return;
+					}
 					List<Integer> groupIndexs = jobInstGroupService.queryJobInstanceGroupIndexs(jobInstanceId);
 					Assert.notEmpty(groupIndexs);
 					Map<Integer, String> groupDispatchResult = new HashMap<Integer, String>();

@@ -2,7 +2,6 @@ package com.yoho.kisjob.reg.persistence.service;
 
 import java.util.Date;
 import java.util.List;
-import java.util.Map;
 
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -11,12 +10,15 @@ import org.springframework.util.Assert;
 import com.yoho.kisjob.common.constant.JobException;
 import com.yoho.kisjob.common.constant.State;
 import com.yoho.kisjob.common.meta.ExecutionMeta;
+import com.yoho.kisjob.common.meta.JobMeta;
+import com.yoho.kisjob.common.meta.JobType;
 import com.yoho.kisjob.common.utils.NodeUtil;
 import com.yoho.kisjob.common.utils.RandomUtils;
 import com.yoho.kisjob.common.utils.Utils;
 import com.yoho.kisjob.reg.persistence.service.jdbc.JdbcService;
 import com.yoho.kisjob.reg.persistence.vo.KisJob;
 import com.yoho.kisjob.reg.persistence.vo.KisJobInstance;
+import com.yoho.kisjob.reg.persistence.vo.KisJobInstanceGroup;
 
 /**
  * 定时任务实例持久化服务类。
@@ -34,6 +36,9 @@ public class JobInstanceService {
 	@Autowired
 	private JobInstGroupService jobInstGroupService;
 
+	@Autowired
+	private JobMetaService jobMetaService;
+
 	public Long addNewInstance(String nodeGroup, String jobName) {
 		List<KisJob> list = jdbcService.query(new KisJob(nodeGroup, jobName));
 		Assert.notEmpty(list, "For " + jobName);
@@ -81,21 +86,37 @@ public class JobInstanceService {
 		condition.setInstanceStatus(State.Running.getCode());
 		List<KisJobInstance> list = jdbcService.query(condition);
 		if (Utils.isNotEmpty(list)) {
+
+			JobMeta jobMeta = jobMetaService.queryByJobId(jobId);
+			JobType jobType = jobMeta != null ? jobMeta.getJobType() : null;
 			for (KisJobInstance jobInstance : list) {
-				if (NodeUtil.getCurrentNodeIp().equals(jobInstance.getNodeIdentifier())) {
+				if (NodeUtil.isCurrentNode(jobInstance.getNodeIdentifier())) {
 					continue;
 				}
 
-				Map<Integer, State> groupStatusMap = jobInstGroupService.queryJobInstanceGroupStatus(jobInstance
-						.getInstanceId());
-				for (Map.Entry<Integer, State> entry : groupStatusMap.entrySet()) {
-					if (entry.getValue() == State.Ready || entry.getValue() == State.Running) {
-						jobInstGroupService.endExecuteGroup(jobInstance.getInstanceId(), entry.getKey(),
-								new JobException("Execute group failed for master crashed."));
-					}
+				if (jobType == JobType.ElastaicJob) {
+					updateUncompletedJobInstGroups(jobInstance);
 				}
+
 				updateInstanceResult(jobInstance.getInstanceId(), null, new JobException(
-						"Execute instance failed for master crashed."));
+						"Execute instance failed for master %s crashed.", jobInstance.getNodeIdentifier()));
+			}
+		}
+	}
+
+	public void updateUncompletedJobInstGroups(KisJobInstance jobInstance) {
+		State groupState;
+		List<KisJobInstanceGroup> groups = jobInstGroupService.queryJobInstanceGroups(new KisJobInstanceGroup(
+				jobInstance.getInstanceId()));
+		if (Utils.isNotEmpty(groups)) {
+			// TODO:暂时先把master上面的分组更新成失败的 这个后续需要优化
+			for (KisJobInstanceGroup group : groups) {
+				groupState = State.build(group.getGroupState());
+				if ((groupState == State.Ready || groupState == State.Running)
+						&& jobInstance.getNodeIdentifier().equals(group.getNodeIdentifier())) {
+					jobInstGroupService.endExecuteGroup(jobInstance.getInstanceId(), group.getGroupIndex(),
+							new JobException("Execute group failed for master %s crashed.", group.getNodeIdentifier()));
+				}
 			}
 		}
 	}