@Test public void testSetHravenQueueName() throws FileNotFoundException { final String JOB_CONF_FILE_NAME = "src/test/resources/job_1329348432655_0001_conf.xml"; Configuration jobConf = new Configuration(); jobConf.addResource(new FileInputStream(JOB_CONF_FILE_NAME)); String USERNAME = "user"; JobKey jobKey = new JobKey("cluster1", USERNAME, "Sleep", 1, "job_1329348432655_0001"); byte[] jobKeyBytes = new JobKeyConverter().toBytes(jobKey); Put jobPut = new Put(jobKeyBytes); byte[] jobConfColumnPrefix = Bytes.toBytes(Constants.JOB_CONF_COLUMN_PREFIX + Constants.SEP); assertEquals(jobPut.size(), 0); JobHistoryService.setHravenQueueNamePut(jobConf, jobPut, jobKey, jobConfColumnPrefix); assertEquals(jobPut.size(), 1); byte[] column = Bytes.add(jobConfColumnPrefix, Constants.HRAVEN_QUEUE_BYTES); assertFoundOnce(column, jobPut, 1, USERNAME); String expH2QName = "hadoop2queue"; String expH1PoolName = "fairpool"; String capacityH1QName = "capacity1queue"; jobConf.set(Constants.CAPACITY_SCHEDULER_QUEUENAME_HADOOP1, capacityH1QName); jobConf.set(Constants.FAIR_SCHEDULER_POOLNAME_HADOOP1, expH1PoolName); jobConf.set(Constants.QUEUENAME_HADOOP2, expH2QName); jobPut = new Put(jobKeyBytes); assertEquals(jobPut.size(), 0); jobConf.set(Constants.CAPACITY_SCHEDULER_QUEUENAME_HADOOP1, capacityH1QName); jobConf.set(Constants.FAIR_SCHEDULER_POOLNAME_HADOOP1, expH1PoolName); jobConf.set(Constants.QUEUENAME_HADOOP2, expH2QName); JobHistoryService.setHravenQueueNamePut(jobConf, jobPut, jobKey, jobConfColumnPrefix); assertEquals(jobPut.size(), 1); assertFoundOnce(column, jobPut, 1, expH2QName); jobConf.set(Constants.CAPACITY_SCHEDULER_QUEUENAME_HADOOP1, capacityH1QName); jobConf.set(Constants.FAIR_SCHEDULER_POOLNAME_HADOOP1, expH1PoolName); jobConf.set(Constants.QUEUENAME_HADOOP2, ""); jobPut = new Put(jobKeyBytes); assertEquals(jobPut.size(), 0); JobHistoryService.setHravenQueueNamePut(jobConf, jobPut, jobKey, jobConfColumnPrefix); assertEquals(jobPut.size(), 1); assertFoundOnce(column, jobPut, 1, expH1PoolName); jobConf.set(Constants.FAIR_SCHEDULER_POOLNAME_HADOOP1, ""); jobConf.set(Constants.QUEUENAME_HADOOP2, ""); jobConf.set(Constants.CAPACITY_SCHEDULER_QUEUENAME_HADOOP1, capacityH1QName); jobPut = new Put(jobKeyBytes); assertEquals(jobPut.size(), 0); JobHistoryService.setHravenQueueNamePut(jobConf, jobPut, jobKey, jobConfColumnPrefix); assertEquals(jobPut.size(), 1); assertFoundOnce(column, jobPut, 1, capacityH1QName); jobConf.set(Constants.CAPACITY_SCHEDULER_QUEUENAME_HADOOP1, ""); jobConf.set(Constants.FAIR_SCHEDULER_POOLNAME_HADOOP1, ""); jobConf.set(Constants.QUEUENAME_HADOOP2, ""); jobPut = new Put(jobKeyBytes); assertEquals(jobPut.size(), 0); JobHistoryService.setHravenQueueNamePut(jobConf, jobPut, jobKey, jobConfColumnPrefix); assertEquals(jobPut.size(), 1); assertFoundOnce(column, jobPut, 1, Constants.DEFAULT_QUEUENAME); }
static void setHravenQueueNamePut(Configuration jobConf, Put jobPut, JobKey jobKey, byte[] jobConfColumnPrefix) { String hRavenQueueName = HadoopConfUtil.getQueueName(jobConf); if (hRavenQueueName.equalsIgnoreCase(Constants.DEFAULT_VALUE_QUEUENAME)) { hRavenQueueName = jobKey.getUserName(); } byte[] column = Bytes.add(jobConfColumnPrefix, Constants.HRAVEN_QUEUE_BYTES); jobPut.addColumn(Constants.INFO_FAM_BYTES, column, Bytes.toBytes(hRavenQueueName)); }
JobHistoryService { static void setHravenQueueNamePut(Configuration jobConf, Put jobPut, JobKey jobKey, byte[] jobConfColumnPrefix) { String hRavenQueueName = HadoopConfUtil.getQueueName(jobConf); if (hRavenQueueName.equalsIgnoreCase(Constants.DEFAULT_VALUE_QUEUENAME)) { hRavenQueueName = jobKey.getUserName(); } byte[] column = Bytes.add(jobConfColumnPrefix, Constants.HRAVEN_QUEUE_BYTES); jobPut.addColumn(Constants.INFO_FAM_BYTES, column, Bytes.toBytes(hRavenQueueName)); } }
JobHistoryService { static void setHravenQueueNamePut(Configuration jobConf, Put jobPut, JobKey jobKey, byte[] jobConfColumnPrefix) { String hRavenQueueName = HadoopConfUtil.getQueueName(jobConf); if (hRavenQueueName.equalsIgnoreCase(Constants.DEFAULT_VALUE_QUEUENAME)) { hRavenQueueName = jobKey.getUserName(); } byte[] column = Bytes.add(jobConfColumnPrefix, Constants.HRAVEN_QUEUE_BYTES); jobPut.addColumn(Constants.INFO_FAM_BYTES, column, Bytes.toBytes(hRavenQueueName)); } JobHistoryService(Configuration hbaseConf, Connection hbaseConnection); }
JobHistoryService { static void setHravenQueueNamePut(Configuration jobConf, Put jobPut, JobKey jobKey, byte[] jobConfColumnPrefix) { String hRavenQueueName = HadoopConfUtil.getQueueName(jobConf); if (hRavenQueueName.equalsIgnoreCase(Constants.DEFAULT_VALUE_QUEUENAME)) { hRavenQueueName = jobKey.getUserName(); } byte[] column = Bytes.add(jobConfColumnPrefix, Constants.HRAVEN_QUEUE_BYTES); jobPut.addColumn(Constants.INFO_FAM_BYTES, column, Bytes.toBytes(hRavenQueueName)); } JobHistoryService(Configuration hbaseConf, Connection hbaseConnection); Flow getLatestFlow(String cluster, String user, String appId); Flow getLatestFlow(String cluster, String user, String appId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      int limit); Flow getFlow(String cluster, String user, String appId, long runId,
      boolean populateTasks); Flow getFlowByJobID(String cluster, String jobId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, int limit); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, long startTime, long endTime,
      int limit); List<Flow> getFlowTimeSeriesStats(String cluster, String user,
      String appId, String version, long startTime, long endTime, int limit,
      byte[] startRow); JobDetails getJobByJobID(String cluster, String jobId); JobDetails getJobByJobID(String cluster, String jobId,
      boolean populateTasks); JobDetails getJobByJobID(QualifiedJobId jobId, boolean populateTasks); static Configuration parseConfiguration(
      Map<byte[], byte[]> keyValues); static CounterMap parseCounters(byte[] prefix,
      Map<byte[], byte[]> keyValues); static List<Put> getHbasePuts(JobDesc jobDesc, Configuration jobConf); int removeJob(JobKey key); }
