private void refreshInputStream ( ) throws IOException { // if stream is null, we probably switched to a new // segment. if ( currentEditLogInputStream == null ) { LOG . info ( "Segment - setup input stream for txid: " + currentSegmentTxId ) ; setupIngestStreamWithRetries ( currentSegmentTxId ) ; if ( currentEditLogInputStreamPosition == - 1 ) { // we are opening a fresh segment currentEditLogInputStreamPosition = currentEditLogInputStream . getPosition ( ) ; } } // if we are re-opening stream previously consumed // set correct position if ( currentEditLogInputStreamPosition != - 1 ) { currentEditLogInputStream . refresh ( currentEditLogInputStreamPosition , mostRecentlyReadTransactionTxId ) ; } }
On normal activity when we reach END_LOG_SEGMENT we will null the stream and at next read we want to instantiate a stream for next segment .