question: @ 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 ) ; }
target:  Implements read loop on the log to feed event queue.