@ Override public void selectInputStreams ( Collection < EditLogInputStream > streams , long fromTxnId , boolean inProgressOk , boolean validateInProgressSegments ) throws IOException { QuorumCall < AsyncLogger , RemoteEditLogManifest > q = loggers . getEditLogManifest ( fromTxnId ) ; // we insist on getting all responses, even if they are to be exceptions // this will fail if we cannot get majority of successes Map < AsyncLogger , RemoteEditLogManifest > resps = loggers . waitForReadQuorumWithAllResponses ( q , selectInputStreamsTimeoutMs , "selectInputStreams" ) ; if ( LOG . isDebugEnabled ( ) ) { LOG . debug ( "selectInputStream manifests:\n" + Joiner . on ( "\n" ) . withKeyValueSeparator ( ": " ) . join ( resps ) ) ; } final PriorityQueue < EditLogInputStream > allStreams = new PriorityQueue < EditLogInputStream > ( 64 , JournalSet . EDIT_LOG_INPUT_STREAM_COMPARATOR ) ; for ( Map . Entry < AsyncLogger , RemoteEditLogManifest > e : resps . entrySet ( ) ) { AsyncLogger logger = e . getKey ( ) ; RemoteEditLogManifest manifest = e . getValue ( ) ; for ( RemoteEditLog remoteLog : manifest . getLogs ( ) ) { EditLogInputStream elis = new URLLogInputStream ( logger , remoteLog . getStartTxId ( ) , httpConnectReadTimeoutMs ) ; if ( elis . isInProgress ( ) && ! inProgressOk ) { continue ; } allStreams . add ( elis ) ; } } // we pass 0 as min redundance as we do not care about this here JournalSet . chainAndMakeRedundantStreams ( streams , allStreams , fromTxnId , inProgressOk , 0 ) ; }
Select input streams . inProgressOk should be true only for tailing not for startup