@Override public void run(){
  long readSeqno=restartSeqno;
  try {
    if (!connection.seek(restartSeqno)) {
      throw new THLException("Unable to locate starting seqno in log: seqno=" + restartSeqno + " store="+ thl.getName()+ " taskId="+ taskId);
    }
    while (!cancelled) {
      THLEvent thlEvent=connection.next();
      readSeqno=thlEvent.getSeqno();
      if (lowWaterMark.get() == 0)       lowWaterMark.set(readSeqno);
      readCount.incrementAndGet();
      if (logger.isDebugEnabled()) {
        logger.debug("Read event from THL: seqno=" + thlEvent.getSeqno() + " fragno="+ thlEvent.getFragno()+ " lastFrag="+ thlEvent.getLastFrag()+ " deserialized="+ (thlEvent.getReplEvent() != null));
      }
      headSeqnoCounter.waitSeqnoGreaterEqual(thlEvent.getSeqno());
      if (logger.isDebugEnabled()) {
        logger.debug("Adding event to parallel queue:  taskId=" + taskId + " seqno="+ thlEvent.getSeqno()+ " fragno="+ thlEvent.getFragno());
      }
      readQueue.post(thlEvent);
    }
  }
 catch (  InterruptedException e) {
    if (!cancelled)     logger.warn("Unexpected interrupt before reader thread was cancelled");
  }
catch (  Throwable e) {
    throwable=e;
    try {
      String msg="Read failed on transaction log: seqno=" + readSeqno + " taskId="+ taskId;
      logger.error(msg,e);
      dispatcher.put(new ErrorNotification(msg,e));
    }
 catch (    InterruptedException e1) {
      logger.warn("Task cancelled while posting error notification",null);
    }
catch (    Throwable t1) {
      logger.warn("Failure while attempting to log an error: " + e,t1);
    }
  }
  logger.info("Terminating parallel reader thread: seqno=" + readSeqno + " store="+ thl.getName()+ " taskId="+ taskId);
}

@ Override public void run ( ) { long readSeqno = restartSeqno ; try { if ( ! connection . seek ( restartSeqno ) ) { throw new THLException ( "Unable to locate starting seqno in log: seqno=" + restartSeqno + " store=" + thl . getName ( ) + " taskId=" + taskId ) ; } while ( ! cancelled ) { THLEvent thlEvent = connection . next ( ) ; readSeqno = thlEvent . getSeqno ( ) ; if ( lowWaterMark . get ( ) == 0 ) lowWaterMark . set ( readSeqno ) ; readCount . incrementAndGet ( ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Read event from THL: seqno=" + thlEvent . getSeqno ( ) + " fragno=" + thlEvent . getFragno ( ) + " lastFrag=" + thlEvent . getLastFrag ( ) + " deserialized=" + ( thlEvent . getReplEvent ( ) != null ) ) ; } headSeqnoCounter . waitSeqnoGreaterEqual ( thlEvent . getSeqno ( ) ) ; if ( logger . isDebugEnabled ( ) ) { logger . debug ( "Adding event to parallel queue:  taskId=" + taskId + " seqno=" + thlEvent . getSeqno ( ) + " fragno=" + thlEvent . getFragno ( ) ) ; } readQueue . post ( thlEvent ) ; } } catch ( InterruptedException e ) { if ( ! cancelled ) logger . warn ( "Unexpected interrupt before reader thread was cancelled" ) ; } catch ( Throwable e ) { throwable = e ; try { String msg = "Read failed on transaction log: seqno=" + readSeqno + " taskId=" + taskId ; logger . error ( msg , e ) ; dispatcher . put ( new ErrorNotification ( msg , e ) ) ; } catch ( InterruptedException e1 ) { logger . warn ( "Task cancelled while posting error notification" , null ) ; } catch ( Throwable t1 ) { logger . warn ( "Failure while attempting to log an error: " + e , t1 ) ; } } logger . info ( "Terminating parallel reader thread: seqno=" + readSeqno + " store=" + thl . getName ( ) + " taskId=" + taskId ) ; }
implements read loop on the log to feed event queue .
