Read a single transaction from the input edit log
@param inputEditLog the log to read from
@return a single edit log entry, null if EOF
@throws IOException on error
513374
private FSEditLogOp ingestFSEdit(EditLogInputStream inputEditLog)
      throws IOException {
    FSEditLogOp op = null;
    try {
      op = inputEditLog.readOp();
      InjectionHandler.processEventIO(InjectionEvent.INGEST_READ_OP);
    } catch (EOFException e) {
      return null; // No more transactions.
    } catch (IOException e) {
      // rethrow, it's handled in ingestFSEdits()
      throw e;
    } catch (Exception e) {
      // some other problem, maybe unchecked exception
      throw new IOException(e);
    }
    return op;
  }
369277
public static String getInput(String prompt) {
        System.out.print(prompt);

        try {
            BufferedReader in =
                new BufferedReader(new InputStreamReader(System.in));
            return in.readLine();
        } catch(IOException e) {
            return null;
        }
    }
708692
private LogChunkType readNextChunk()
    {
        // clear the current chunk
        currentChunk = Slices.EMPTY_SLICE;

        // read the next block if necessary
        if (currentBlock.available() < HEADER_SIZE) {
            if (!readNextBlock()) {
                if (eof) {
                    return EOF;
                }
            }
        }

        // parse header
        int expectedChecksum = currentBlock.readInt();
        int length = currentBlock.readUnsignedByte();
        length = length | currentBlock.readUnsignedByte() << 8;
        byte chunkTypeId = currentBlock.readByte();
        LogChunkType chunkType = getLogChunkTypeByPersistentId(chunkTypeId);

        // verify length
        if (length > currentBlock.available()) {
            int dropSize = currentBlock.available() + HEADER_SIZE;
            reportCorruption(dropSize, "Invalid chunk length");
            currentBlock = Slices.EMPTY_SLICE.input();
            return BAD_CHUNK;
        }

        // skip zero length records
        if (chunkType == ZERO_TYPE && length == 0) {
            // Skip zero length record without reporting any drops since
            // such records are produced by the writing code.
            currentBlock = Slices.EMPTY_SLICE.input();
            return BAD_CHUNK;
        }

        // Skip physical record that started before initialOffset
        if (endOfBufferOffset - HEADER_SIZE - length < initialOffset) {
            currentBlock.skipBytes(length);
            return BAD_CHUNK;
        }

        // read the chunk
        currentChunk = currentBlock.readBytes(length);

        if (verifyChecksums) {
            int actualChecksum = getChunkChecksum(chunkTypeId, currentChunk);
            if (actualChecksum != expectedChecksum) {
                // Drop the rest of the buffer since "length" itself may have
                // been corrupted and if we trust it, we could find some
                // fragment of a real log record that just happens to look
                // like a valid log record.
                int dropSize = currentBlock.available() + HEADER_SIZE;
                currentBlock = Slices.EMPTY_SLICE.input();
                reportCorruption(dropSize, "Invalid chunk checksum");
                return BAD_CHUNK;
            }
        }

        // Skip unknown chunk types
        // Since this comes last so we the, know it is a valid chunk, and is just a type we don't understand
        if (chunkType == UNKNOWN) {
            reportCorruption(length, String.format("Unknown chunk type %d", chunkType.getPersistentId()));
            return BAD_CHUNK;
        }

        return chunkType;
    }
471291
public int read() throws IOException {
    if (this.bytesRead == this.length) {
      // All chunks and final additional chunk are read.
      // This means we have reached EOF.
      return -1;
    }

    try {
      // Read a chunk from given input stream when
      // it is first chunk or all bytes in chunk body is read
      if (this.streamBytesRead == 0 || this.chunkPos == this.chunkBody.length) {
        // Check if there are data available to read from given input stream.
        if (this.streamBytesRead != this.streamSize) {
          // Send all data chunks.
          int chunkSize = CHUNK_SIZE;
          if (this.streamBytesRead + chunkSize > this.streamSize) {
            chunkSize = this.streamSize - this.streamBytesRead;
          }

          if (readChunk(chunkSize) < 0) {
            return -1;
          }

          this.streamBytesRead += chunkSize;
        } else {
          // Send final additional chunk to complete chunk upload.
          byte[] chunk = new byte[0];
          createChunkBody(chunk);
        }
      }

      this.bytesRead++;
      // Value must be between 0 to 255.
      int value = this.chunkBody[this.chunkPos] & 0xFF;
      this.chunkPos++;
      return value;
    } catch (NoSuchAlgorithmException | InvalidKeyException | InsufficientDataException | InternalException e) {
      throw new IOException(e.getCause());
    }
  }
