protected void createWorkUnits ( SourceState state ) throws IOException { createWorkunitsFromPreviousState ( state ) ; if ( this . datasets . isEmpty ( ) ) { return ; } for ( HivePartitionDataset dataset : this . datasets ) { Optional < String > owner = dataset . getOwner ( ) ; if ( workUnitsExceeded ( ) ) { log . info ( "Workunits exceeded" ) ; setJobWatermark ( state , dataset . datasetURN ( ) ) ; return ; } if ( ! this . policy . shouldPurge ( dataset ) ) { continue ; } WorkUnit workUnit = createNewWorkUnit ( dataset ) ; log . info ( "Created new work unit with partition " + workUnit . getProp ( ComplianceConfigurationKeys . PARTITION_NAME ) ) ; this . workUnitMap . put ( workUnit . getProp ( ComplianceConfigurationKeys . PARTITION_NAME ) , workUnit ) ; this . workUnitsCreatedCount ++ ; } if ( ! state . contains ( ComplianceConfigurationKeys . HIVE_PURGER_WATERMARK ) ) { this . setJobWatermark ( state , ComplianceConfigurationKeys . NO_PREVIOUS_WATERMARK ) ; } }
This method creates the list of all work units needed for the current execution . Fresh work units are created for each partition starting from watermark and failed work units from the previous run will be added to the list .