protected void initialize ( ) throws IOException { for ( int i = 0 ; i < 3 ; i ++ ) { try { LOG . info ( "Detecting current primary node - attempt " + i ) ; detectJournalManager ( ) ; LOG . info ( "Finding oldest segment txid - attempt " + i ) ; currentSegmentTxId = findOldestLogSegmentTxid ( ) ; LOG . info ( "Setting up input stream for txid: " + currentSegmentTxId + " - attempt " + i ) ; setupIngestStreamWithRetries ( currentSegmentTxId ) ; return ; } catch ( IOException e ) { LOG . warn ( "Initialization exception" , e ) ; if ( i == 2 ) { LOG . error ( "Initialization failed." ) ; throw e ; } } } }
Initialize first stream . Just for startup we are extra careful to try 3 times as there is potential race between finding the txid and then setting up the stream .