public ReplDBMSEvent filter(ReplDBMSEvent event) throws ReplicatorException, InterruptedException {
  totalEvents++;
  Timestamp sourceTstamp=event.getExtractedTstamp();
  long currentTime=System.currentTimeMillis();
  prefetchLatency=currentTime - sourceTstamp.getTime();
  if (logger.isDebugEnabled()) {
    logger.debug("Processing event: seqno=" + event.getSeqno() + " prefetchLatency="+ prefetchLatency);
  }
  if (lastChecked == 0 || (currentTime - lastChecked >= interval)) {
    getCurrentSlaveHeader();
  }
  if (allowAll) {
    prefetchEvents++;
    return event;
  }
  if (event.getSeqno() <= slaveSeqno) {
    if (logger.isDebugEnabled()) {
      logger.debug("Discarding event already executed by slave seqno=" + event.getSeqno() + " slaveSeqno="+ slaveSeqno);
    }
    return null;
  }
  if ((slaveLatency - prefetchLatency) < aheadMinMillis) {
    if (logger.isDebugEnabled()) {
      logger.debug("Discarding event that is under minimum prefetch latency: seqno=" + event.getSeqno() + " prefetchLatency="+ prefetchLatency+ " slaveLatency="+ slaveLatency);
    }
    return null;
  }
  long aheadMillis=0;
  long originalSlaveLatency=slaveLatency;
  while ((aheadMillis=originalSlaveLatency - prefetchLatency) > aheadMaxMillis) {
    if (logger.isDebugEnabled())     logger.debug("Event is too far ahead of current slave position: aheadMillis=" + aheadMillis);
    long sleepStartMillis=System.currentTimeMillis();
    try {
      prefetchState=PrefetchState.sleeping;
      Thread.sleep(sleepTime);
    }
  finally {
      prefetchState=PrefetchState.active;
      sleepTimeMillis+=(System.currentTimeMillis() - sleepStartMillis);
    }
    currentTime=System.currentTimeMillis();
    prefetchLatency=currentTime - sourceTstamp.getTime();
    getCurrentSlaveHeader();
    if (event.getSeqno() <= slaveSeqno) {
      if (logger.isDebugEnabled()) {
        logger.debug("Discarding event already executed by slave seqno=" + event.getSeqno() + " slaveSeqno="+ slaveSeqno);
      }
      return null;
    }
  }
  prefetchEvents++;
  if (logger.isDebugEnabled() && totalEvents % 20000 == 0)   logger.debug("Prefetched " + prefetchEvents + " events - Ratio "+ (100 * prefetchEvents / totalEvents)+ "%");
  return event;
}

public ReplDBMSEvent filter ( ReplDBMSEvent event ) throws ReplicatorException , InterruptedException { totalEvents ++ ; Timestamp sourceTstamp = event . getExtractedTstamp ( ) ; long currentTime = System . currentTimeMillis ( ) ; prefetchLatency = currentTime - sourceTstamp . getTime ( ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Processing event: seqno=" + event . getSeqno ( ) + " prefetchLatency=" + prefetchLatency ) ; } if ( lastChecked == 0 || ( currentTime - lastChecked >= interval ) ) { getCurrentSlaveHeader ( ) ; } if ( allowAll ) { prefetchEvents ++ ; return event ; } if ( event . getSeqno ( ) <= slaveSeqno ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Discarding event already executed by slave seqno=" + event . getSeqno ( ) + " slaveSeqno=" + slaveSeqno ) ; } return null ; } if ( ( slaveLatency - prefetchLatency ) < aheadMinMillis ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Discarding event that is under minimum prefetch latency: seqno=" + event . getSeqno ( ) + " prefetchLatency=" + prefetchLatency + " slaveLatency=" + slaveLatency ) ; } return null ; } long aheadMillis = 0 ; long originalSlaveLatency = slaveLatency ; while ( ( aheadMillis = originalSlaveLatency - prefetchLatency ) > aheadMaxMillis ) { if ( logger . isDebugEnabled ( ) ) logger . debug ( "Event is too far ahead of current slave position: aheadMillis=" + aheadMillis ) ; long sleepStartMillis = System . currentTimeMillis ( ) ; try { prefetchState = PrefetchState . sleeping ; Thread . sleep ( sleepTime ) ; } finally { prefetchState = PrefetchState . active ; sleepTimeMillis += ( System . currentTimeMillis ( ) - sleepStartMillis ) ; } currentTime = System . currentTimeMillis ( ) ; prefetchLatency = currentTime - sourceTstamp . getTime ( ) ; getCurrentSlaveHeader ( ) ; if ( event . getSeqno ( ) <= slaveSeqno ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Discarding event already executed by slave seqno=" + event . getSeqno ( ) + " slaveSeqno=" + slaveSeqno ) ; } return null ; } } prefetchEvents ++ ; if ( logger . isDebugEnabled ( ) && totalEvents % 20000 == 0 ) logger . debug ( "Prefetched " + prefetchEvents + " events - Ratio " + ( 100 * prefetchEvents / totalEvents ) + "%" ) ; return event ; }
filter the event if it has already been executed .