496467
@Override
  public void loadEdits() throws IOException {

    try {
      v.start();
      v.visitEnclosingElement(EditsElement.EDITS);

      IntToken editsVersionToken = v.visitInt(EditsElement.EDITS_VERSION);
      editsVersion = editsVersionToken.value;
      if(!canLoadVersion(editsVersion)) {
        throw new IOException("Cannot process editLog version " +
          editsVersionToken.value);
      }

      FSEditLogOpCodes editsOpCode;
      do {
        v.visitEnclosingElement(EditsElement.RECORD);

        ByteToken opCodeToken;
        try {
          opCodeToken = v.visitByte(EditsElement.OPCODE);
        } catch (EOFException eof) {
          // Getting EOF when reading the opcode is fine --
          // it's just a finalized edits file
          // Just fake the OP_INVALID here.
          opCodeToken = new ByteToken(EditsElement.OPCODE);
          opCodeToken.fromByte(FSEditLogOpCodes.OP_INVALID.getOpCode());
          v.visit(opCodeToken);
        }
        editsOpCode = FSEditLogOpCodes.fromByte(opCodeToken.value);

        v.visitEnclosingElement(EditsElement.DATA);

        visitOpCode(editsOpCode);

        v.leaveEnclosingElement(); // DATA
        
        if (editsOpCode != FSEditLogOpCodes.OP_INVALID && 
            LayoutVersion.supports(Feature.EDITS_CHESKUM, editsVersion)) {
          v.visitInt(EditsElement.CHECKSUM);
        }
        v.leaveEnclosingElement(); // RECORD
      } while(editsOpCode != FSEditLogOpCodes.OP_INVALID);

      v.leaveEnclosingElement(); // EDITS
      v.finish();
    } catch(IOException e) {
      // Tell the visitor to clean up, then re-throw the exception
      v.finishAbnormally();
      throw e;
    }
  }
255825
private Journal.JournalEntry readInternal() throws IOException {
    if (mInputStream == null) {
      return null;
    }
    JournalEntry entry = mInputStream.mReader.readEntry();
    if (entry != null) {
      return entry;
    }
    if (mInputStream.mFile.isIncompleteLog()) {
      // Incomplete logs may end early.
      return null;
    } else {
      Preconditions.checkState(mInputStream.mFile.isCompletedLog(),
          "Expected log to be either checkpoint, incomplete, or complete");
      ProcessUtils.fatalError(LOG, "Journal entry %s was truncated", mNextSequenceNumber);
      return null;
    }
  }
296079
@Override
    public synchronized int read(final byte buf[], final int off, final int len)
            throws IOException {
        while (this.currentRecord.needsData()) {
            final byte[] data = new byte[len];
            final int read = this.inputStream.read(data);
            if (read == -1) {
                return read;
            }
            lastBuffer = ByteBuffer.wrap(data, 0, read);
            this.currentRecord.addData(lastBuffer);
        }
        final int bytesRead = this.currentRecord.drainData(buf, off, len);
        if (!this.currentRecord.hasMoreData()) {
            LOG.info("Resetting app record");
            this.currentRecord = new InputRecord(readKey);
            if (lastBuffer != null) {
                this.currentRecord.addData(lastBuffer);
            } else {
                LOG.warn("No existing buffer?");
            }
        }
        return bytesRead;
    }