JobHistoryService { static void setHravenQueueNamePut(Configuration jobConf, Put jobPut, JobKey jobKey, byte[] jobConfColumnPrefix) { String hRavenQueueName = HadoopConfUtil.getQueueName(jobConf); if (hRavenQueueName.equalsIgnoreCase(Constants.DEFAULT_VALUE_QUEUENAME)) { hRavenQueueName = jobKey.getUserName(); } byte[] column = Bytes.add(jobConfColumnPrefix, Constants.HRAVEN_QUEUE_BYTES); jobPut.addColumn(Constants.INFO_FAM_BYTES, column, Bytes.toBytes(hRavenQueueName)); } JobHistoryService(Configuration hbaseConf, Connection hbaseConnection); Flow getLatestFlow(String cluster, String user, String appId); Flow getLatestFlow(String cluster, String user, String appId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      int limit); Flow getFlow(String cluster, String user, String appId, long runId,
      boolean populateTasks); Flow getFlowByJobID(String cluster, String jobId,
      boolean populateTasks); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, int limit); List<Flow> getFlowSeries(String cluster, String user, String appId,
      String version, boolean populateTasks, long startTime, long endTime,
      int limit); List<Flow> getFlowTimeSeriesStats(String cluster, String user,
      String appId, String version, long startTime, long endTime, int limit,
      byte[] startRow); JobDetails getJobByJobID(String cluster, String jobId); JobDetails getJobByJobID(String cluster, String jobId,
      boolean populateTasks); JobDetails getJobByJobID(QualifiedJobId jobId, boolean populateTasks); static Configuration parseConfiguration(
      Map<byte[], byte[]> keyValues); static CounterMap parseCounters(byte[] prefix,
      Map<byte[], byte[]> keyValues); static List<Put> getHbasePuts(JobDesc jobDesc, Configuration jobConf); int removeJob(JobKey key);  }