question: public void run ( long jobsToComplete ) throws AuditLogEntryException , IOException , StateUpdateException , SQLException { clearCounters ( ) ; long lastPersistedAuditLogId = 0 ; if ( startAfterAuditLogId . isPresent ( ) ) { lastPersistedAuditLogId = startAfterAuditLogId . get ( ) ; } else { LOG . debug ( "Fetching last persisted audit log ID" ) ; Optional < String > lastPersistedIdString = keyValueStore . get ( LAST_PERSISTED_AUDIT_LOG_ID_KEY ) ; if ( ! lastPersistedIdString . isPresent ( ) ) { Optional < Long > maxId = auditLogReader . getMaxId ( ) ; lastPersistedAuditLogId = maxId . orElse ( Long . valueOf ( 0 ) ) ; LOG . warn ( String . format ( "Since the last persisted ID was not " + "previously set, using max ID in the audit log: %s" , lastPersistedAuditLogId ) ) ; } else { lastPersistedAuditLogId = Long . parseLong ( lastPersistedIdString . get ( ) ) ; } } LOG . info ( "Using last persisted ID of " + lastPersistedAuditLogId ) ; auditLogReader . setReadAfterId ( lastPersistedAuditLogId ) ; for ( PersistedJobInfo jobInfo : jobInfoStore . getRunnableFromDb ( ) ) { LOG . debug ( String . format ( "Restoring %s to (re)run" , jobInfo ) ) ; ReplicationJob job = restoreReplicationJob ( jobInfo ) ; prettyLogStart ( job ) ; jobRegistry . registerJob ( job ) ; queueJobForExecution ( job ) ; } TimeZone tz = TimeZone . getTimeZone ( "UTC" ) ; DateFormat df = new SimpleDateFormat ( "yyyy-MM-dd'T'HH:mm'Z'" ) ; df . setTimeZone ( tz ) ; statsTracker . start ( ) ; long updateTimeForLastPersistedId = 0 ; while ( true ) { if ( pauseRequested ) { LOG . debug ( "Pause requested. Sleeping..." ) ; ReplicationUtils . sleep ( pollWaitTimeMs ) ; continue ; } long completedJobs = counters . getCounter ( ReplicationCounters . Type . SUCCESSFUL_TASKS ) + counters . getCounter ( ReplicationCounters . Type . NOT_COMPLETABLE_TASKS ) ; if ( jobsToComplete > 0 && completedJobs >= jobsToComplete ) { LOG . debug ( String . format ( "Hit the limit for the number of " + "successful jobs (%d) - returning." , jobsToComplete ) ) ; return ; } if ( jobExecutor . getNotDoneJobCount ( ) > maxJobsInMemory ) { LOG . debug ( String . format ( "There are too many jobs in memory. " + "Waiting until more complete. (limit: %d)" , maxJobsInMemory ) ) ; ReplicationUtils . sleep ( pollWaitTimeMs ) ; continue ; } LOG . debug ( "Fetching the next entry from the audit log" ) ; Optional < AuditLogEntry > auditLogEntry = auditLogReader . resilientNext ( ) ; if ( ! auditLogEntry . isPresent ( ) ) { LOG . debug ( String . format ( "No more entries from the audit log. " + "Sleeping for %s ms" , pollWaitTimeMs ) ) ; ReplicationUtils . sleep ( pollWaitTimeMs ) ; continue ; } AuditLogEntry entry = auditLogEntry . get ( ) ; LOG . debug ( "Got audit log entry: " + entry ) ; List < ReplicationJob > replicationJobs = jobFactory . createReplicationJobs ( auditLogEntry . get ( ) , replicationFilters ) ; LOG . debug ( String . format ( "Audit log entry id: %s converted to %s" , entry . getId ( ) , replicationJobs ) ) ; for ( ReplicationJob job : replicationJobs ) { jobRegistry . registerJob ( job ) ; } if ( System . currentTimeMillis ( ) - updateTimeForLastPersistedId > 10000 ) { keyValueStore . resilientSet ( LAST_PERSISTED_AUDIT_LOG_ID_KEY , Long . toString ( entry . getId ( ) ) ) ; updateTimeForLastPersistedId = System . currentTimeMillis ( ) ; } for ( ReplicationJob replicationJob : replicationJobs ) { LOG . debug ( "Scheduling: " + replicationJob ) ; prettyLogStart ( replicationJob ) ; long tasksSubmittedForExecution = counters . getCounter ( ReplicationCounters . Type . EXECUTION_SUBMITTED_TASKS ) ; if ( tasksSubmittedForExecution >= jobsToComplete ) { LOG . warn ( String . format ( "Not submitting %s for execution " + " due to the limit for the number of " + "jobs to execute" , replicationJob ) ) ; continue ; } else { queueJobForExecution ( replicationJob ) ; } } } }
target:  Start reading the audit log and replicate entries.