191376
synchronized void startLogSegment(final long segmentTxId,
      boolean writeHeaderTxn) throws IOException {
    LOG.info("Starting log segment at " + segmentTxId);
    if (segmentTxId < 0) {
      throw new IOException("Bad txid: " + segmentTxId);
    }
    if (state != State.BETWEEN_LOG_SEGMENTS) {
      throw new IOException("Bad state: " + state);
    }
    if (segmentTxId <= curSegmentTxId) {
      throw new IOException("Cannot start writing to log segment "
          + segmentTxId + " when previous log segment started at "
          + curSegmentTxId);
    }
    if (segmentTxId != txid + 1) {
      throw new IOException("Cannot start log segment at txid " + segmentTxId
          + " when next expected " + (txid + 1));
    }

    numTransactions = totalTimeTransactions = numTransactionsBatchedInSync = 0;

    // TODO no need to link this back to storage anymore!
    // See HDFS-2174.
    storage.attemptRestoreRemovedStorage();
    try {
      editLogStream = journalSet.startLogSegment(segmentTxId);
    } catch (IOException ex) {
      throw new IOException("Unable to start log segment " + segmentTxId
          + ": no journals successfully started.");

    }
    curSegmentTxId = segmentTxId;
    state = State.IN_SEGMENT;
    if (writeHeaderTxn) {
      logEdit(LogSegmentOp.getInstance(FSEditLogOpCodes.OP_START_LOG_SEGMENT));
      logSync();
    }
    
    // force update of journal and image metrics
    journalSet.updateJournalMetrics();
    // If it is configured, we want to schedule an automatic edits roll
    if (timeoutRollEdits > 0) {
      FSNamesystem fsn = this.journalSet.getImage().getFSNamesystem();
      if (fsn != null) {
        // In some test cases fsn is NULL in images. Simply skip the feature.
        AutomaticEditsRoller aer = fsn.automaticEditsRoller;
        if (aer != null) {
          aer.setNextRollTime(System.currentTimeMillis() + timeoutRollEdits);
        } else {
          LOG.warn("Automatic edits roll is enabled but the roller thread "
              + "is not enabled. Should only happen in unit tests.");
        }
      } else {
        LOG.warn("FSNamesystem is NULL in FSEditLog.");
      }
    }
  }
1310392
public String readLineUntilBoundary() throws IOException {
		StringBuilder bldr = new StringBuilder();
		while (true) {
			int b = read();
			if (bldr.indexOf(boundary) == 0) {
				leftover = ByteBuffer.wrap(boundaryBytes);
				return null;
			} else if (b == -1 || b == 10) {
				break;
			} else if (b != '\r') {
				bldr.append((char) ((byte) b));
			}
		}
		return bldr.toString();
	}
197116
private String readLine() throws IOException {
        String line = input.readLine();
        lineNumber = lineNumber + 1;
        logger.debug("LINE (" + lineNumber + "): ", line);
        return line;
    }
814043
@Override
    public final int read() throws IOException
    {
        assert cursor <= end;
        if (cursor >= end)
        {
            if (includeLevel.in == null)
            {
                return -1;
            }
            int cp = (int)(cursor % size);
            long len = size-(cursor-waterMark);
            int il;
            if (len > size - cp)
            {
                buffer1.position(cp);
                buffer1.limit(size);
                buffer2.position(0);
                buffer2.limit((int)(len-(size-cp)));
                if (!buffer1.hasRemaining() && !buffer2.hasRemaining())
                {
                    throw new UnderflowException("Buffer size="+size+" too small for operation");
                }
                il = fill(includeLevel.in, array2);
            }
            else
            {
                buffer1.position(cp);
                buffer1.limit((int)(cp+len));
                if (!buffer1.hasRemaining())
                {
                    throw new UnderflowException("Buffer size="+size+" too small for operation");
                }
                il = fill(includeLevel.in, array1);
            }
            if (il == -1)
            {
                if (includeStack != null)
                {
                    while (!includeStack.isEmpty() && il == -1)
                    {
                        close(includeLevel.in);
                        includeLevel = includeStack.pop();
                        return read();
                    }
                }
                return -1;
            }
            if (il == 0)
            {
                throw new IOException("No input! Use blocking mode?");
            }
            buffer1.clear();
            buffer2.clear();
            end+=il;
            if (end < 0)
            {
                throw new IOException("end = "+end);
            }
        }
        int rc = get(cursor++);
        if (cursor < 0)
        {
            throw new IOException("cursor = "+cursor);
        }
        includeLevel.forward(rc);
        length++;
        if (length > size)
        {
            throw new IOException("input size "+length+" exceeds buffer size "+size);
        }
        if (checksum != null)
        {
            checksum.update(cursor-1, rc);
        }
        return rc;
    }
