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);
    }
  }
private void refreshInputStream ( ) throws IOException { if ( currentEditLogInputStream == null ) { LOG . info ( "Segment - setup input stream for txid: " + currentSegmentTxId ) ; setupIngestStreamWithRetries ( currentSegmentTxId ) ; if ( currentEditLogInputStreamPosition == - 1 ) { currentEditLogInputStreamPosition = currentEditLogInputStream . getPosition ( ) ; } } 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.

This function repositions the stream to the last correct known position as
well.