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 ; // delete all task rows Scan taskScan = getTaskScan ( key ) ; // only need the row keys back to delete (all should have taskid) taskScan . addColumn ( Constants . INFO_FAM_BYTES , JobHistoryKeys . KEYS_TO_BYTES . get ( JobHistoryKeys . TASKID ) ) ; // no reason to cache rows we're deleting 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 ( ) ) ) ; } } // Hang on the count because delete will modify our list. 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 ; }
Removes the job s row from the job_history table and all related task rows from the job_history_task table .