633872
@SuppressWarnings("unchecked")
  private void consume(LogRecord record) {
    // Decode the raw log operation from the record.
    LogOperation operation = decodeInternal(record.value());

    // If this operation is not destined for this primitive, ignore it.
    // TODO: If multiple primitives of different types are created and destroyed on the same distributed log,
    // we need to be able to differentiate between different instances of a service by the service ID.
    if (!operation.primitive().equals(name())) {
      return;
    }

    // Create a session from the log record.
    Session session = getOrCreateSession(operation.sessionId());

    // Update the local context for the service.
    currentIndex = record.index();
    currentSession = session;
    currentOperation = operation.operationId().type();
    currentTimestamp = record.timestamp();

    // Apply the operation to the service.
    byte[] output = service.apply(new DefaultCommit<>(
        currentIndex,
        operation.operationId(),
        operation.operation(),
        currentSession,
        currentTimestamp));

    // If the operation session matches the local session, complete the write future.
    if (operation.sessionId().equals(this.session.sessionId())) {
      CompletableFuture future = writeFutures.remove(operation.operationIndex());
      if (future != null) {
        future.complete(decode(output));
      }
    }

    // Iterate through pending reads and complete any reads at indexes less than or equal to the applied index.
    PendingRead pendingRead = pendingReads.peek();
    while (pendingRead != null && pendingRead.index <= record.index()) {
      session = getOrCreateSession(this.session.sessionId());
      currentSession = session;
      currentOperation = OperationType.QUERY;
      try {
        output = service.apply(new DefaultCommit<>(
            currentIndex,
            pendingRead.operationId,
            pendingRead.bytes,
            session,
            currentTimestamp));
        pendingRead.future.complete(output);
      } catch (Exception e) {
        pendingRead.future.completeExceptionally(new PrimitiveException.ServiceException());
      }
      pendingReads.remove();
      pendingRead = pendingReads.peek();
    }
  }
868437
private String readTrimmedLine() throws IOException {
        String line = readLine();
        if (line != null)
            return line.trim();
        else
            throw new EOFException();
    }
