AbstractHadoopJavaProcessJob extends JavaProcessJob implements IHadoopJob { @Override public void setupHadoopJobProperties() { String[] tagKeys = new String[]{ CommonJobProperties.EXEC_ID, CommonJobProperties.FLOW_ID, CommonJobProperties.PROJECT_NAME, CommonJobProperties.AZKABAN_WEBSERVERHOST, CommonJobProperties.JOB_ID, CommonJobProperties.JOB_ATTEMPT }; Props jobProps = getJobProps(); String tagList = HadoopJobUtils.constructHadoopTags(jobProps, tagKeys); if (jobProps.containsKey(CommonJobProperties.PROJECT_NAME) && jobProps.containsKey(CommonJobProperties.FLOW_ID)) { String workflowTag = "workflowid:" + jobProps.get(CommonJobProperties.PROJECT_NAME) + HadoopConfigurationInjector.WORKFLOW_ID_SEPERATOR + jobProps.get(CommonJobProperties.FLOW_ID); workflowTag = workflowTag.substring(0, Math.min(workflowTag.length(), HadoopJobUtils.APPLICATION_TAG_MAX_LENGTH)); tagList = Joiner.on(',').skipNulls().join(tagList, workflowTag); } getJobProps().put( HadoopConfigurationInjector.INJECT_PREFIX + HadoopJobUtils.MAPREDUCE_JOB_TAGS, tagList ); } AbstractHadoopJavaProcessJob(String jobid, Props sysProps, Props jobProps, Logger logger);  AbstractHadoopJavaProcessJob(String jobid, Props sysProps, Props jobProps,
      Props privateProps, Logger logger); @Override void run(); @Override void setupHadoopJobProperties(); @Override Props appendExtraProps(Props props); @Override HadoopProxy getHadoopProxy();  }
@Test public void testWorkflowIdTag() { Props props = new Props(); props.put(CommonJobProperties.EXEC_ID, "123"); props.put(CommonJobProperties.PROJECT_NAME, "project-name"); props.put(CommonJobProperties.FLOW_ID, "flow-id"); AbstractHadoopJavaProcessJob job = new AbstractHadoopJavaProcessJob("test", new Props(), props, null) {}; job.setupHadoopJobProperties(); assertThat(job.getJobProps().get(HadoopConfigurationInjector.INJECT_PREFIX + HadoopJobUtils.MAPREDUCE_JOB_TAGS)) .isEqualTo("azkaban.flow.execid:123,azkaban.flow.flowid:flow-id" + ",azkaban.flow.projectname:project-name,workflowid:project-name$flow-id"); }
@Test public void testLongWorkflowIdTag() { Props props = new Props(); StringBuffer flowIdBuffer = new StringBuffer("flow-id"); for (int i = 0; i < 150; i++) { flowIdBuffer.append("f"); } props.put(CommonJobProperties.EXEC_ID, "123"); props.put(CommonJobProperties.PROJECT_NAME, "project-name"); props.put(CommonJobProperties.FLOW_ID, flowIdBuffer.toString()); AbstractHadoopJavaProcessJob job = new AbstractHadoopJavaProcessJob("test2", new Props(), props, null) {}; job.setupHadoopJobProperties(); String[] actualTags = job.getJobProps().get( HadoopConfigurationInjector.INJECT_PREFIX + HadoopJobUtils.MAPREDUCE_JOB_TAGS) .split(","); assertThat(actualTags.length).isEqualTo(4); assertThat(actualTags[0]).isEqualTo("azkaban.flow.execid:123"); String flowPrefix = "azkaban.flow.flowid:flow-id"; StringBuffer expectedFlowTagBuffer = new StringBuffer(flowPrefix); for (int i = flowPrefix.length(); i < HadoopJobUtils.APPLICATION_TAG_MAX_LENGTH; i++) { expectedFlowTagBuffer.append("f"); } assertThat(actualTags[1]).isEqualTo(expectedFlowTagBuffer.toString()); assertThat(actualTags[2]).isEqualTo("azkaban.flow.projectname:project-name"); assertThat(actualTags[3].length()).isLessThanOrEqualTo(HadoopJobUtils.APPLICATION_TAG_MAX_LENGTH); String workflowPrefix = "workflowid:project-name$flow-id"; StringBuffer expectedTagBuffer = new StringBuffer(); expectedTagBuffer.append(workflowPrefix); for (int i = workflowPrefix.length(); i < HadoopJobUtils.APPLICATION_TAG_MAX_LENGTH; i++) { expectedTagBuffer.append("f"); } assertThat(actualTags[3]).isEqualTo(expectedTagBuffer.toString()); }