|
|
@ -98,7 +98,11 @@ public class MasterSchedulerThread implements Runnable { |
|
|
|
InterProcessMutex mutex = null; |
|
|
|
InterProcessMutex mutex = null; |
|
|
|
try { |
|
|
|
try { |
|
|
|
|
|
|
|
|
|
|
|
if(OSUtils.checkResource(masterConfig.getMasterMaxCpuloadAvg(), masterConfig.getMasterReservedMemory())){ |
|
|
|
boolean runCheckFlag = OSUtils.checkResource(masterConfig.getMasterMaxCpuloadAvg(), masterConfig.getMasterReservedMemory()); |
|
|
|
|
|
|
|
if(!runCheckFlag) { |
|
|
|
|
|
|
|
Thread.sleep(Constants.SLEEP_TIME_MILLIS); |
|
|
|
|
|
|
|
continue; |
|
|
|
|
|
|
|
} |
|
|
|
if (zkMasterClient.getZkClient().getState() == CuratorFrameworkState.STARTED) { |
|
|
|
if (zkMasterClient.getZkClient().getState() == CuratorFrameworkState.STARTED) { |
|
|
|
|
|
|
|
|
|
|
|
// create distributed lock with the root node path of the lock space as /dolphinscheduler/lock/masters
|
|
|
|
// create distributed lock with the root node path of the lock space as /dolphinscheduler/lock/masters
|
|
|
@ -124,13 +128,11 @@ public class MasterSchedulerThread implements Runnable { |
|
|
|
logger.error("scan command error ", e); |
|
|
|
logger.error("scan command error ", e); |
|
|
|
processDao.moveToErrorCommand(command, e.toString()); |
|
|
|
processDao.moveToErrorCommand(command, e.toString()); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
} else{ |
|
|
|
|
|
|
|
//indicate that no command ,sleep for 1s
|
|
|
|
|
|
|
|
Thread.sleep(Constants.SLEEP_TIME_MILLIS); |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// accessing the command table every SLEEP_TIME_MILLIS milliseconds
|
|
|
|
|
|
|
|
Thread.sleep(Constants.SLEEP_TIME_MILLIS); |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
}catch (Exception e){ |
|
|
|
}catch (Exception e){ |
|
|
|
logger.error("master scheduler thread exception : " + e.getMessage(),e); |
|
|
|
logger.error("master scheduler thread exception : " + e.getMessage(),e); |
|
|
|
}finally{ |
|
|
|
}finally{ |
|
|
|