1165452
int ingestFSEdits() throws IOException {
    FSDirectory fsDir = fsNamesys.dir;
    int numEdits = 0;
    
    long recentOpcodeOffsets[] = new long[2];
    Arrays.fill(recentOpcodeOffsets, -1);
    EnumMap<FSEditLogOpCodes, Holder<Integer>> opCounts =
        new EnumMap<FSEditLogOpCodes, Holder<Integer>>(FSEditLogOpCodes.class);
    
    boolean error = false;
    boolean reopen = false;    
    boolean quitAfterScan = false;
    
    long sharedLogTxId = FSEditLogLoader.TXID_IGNORE;
    long localLogTxId = FSEditLogLoader.TXID_IGNORE;

    FSEditLogOp op = null;
    FSEditLog localEditLog = fsDir.fsImage.getEditLog();

    while (running && !quitAfterScan) {      
      // if the application requested that we make a final pass over 
      // the transaction log, then we remember it here. We close and
      // reopen the file to ensure that we can see all the data in the
      // file, one reason being that NFS has open-to-close cache
      // coherancy and the edit log could be stored in NFS.
      //
      if (reopen || lastScan) {
        inputEditStream.close();
        inputEditStream = standby.setupIngestStreamWithRetries(startTxId);
        if (lastScan) {
          // QUIESCE requested by Standby thread
          LOG.info("Ingest: Starting last scan of transaction log: " + this.toString());
          quitAfterScan = true;
        }

        // discard older buffers and start a fresh one.
        inputEditStream.refresh(currentPosition, localEditLog.getLastWrittenTxId());       
        setCatchingUp();
        reopen = false;
      }

      //
      // Process all existing transactions till end of file
      //
      while (running) {
        
        if (lastScan && !quitAfterScan) {
          // Standby thread informed the ingest to quiesce
          // we should refresh the input stream as soon as possible
          // then quitAfterScan will be true
          break;
        }
        
        // record the current file offset.
        currentPosition = inputEditStream.getPosition(); 
        InjectionHandler.processEvent(InjectionEvent.INGEST_BEFORE_LOAD_EDIT);

        fsNamesys.writeLock();
        try {
          error = false;
          op = ingestFSEdit(inputEditStream);
          
          /*
           * In the case of segments recovered on primary namenode startup, we
           * have segments that are finalized (by name), but not containing the
           * ending transaction. Without this check, we will keep looping until
           * the next checkpoint to discover this situation.
           */
          if (!inputEditStream.isInProgress()
              && standby.getLastCorrectTxId() == inputEditStream.getLastTxId()) {
            // this is a correct segment with no end segment transaction
            LOG.info("Ingest: Reached finalized log segment end with no end marker. "
                + this.toString());
            tearDown(localEditLog, false, true);
            break;
          }

          if (op == null) {
            FSNamesystem.LOG.debug("Ingest: Invalid opcode, reached end of log " +
                                   "Number of transactions found " + 
                                   numEdits);
            break; // No more transactions.
          }
          
          sharedLogTxId = op.txid;
          // Verify transaction ids match.
          localLogTxId = localEditLog.getLastWrittenTxId() + 1;
          // Fatal error only when the log contains transactions from the future
          // we allow to process a transaction with smaller txid than local
          // we will simply skip it later after reading from the ingest edits
          if (localLogTxId < sharedLogTxId
              || InjectionHandler
                  .falseCondition(InjectionEvent.INGEST_TXID_CHECK)) {
            String message = "The transaction id in the edit log : "
                + sharedLogTxId
                + " does not match the transaction id inferred"
                + " from FSIMAGE : " + localLogTxId;
            LOG.fatal(message);
            throw new RuntimeException(message);
          }

          // skip previously loaded transactions
          if (!canApplyTransaction(sharedLogTxId, localLogTxId, op))
            continue;
          
          // for recovery, we do not want to re-load transactions,
          // but we want to populate local log with them
          if (shouldLoad(sharedLogTxId)) {
            FSEditLogLoader.loadEditRecord(
                logVersion, 
                inputEditStream,
                recentOpcodeOffsets, 
                opCounts, 
                fsNamesys, 
                fsDir, 
                numEdits, 
                op);
          }     
          
          LOG.info("Ingest: " + this.toString() 
                            + ", size: " + inputEditStream.length()
                            + ", processing transaction at offset: " + currentPosition 
                            + ", txid: " + op.txid
                            + ", opcode: " + op.opCode);
          
          if (op.opCode == FSEditLogOpCodes.OP_START_LOG_SEGMENT) {
            LOG.info("Ingest: Opening log segment: " + this.toString());
            localEditLog.open();
          } else if (op.opCode == FSEditLogOpCodes.OP_END_LOG_SEGMENT) {
            InjectionHandler
                .processEventIO(InjectionEvent.INGEST_CLEAR_STANDBY_STATE);
            LOG.info("Ingest: Closing log segment: " + this.toString());
            tearDown(localEditLog, true, true);
            numEdits++;
            LOG.info("Ingest: Reached log segment end. " + this.toString());
            break;
          } else {
            localEditLog.logEdit(op);
            if (inputEditStream.getReadChecksum() != FSEditLog
                .getChecksumForWrite().getValue()) {
              throw new IOException(
                  "Ingest: mismatched r/w checksums for transaction #"
                      + numEdits);
            }
          }
          
          numEdits++;
          standby.setLastCorrectTxId(op.txid);
        }
        catch (ChecksumException cex) {
          LOG.info("Checksum error reading the transaction #" + numEdits +
                   " reopening the file");
          reopen = true;
          break;
        }
        catch (IOException e) {
          LOG.info("Encountered error reading transaction", e);
          error = true; // if we haven't reached eof, then error.
          break;
        } finally {
          if (localEditLog.isOpen()) {
            localEditLog.logSyncIfNeeded();
          }
          fsNamesys.writeUnlock();
        }
      } // end inner while(running) -- all breaks come here
   
      // if we failed to read the entire transaction from disk, 
      // then roll back to the offset where there was a last good 
      // read, sleep for sometime for new transaction to
      // appear in the file and then continue;
      if (error || running) {
        // discard older buffers and start a fresh one.
      	inputEditStream.refresh(currentPosition, localEditLog.getLastWrittenTxId());
        setCatchingUp();

        if (error) {
          LOG.info("Ingest: Incomplete transaction record at offset " + 
              inputEditStream.getPosition() +
                   " but the file is of size " + inputEditStream.length() + 
                   ". Continuing....");
        }

        if (running && !lastScan) {
          try {
            Thread.sleep(100); // sleep for a second
          } catch (InterruptedException e) {
            // break out of waiting if we receive an interrupt.
          }
        }
      }
    } //end outer while(running)
    
    ///////////////////// FINAL ACTIONS /////////////////////
    
    // This was the last scan of the file but we could not read a full
    // transaction from disk. If we proceed this will corrupt the image
    if (error) {
      String errorMessage = FSEditLogLoader.getErrorMessage(recentOpcodeOffsets, currentPosition);
      LOG.error(errorMessage);
      throw new IOException("Failed to read the edits log. " + 
          "Incomplete transaction at " + currentPosition);
    }
    
    // If the last Scan was completed, then stop the Ingest thread.
    if (lastScan && quitAfterScan) {
      LOG.info("Ingest: lastScan completed. " + this.toString());
      running = false;
      if(localEditLog.isOpen()) {
        // quiesced non-finalized segment
        LOG.info("Ingest: Reached non-finalized log segment end. "+ this.toString());
        tearDown(localEditLog, false, localLogTxId != startTxId);
      }
    }
    FSEditLogLoader.dumpOpCounts(opCounts);
    return numEdits; // total transactions consumed
  }
