public List < AppSummary > getNewApps ( JobHistoryService jhs , String cluster , String user , long startTime , long endTime , int limit ) throws IOException { byte [ ] startRow = null ; if ( StringUtils . isNotBlank ( user ) ) { startRow = ByteUtil . join ( Constants . SEP_BYTES , Bytes . toBytes ( cluster ) , Bytes . toBytes ( user ) ) ; } else { startRow = ByteUtil . join ( Constants . SEP_BYTES , Bytes . toBytes ( cluster ) ) ; } LOG . info ( "Reading app version rows start at " + Bytes . toStringBinary ( startRow ) ) ; Scan scan = new Scan ( ) ; // start scanning app version table at cluster!user! scan . setStartRow ( startRow ) ; // require that all results match this flow prefix FilterList filters = new FilterList ( FilterList . Operator . MUST_PASS_ALL ) ; filters . addFilter ( new WhileMatchFilter ( new PrefixFilter ( startRow ) ) ) ; scan . setFilter ( filters ) ; List < AppKey > newAppsKeys = new ArrayList < AppKey > ( ) ; try { newAppsKeys = createNewAppKeysFromResults ( scan , startTime , endTime , limit ) ; } catch ( IOException e ) { LOG . error ( "Caught exception while trying to scan, returning empty list of flows: " + e . toString ( ) ) ; } List < AppSummary > newApps = new ArrayList < AppSummary > ( ) ; for ( AppKey ak : newAppsKeys ) { AppSummary anApp = new AppSummary ( ak ) ; List < Flow > flows = jhs . getFlowSeries ( ak . getCluster ( ) , ak . getUserName ( ) , ak . getAppId ( ) , null , Boolean . FALSE , startTime , endTime , Integer . MAX_VALUE ) ; for ( Flow f : flows ) { anApp . addFlow ( f ) ; } newApps . add ( anApp ) ; } return newApps ; }
scans the app version table to look for jobs that showed up in the given time range creates the flow key that maps to these apps