@Test public void testRemoveJob() throws Exception { flowDataGen.loadFlow("c1@local", "ruser", "removeJob", 1234, "a", 3, 10, idService, historyTable); JobHistoryService service = new JobHistoryService(UTIL.getConfiguration(), hbaseConnection); Flow flow = service.getLatestFlow("c1@local", "ruser", "removeJob"); assertNotNull(flow); assertEquals(3, flow.getJobs().size()); List<JobDetails> origJobs = flow.getJobs(); JobDetails toRemove = origJobs.get(0); origJobs.remove(0); LOG.info("Removing job " + toRemove.getJobKey()); service.removeJob(toRemove.getJobKey()); Flow flow2 = service.getLatestFlow("c1@local", "ruser", "removeJob"); assertNotNull(flow2); assertEquals(2, flow2.getJobs().size()); for (JobDetails j : flow2.getJobs()) { if (j.getJobKey().equals(toRemove.getJobKey())) { fail("Removed job (" + toRemove.getJobKey() + ") is still present in flow!"); } } List<JobDetails> flow2Jobs = flow2.getJobs(); assertEquals(origJobs.size(), flow2Jobs.size()); for (int i = 0; i < origJobs.size(); i++) { JobDetails j1 = origJobs.get(i); JobDetails j2 = flow2Jobs.get(i); assertJob(j1, j2); } }
public int removeJob(JobKey key) throws IOException { byte[] jobRow = jobKeyConv.toBytes(key); Table historyTable = hbaseConnection.getTable(TableName.valueOf(Constants.HISTORY_TABLE)); historyTable.delete(new Delete(jobRow)); historyTable.close(); int deleteCount = 1; Scan taskScan = getTaskScan(key); taskScan.addColumn(Constants.INFO_FAM_BYTES, JobHistoryKeys.KEYS_TO_BYTES.get(JobHistoryKeys.TASKID)); taskScan.setCacheBlocks(false); List<Delete> taskDeletes = new ArrayList<Delete>(); Table taskTable = hbaseConnection .getTable(TableName.valueOf(Constants.HISTORY_TASK_TABLE)); ResultScanner scanner = taskTable.getScanner(taskScan); try { for (Result r : scanner) { if (r != null && !r.isEmpty()) { byte[] rowKey = r.getRow(); TaskKey taskKey = taskKeyConv.fromBytes(rowKey); if (!key.equals(taskKey)) { LOG.warn("Found task not in the current job " + Bytes.toStringBinary(rowKey)); break; } taskDeletes.add(new Delete(r.getRow())); } } deleteCount += taskDeletes.size(); if (taskDeletes.size() > 0) { LOG.info("Deleting " + taskDeletes.size() + " tasks for job " + key); taskTable.delete(taskDeletes); } } finally { scanner.close(); taskTable.close(); } return deleteCount; }
JobHistoryService { public int removeJob(JobKey key) throws IOException { byte[] jobRow = jobKeyConv.toBytes(key); Table historyTable = hbaseConnection.getTable(TableName.valueOf(Constants.HISTORY_TABLE)); historyTable.delete(new Delete(jobRow)); historyTable.close(); int deleteCount = 1; Scan taskScan = getTaskScan(key); taskScan.addColumn(Constants.INFO_FAM_BYTES, JobHistoryKeys.KEYS_TO_BYTES.get(JobHistoryKeys.TASKID)); taskScan.setCacheBlocks(false); List<Delete> taskDeletes = new ArrayList<Delete>(); Table taskTable = hbaseConnection .getTable(TableName.valueOf(Constants.HISTORY_TASK_TABLE)); ResultScanner scanner = taskTable.getScanner(taskScan); try { for (Result r : scanner) { if (r != null && !r.isEmpty()) { byte[] rowKey = r.getRow(); TaskKey taskKey = taskKeyConv.fromBytes(rowKey); if (!key.equals(taskKey)) { LOG.warn("Found task not in the current job " + Bytes.toStringBinary(rowKey)); break; } taskDeletes.add(new Delete(r.getRow())); } } deleteCount += taskDeletes.size(); if (taskDeletes.size() > 0) { LOG.info("Deleting " + taskDeletes.size() + " tasks for job " + key); taskTable.delete(taskDeletes); } } finally { scanner.close(); taskTable.close(); } return deleteCount; } }
JobHistoryService { public int removeJob(JobKey key) throws IOException { byte[] jobRow = jobKeyConv.toBytes(key); Table historyTable = hbaseConnection.getTable(TableName.valueOf(Constants.HISTORY_TABLE)); historyTable.delete(new Delete(jobRow)); historyTable.close(); int deleteCount = 1; Scan taskScan = getTaskScan(key); taskScan.addColumn(Constants.INFO_FAM_BYTES, JobHistoryKeys.KEYS_TO_BYTES.get(JobHistoryKeys.TASKID)); taskScan.setCacheBlocks(false); List<Delete> taskDeletes = new ArrayList<Delete>(); Table taskTable = hbaseConnection .getTable(TableName.valueOf(Constants.HISTORY_TASK_TABLE)); ResultScanner scanner = taskTable.getScanner(taskScan); try { for (Result r : scanner) { if (r != null && !r.isEmpty()) { byte[] rowKey = r.getRow(); TaskKey taskKey = taskKeyConv.fromBytes(rowKey); if (!key.equals(taskKey)) { LOG.warn("Found task not in the current job " + Bytes.toStringBinary(rowKey)); break; } taskDeletes.add(new Delete(r.getRow())); } } deleteCount += taskDeletes.size(); if (taskDeletes.size() > 0) { LOG.info("Deleting " + taskDeletes.size() + " tasks for job " + key); taskTable.delete(taskDeletes); } } finally { scanner.close(); taskTable.close(); } return deleteCount; } JobHistoryService(Configuration hbaseConf, Connection hbaseConnection); }
JobHistoryService { public int removeJob(JobKey key) throws IOException { byte[] jobRow = jobKeyConv.toBytes(key); Table historyTable = hbaseConnection.getTable(TableName.valueOf(Constants.HISTORY_TABLE)); historyTable.delete(new Delete(jobRow)); historyTable.close(); int deleteCount = 1; Scan taskScan = getTaskScan(key); taskScan.addColumn(Constants.INFO_FAM_BYTES, JobHistoryKeys.KEYS_TO_BYTES.get(JobHistoryKeys.TASKID)); taskScan.setCacheBlocks(false); List<Delete> taskDeletes = new ArrayList<Delete>(); Table taskTable = hbaseConnection .getTable(TableName.valueOf(Constants.HISTORY_TASK_TABLE)); ResultScanner scanner = taskTable.getScanner(taskScan); try { for (Result r : scanner) { if (r != null && !r.isEmpty()) { byte[] rowKey = r.getRow(); TaskKey taskKey = taskKeyConv.fromBytes(rowKey); if (!key.equals(taskKey)) { LOG.warn("Found task not in the current job " + Bytes.toStringBinary(rowKey)); break; } taskDeletes.add(new Delete(r.getRow())); } } deleteCount += taskDeletes.size(); if (taskDeletes.size() > 0) { LOG.info("Deleting " + taskDeletes.size() + " tasks for job " + key); taskTable.delete(taskDeletes); } } finally { scanner.close(); taskTable.close(); } return deleteCount; } 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 { public int removeJob(JobKey key) throws IOException { byte[] jobRow = jobKeyConv.toBytes(key); Table historyTable = hbaseConnection.getTable(TableName.valueOf(Constants.HISTORY_TABLE)); historyTable.delete(new Delete(jobRow)); historyTable.close(); int deleteCount = 1; Scan taskScan = getTaskScan(key); taskScan.addColumn(Constants.INFO_FAM_BYTES, JobHistoryKeys.KEYS_TO_BYTES.get(JobHistoryKeys.TASKID)); taskScan.setCacheBlocks(false); List<Delete> taskDeletes = new ArrayList<Delete>(); Table taskTable = hbaseConnection .getTable(TableName.valueOf(Constants.HISTORY_TASK_TABLE)); ResultScanner scanner = taskTable.getScanner(taskScan); try { for (Result r : scanner) { if (r != null && !r.isEmpty()) { byte[] rowKey = r.getRow(); TaskKey taskKey = taskKeyConv.fromBytes(rowKey); if (!key.equals(taskKey)) { LOG.warn("Found task not in the current job " + Bytes.toStringBinary(rowKey)); break; } taskDeletes.add(new Delete(r.getRow())); } } deleteCount += taskDeletes.size(); if (taskDeletes.size() > 0) { LOG.info("Deleting " + taskDeletes.size() + " tasks for job " + key); taskTable.delete(taskDeletes); } } finally { scanner.close(); taskTable.close(); } return deleteCount; } 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);  }