728113
private void openInputStream(File txFile) throws IOException {
    int ioExceptionRetryCount = 0, logHeaderCorruptRetryCount = 0;
    
    LOG.info("Trying to load the edit log from " + txFile.getAbsolutePath());
    
    do {
      try {
        inputStream = new EditLogFileInputStream(txFile);
        editLogFilePosition = inputStream.getPosition();
        curStreamConsumed = false;
        curStreamFinished = false;
        readNullAfterStreamFinished = false;
        LOG.info("Successfully loaded the edits log from " +
            txFile.getAbsolutePath());
        break;
        
      } catch (LogHeaderCorruptException e1) {
        if (logHeaderCorruptRetryCount == LOG_HEADER_CORRUPT_RETRY_MAX) {
          LOG.error("Failed to load the edit log. No retries left.", e1);
          throw new IOException("Could not load the edit log");
        }
        logHeaderCorruptRetryCount ++;
        LOG.warn("Failed to load the edit log. Retry " +
            logHeaderCorruptRetryCount + " ...", e1);
        try {
          Thread.sleep(ioExceptionRetryCount * LOG_HEADER_CORRUPT_BASE_SLEEP);
        } catch (InterruptedException e) {
          throw new IOException(e);
        }
        
      } catch (IOException e2) {
        if (ioExceptionRetryCount == IO_EXCEPTION_RETRY_MAX) {
          LOG.error("Failed to load the edit log. No retries left.", e2);
          throw new IOException("Could not load the edit log");
        }
        ioExceptionRetryCount ++;
        LOG.warn("Failed to load the edit log. Retry " +
            ioExceptionRetryCount + " ...", e2);
        try {
          Thread.sleep(ioExceptionRetryCount * IO_EXCEPTION_BASE_SLEEP);
        } catch (InterruptedException e) {
          throw new IOException(e);
        }
      }
      
    } while (true);
  }
1125429
private void updateState(FSEditLogOp op, boolean checkTxnId) throws IOException {
    InjectionHandler.processEvent(InjectionEvent.SERVERLOGREADER_UPDATE, op);
    
    if (checkTxnId) {
      mostRecentlyReadTransactionTxId = ServerLogReaderUtil.checkTransactionId(
        mostRecentlyReadTransactionTxId, op);
    }
    
    updateStreamPosition();

    // read a valid operation
    core.getMetrics().readOperations.inc();
    mostRecentlyReadTransactionTime = now();

    // current log segment ends normally
    if (op.opCode == FSEditLogOpCodes.OP_END_LOG_SEGMENT) {
      LOG.info("Segment - ending log segment start txid: " + currentSegmentTxId
          + ", end txid: " + op.getTransactionId());
      // move forward with next segment
      currentSegmentTxId = op.getTransactionId() + 1;
      // set the stream to null so the next getNotification()
      // will recreate it
      currentEditLogInputStream = null;
      // indicate that a new stream will be opened
      currentEditLogInputStreamPosition = -1;
    } else if (op.opCode == FSEditLogOpCodes.OP_START_LOG_SEGMENT) {
      LOG.info("Segment - starting log segment start txid: "
          + currentSegmentTxId);
    }
  }
393464
public static EditLogInputStream getInputStream(JournalManager jm, long txid)
      throws IOException {
    List<EditLogInputStream> streams = new ArrayList<EditLogInputStream>();
    jm.selectInputStreams(streams, txid, true, false);
    if (streams.size() < 1) {
      throw new IOException("Cannot obtain stream for txid: " + txid);
    }
    Collections.sort(streams, JournalSet.EDIT_LOG_INPUT_STREAM_COMPARATOR);
    
    // we want the "oldest" available stream
    if (txid == HdfsConstants.INVALID_TXID) {
      return streams.get(0);
    }
    
    // we want a specific stream
    for (EditLogInputStream elis : streams) {
      if (elis.getFirstTxId() == txid) {
        return elis;
      }
    }
    // we cannot obtain the stream
    throw new IOException("Cannot obtain stream for txid: " + txid);
  }