public synchronized LogRecord readRecord(int waitMillis) throws IOException, InterruptedException, LogTimeoutException, ReplicatorException {
  assertReadMode();
  long offset=dataInput.getOffset();
  if (logger.isDebugEnabled())   logger.debug("Reading log file position=" + offset);
  dataInput.mark(65636);
  long startIntervalMillis=System.currentTimeMillis();
  long available=dataInput.waitAvailable(RECORD_LENGTH_SIZE,waitMillis);
  if (available < RECORD_LENGTH_SIZE) {
    dataInput.reset();
    if (waitMillis > 0) {
      throw new LogTimeoutException("Log read timeout: waitMillis=" + waitMillis + " file="+ file.getName()+ " offset="+ offset);
    }
 else     if (available == 0) {
      if (logger.isDebugEnabled())       logger.debug("Read empty record");
      return new LogRecord(file,offset,false);
    }
 else     if (available < RECORD_LENGTH_SIZE) {
      if (logger.isDebugEnabled())       logger.debug("Length is truncated; returning immediately");
      return new LogRecord(file,offset,true);
    }
  }
  int recordLength=dataInput.readInt();
  if (recordLength < LogRecord.NON_DATA_BYTES || recordLength > MAX_RECORD_LENGTH) {
    logger.warn("Record length is invalid, log may be corrupt: offset=" + offset + " record length="+ recordLength);
    dataInput.reset();
    return new LogRecord(file,offset,true);
  }
  if (logger.isDebugEnabled())   logger.debug("Record length=" + recordLength);
  waitMillis=waitMillis + (int)(startIntervalMillis - System.currentTimeMillis());
  int remainingRecordLength=recordLength - RECORD_LENGTH_SIZE;
  available=dataInput.waitAvailable(remainingRecordLength,waitMillis);
  if (available < remainingRecordLength) {
    dataInput.reset();
    if (waitMillis > 0) {
      throw new LogTimeoutException("Log read timeout: waitMillis=" + waitMillis + " file="+ file.getName()+ " offset="+ offset);
    }
 else {
      return new LogRecord(file,offset,true);
    }
  }
  byte[] bytesToRead=new byte[recordLength - LogRecord.NON_DATA_BYTES];
  dataInput.readFully(bytesToRead);
  byte crcType=dataInput.readByte();
  long crc=dataInput.readLong();
  return new LogRecord(file,offset,bytesToRead,crcType,crc);
}

public synchronized LogRecord readRecord ( int waitMillis ) throws IOException , InterruptedException , LogTimeoutException , ReplicatorException { assertReadMode ( ) ; long offset = dataInput . getOffset ( ) ; if ( logger . isDebugEnabled ( ) ) logger . debug ( "Reading log file position=" + offset ) ; dataInput . mark ( 65636 ) ; long startIntervalMillis = System . currentTimeMillis ( ) ; long available = dataInput . waitAvailable ( RECORD_LENGTH_SIZE , waitMillis ) ; if ( available < RECORD_LENGTH_SIZE ) { dataInput . reset ( ) ; if ( waitMillis > 0 ) { throw new LogTimeoutException ( "Log read timeout: waitMillis=" + waitMillis + " file=" + file . getName ( ) + " offset=" + offset ) ; } else if ( available == 0 ) { if ( logger . isDebugEnabled ( ) ) logger . debug ( "Read empty record" ) ; return new LogRecord ( file , offset , false ) ; } else if ( available < RECORD_LENGTH_SIZE ) { if ( logger . isDebugEnabled ( ) ) logger . debug ( "Length is truncated; returning immediately" ) ; return new LogRecord ( file , offset , true ) ; } } int recordLength = dataInput . readInt ( ) ; if ( recordLength < LogRecord . NON_DATA_BYTES || recordLength > MAX_RECORD_LENGTH ) { logger . warn ( "Record length is invalid, log may be corrupt: offset=" + offset + " record length=" + recordLength ) ; dataInput . reset ( ) ; return new LogRecord ( file , offset , true ) ; } if ( logger . isDebugEnabled ( ) ) logger . debug ( "Record length=" + recordLength ) ; waitMillis = waitMillis + ( int ) ( startIntervalMillis - System . currentTimeMillis ( ) ) ; int remainingRecordLength = recordLength - RECORD_LENGTH_SIZE ; available = dataInput . waitAvailable ( remainingRecordLength , waitMillis ) ; if ( available < remainingRecordLength ) { dataInput . reset ( ) ; if ( waitMillis > 0 ) { throw new LogTimeoutException ( "Log read timeout: waitMillis=" + waitMillis + " file=" + file . getName ( ) + " offset=" + offset ) ; } else { return new LogRecord ( file , offset , true ) ; } } byte [ ] bytesToRead = new byte [ recordLength - LogRecord . NON_DATA_BYTES ] ; dataInput . readFully ( bytesToRead ) ; byte crcType = dataInput . readByte ( ) ; long crc = dataInput . readLong ( ) ; return new LogRecord ( file , offset , bytesToRead , crcType , crc ) ; }
reads a record from the file into a byte array . we may encounter a number of unpredictable conditions at this point that we need to report accurately to layers above us that will decide whether it represents a problem .
