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);
      }
    }
  }
}

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 ) ; } } } }
start reading the audit log and replicate entries .
