public void pollEvent ( DFSInotifyEventInputStream eventStream ) { EventBatch batch ; LOG . debug ( "Polling thread starting, with timeout {} ms" , mActiveUfsPollTimeoutMs ) ; int count = 0 ; long start = System . currentTimeMillis ( ) ; long behind = eventStream . getTxidsBehindEstimate ( ) ; while ( ! Thread . currentThread ( ) . isInterrupted ( ) ) { try { batch = eventStream . poll ( mActiveUfsPollTimeoutMs , TimeUnit . MILLISECONDS ) ; if ( batch != null ) { long txId = batch . getTxid ( ) ; count ++ ; for ( Event event : batch . getEvents ( ) ) { processEvent ( event , mUfsUriList , txId ) ; } } long end = System . currentTimeMillis ( ) ; if ( end > ( start + mActiveUfsSyncEventRateInterval ) ) { long currentlyBehind = eventStream . getTxidsBehindEstimate ( ) ; LOG . info ( "HDFS generated {} events in {} ms, at a rate of {} rps" , count + currentlyBehind - behind , end - start , String . format ( "%.2f" , ( count + currentlyBehind - behind ) * 1000.0 / ( end - start ) ) ) ; LOG . info ( "processed {} events in {} ms, at a rate of {} rps" , count , end - start , String . format ( "%.2f" , count * 1000.0 / ( end - start ) ) ) ; LOG . info ( "Currently TxidsBehindEstimate by {}" , currentlyBehind ) ; behind = currentlyBehind ; start = end ; count = 0 ; } } catch ( IOException e ) { LOG . warn ( "IOException occured during polling inotify {}" , e ) ; if ( e . getCause ( ) instanceof InterruptedException ) { return ; } } catch ( MissingEventsException e ) { LOG . warn ( "MissingEventException during polling {}" , e ) ; mEventMissed = true ; // need to sync all syncpoints at this point } catch ( InterruptedException e ) { LOG . warn ( "InterruptedException during polling {}" , e ) ; return ; } } }
Fetch and process events .