@Test(expectedExceptions = IllegalArgumentException.class) public void testTaskExecutorConfShouldThrowException() throws Exception { TaskExecutor taskExecutor = new TaskExecutor(); Configuration tonyConf = new Configuration(false); tonyConf.setInt(TonyConfigurationKeys.TASK_HEARTBEAT_INTERVAL_MS, 2000); File confFile = new File(System.getProperty("user.dir"), Constants.TONY_FINAL_XML); try (OutputStream os = new FileOutputStream(confFile)) { tonyConf.writeXml(os); } if (!confFile.delete()) { throw new RuntimeException("Failed to delete conf file"); } taskExecutor.initConfigs(); }
protected void initConfigs() { jobName = System.getenv(Constants.JOB_NAME); appIdString = System.getenv(Constants.APPID); taskIndex = Integer.parseInt(System.getenv(Constants.TASK_INDEX)); numTasks = Integer.parseInt(System.getenv(Constants.TASK_NUM)); taskId = jobName + ":" + taskIndex; LOG.info("Executor is running task " + taskId); String isChiefEnvValue = System.getenv(Constants.IS_CHIEF); isChief = Boolean.parseBoolean(isChiefEnvValue); amHost = System.getenv(Constants.AM_HOST); amPort = Integer.parseInt(System.getenv(Constants.AM_PORT)); tonyConf.addResource(new Path(Constants.TONY_FINAL_XML)); timeOut = tonyConf.getInt(TonyConfigurationKeys.WORKER_TIMEOUT, TonyConfigurationKeys.DEFAULT_WORKER_TIMEOUT); hbInterval = tonyConf.getInt(TonyConfigurationKeys.TASK_HEARTBEAT_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_HEARTBEAT_INTERVAL_MS); String[] shellEnvs = tonyConf.getStrings(TonyConfigurationKeys.EXECUTION_ENV); shellEnv = Utils.parseKeyValue(shellEnvs); taskCommand = tonyConf.get(TonyConfigurationKeys.getExecuteCommandKey(jobName), tonyConf.get(TonyConfigurationKeys.getContainerExecuteCommandKey())); if (taskCommand == null) { LOG.fatal("Task command is empty. Please set tony.[jobtype].command " + "or pass --executes in command line"); throw new IllegalArgumentException("Task command is empty."); } LOG.info("Task command: " + taskCommand); framework = MLFramework.valueOf( tonyConf.get(TonyConfigurationKeys.FRAMEWORK_NAME, TonyConfigurationKeys.DEFAULT_FRAMEWORK_NAME).toUpperCase()); metricsRPCPort = Integer.parseInt(System.getenv(Constants.METRICS_RPC_PORT)); metricsIntervalMs = tonyConf.getInt(TonyConfigurationKeys.TASK_METRICS_UPDATE_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_METRICS_UPDATE_INTERVAL_MS); Utils.initYarnConf(yarnConf); Utils.initHdfsConf(hdfsConf); }
TaskExecutor { protected void initConfigs() { jobName = System.getenv(Constants.JOB_NAME); appIdString = System.getenv(Constants.APPID); taskIndex = Integer.parseInt(System.getenv(Constants.TASK_INDEX)); numTasks = Integer.parseInt(System.getenv(Constants.TASK_NUM)); taskId = jobName + ":" + taskIndex; LOG.info("Executor is running task " + taskId); String isChiefEnvValue = System.getenv(Constants.IS_CHIEF); isChief = Boolean.parseBoolean(isChiefEnvValue); amHost = System.getenv(Constants.AM_HOST); amPort = Integer.parseInt(System.getenv(Constants.AM_PORT)); tonyConf.addResource(new Path(Constants.TONY_FINAL_XML)); timeOut = tonyConf.getInt(TonyConfigurationKeys.WORKER_TIMEOUT, TonyConfigurationKeys.DEFAULT_WORKER_TIMEOUT); hbInterval = tonyConf.getInt(TonyConfigurationKeys.TASK_HEARTBEAT_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_HEARTBEAT_INTERVAL_MS); String[] shellEnvs = tonyConf.getStrings(TonyConfigurationKeys.EXECUTION_ENV); shellEnv = Utils.parseKeyValue(shellEnvs); taskCommand = tonyConf.get(TonyConfigurationKeys.getExecuteCommandKey(jobName), tonyConf.get(TonyConfigurationKeys.getContainerExecuteCommandKey())); if (taskCommand == null) { LOG.fatal("Task command is empty. Please set tony.[jobtype].command " + "or pass --executes in command line"); throw new IllegalArgumentException("Task command is empty."); } LOG.info("Task command: " + taskCommand); framework = MLFramework.valueOf( tonyConf.get(TonyConfigurationKeys.FRAMEWORK_NAME, TonyConfigurationKeys.DEFAULT_FRAMEWORK_NAME).toUpperCase()); metricsRPCPort = Integer.parseInt(System.getenv(Constants.METRICS_RPC_PORT)); metricsIntervalMs = tonyConf.getInt(TonyConfigurationKeys.TASK_METRICS_UPDATE_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_METRICS_UPDATE_INTERVAL_MS); Utils.initYarnConf(yarnConf); Utils.initHdfsConf(hdfsConf); } }
TaskExecutor { protected void initConfigs() { jobName = System.getenv(Constants.JOB_NAME); appIdString = System.getenv(Constants.APPID); taskIndex = Integer.parseInt(System.getenv(Constants.TASK_INDEX)); numTasks = Integer.parseInt(System.getenv(Constants.TASK_NUM)); taskId = jobName + ":" + taskIndex; LOG.info("Executor is running task " + taskId); String isChiefEnvValue = System.getenv(Constants.IS_CHIEF); isChief = Boolean.parseBoolean(isChiefEnvValue); amHost = System.getenv(Constants.AM_HOST); amPort = Integer.parseInt(System.getenv(Constants.AM_PORT)); tonyConf.addResource(new Path(Constants.TONY_FINAL_XML)); timeOut = tonyConf.getInt(TonyConfigurationKeys.WORKER_TIMEOUT, TonyConfigurationKeys.DEFAULT_WORKER_TIMEOUT); hbInterval = tonyConf.getInt(TonyConfigurationKeys.TASK_HEARTBEAT_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_HEARTBEAT_INTERVAL_MS); String[] shellEnvs = tonyConf.getStrings(TonyConfigurationKeys.EXECUTION_ENV); shellEnv = Utils.parseKeyValue(shellEnvs); taskCommand = tonyConf.get(TonyConfigurationKeys.getExecuteCommandKey(jobName), tonyConf.get(TonyConfigurationKeys.getContainerExecuteCommandKey())); if (taskCommand == null) { LOG.fatal("Task command is empty. Please set tony.[jobtype].command " + "or pass --executes in command line"); throw new IllegalArgumentException("Task command is empty."); } LOG.info("Task command: " + taskCommand); framework = MLFramework.valueOf( tonyConf.get(TonyConfigurationKeys.FRAMEWORK_NAME, TonyConfigurationKeys.DEFAULT_FRAMEWORK_NAME).toUpperCase()); metricsRPCPort = Integer.parseInt(System.getenv(Constants.METRICS_RPC_PORT)); metricsIntervalMs = tonyConf.getInt(TonyConfigurationKeys.TASK_METRICS_UPDATE_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_METRICS_UPDATE_INTERVAL_MS); Utils.initYarnConf(yarnConf); Utils.initHdfsConf(hdfsConf); } protected  TaskExecutor(); }
TaskExecutor { protected void initConfigs() { jobName = System.getenv(Constants.JOB_NAME); appIdString = System.getenv(Constants.APPID); taskIndex = Integer.parseInt(System.getenv(Constants.TASK_INDEX)); numTasks = Integer.parseInt(System.getenv(Constants.TASK_NUM)); taskId = jobName + ":" + taskIndex; LOG.info("Executor is running task " + taskId); String isChiefEnvValue = System.getenv(Constants.IS_CHIEF); isChief = Boolean.parseBoolean(isChiefEnvValue); amHost = System.getenv(Constants.AM_HOST); amPort = Integer.parseInt(System.getenv(Constants.AM_PORT)); tonyConf.addResource(new Path(Constants.TONY_FINAL_XML)); timeOut = tonyConf.getInt(TonyConfigurationKeys.WORKER_TIMEOUT, TonyConfigurationKeys.DEFAULT_WORKER_TIMEOUT); hbInterval = tonyConf.getInt(TonyConfigurationKeys.TASK_HEARTBEAT_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_HEARTBEAT_INTERVAL_MS); String[] shellEnvs = tonyConf.getStrings(TonyConfigurationKeys.EXECUTION_ENV); shellEnv = Utils.parseKeyValue(shellEnvs); taskCommand = tonyConf.get(TonyConfigurationKeys.getExecuteCommandKey(jobName), tonyConf.get(TonyConfigurationKeys.getContainerExecuteCommandKey())); if (taskCommand == null) { LOG.fatal("Task command is empty. Please set tony.[jobtype].command " + "or pass --executes in command line"); throw new IllegalArgumentException("Task command is empty."); } LOG.info("Task command: " + taskCommand); framework = MLFramework.valueOf( tonyConf.get(TonyConfigurationKeys.FRAMEWORK_NAME, TonyConfigurationKeys.DEFAULT_FRAMEWORK_NAME).toUpperCase()); metricsRPCPort = Integer.parseInt(System.getenv(Constants.METRICS_RPC_PORT)); metricsIntervalMs = tonyConf.getInt(TonyConfigurationKeys.TASK_METRICS_UPDATE_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_METRICS_UPDATE_INTERVAL_MS); Utils.initYarnConf(yarnConf); Utils.initHdfsConf(hdfsConf); } protected  TaskExecutor(); static void main(String[] unused); }
TaskExecutor { protected void initConfigs() { jobName = System.getenv(Constants.JOB_NAME); appIdString = System.getenv(Constants.APPID); taskIndex = Integer.parseInt(System.getenv(Constants.TASK_INDEX)); numTasks = Integer.parseInt(System.getenv(Constants.TASK_NUM)); taskId = jobName + ":" + taskIndex; LOG.info("Executor is running task " + taskId); String isChiefEnvValue = System.getenv(Constants.IS_CHIEF); isChief = Boolean.parseBoolean(isChiefEnvValue); amHost = System.getenv(Constants.AM_HOST); amPort = Integer.parseInt(System.getenv(Constants.AM_PORT)); tonyConf.addResource(new Path(Constants.TONY_FINAL_XML)); timeOut = tonyConf.getInt(TonyConfigurationKeys.WORKER_TIMEOUT, TonyConfigurationKeys.DEFAULT_WORKER_TIMEOUT); hbInterval = tonyConf.getInt(TonyConfigurationKeys.TASK_HEARTBEAT_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_HEARTBEAT_INTERVAL_MS); String[] shellEnvs = tonyConf.getStrings(TonyConfigurationKeys.EXECUTION_ENV); shellEnv = Utils.parseKeyValue(shellEnvs); taskCommand = tonyConf.get(TonyConfigurationKeys.getExecuteCommandKey(jobName), tonyConf.get(TonyConfigurationKeys.getContainerExecuteCommandKey())); if (taskCommand == null) { LOG.fatal("Task command is empty. Please set tony.[jobtype].command " + "or pass --executes in command line"); throw new IllegalArgumentException("Task command is empty."); } LOG.info("Task command: " + taskCommand); framework = MLFramework.valueOf( tonyConf.get(TonyConfigurationKeys.FRAMEWORK_NAME, TonyConfigurationKeys.DEFAULT_FRAMEWORK_NAME).toUpperCase()); metricsRPCPort = Integer.parseInt(System.getenv(Constants.METRICS_RPC_PORT)); metricsIntervalMs = tonyConf.getInt(TonyConfigurationKeys.TASK_METRICS_UPDATE_INTERVAL_MS, TonyConfigurationKeys.DEFAULT_TASK_METRICS_UPDATE_INTERVAL_MS); Utils.initYarnConf(yarnConf); Utils.initHdfsConf(hdfsConf); } protected  TaskExecutor(); static void main(String[] unused);  }