@Test public void testReadLogMessageAndPosition() throws Exception { String path = FilenameUtils.concat(getTempPath(), "thrift.log"); SimpleThriftLogger<LogMessage> logger = new SimpleThriftLogger<>(path); long startOffset = 0L; List<LogMessageAndPosition> messagesWritten = null; try { writeThriftLogMessages(logger, 200, 500, 50); startOffset = logger.getByteOffset(); messagesWritten = writeThriftLogMessages(logger, 3, 500, 5000); } finally { logger.close(); } LogFileReader reader = new ThriftLogFileReader(logger.getLogFile(), path, 0L, 16000, 500); try { reader.setByteOffset(startOffset); reader.readLogMessageAndPosition(); fail("Should throw when thrift message is bigger than max message size"); } catch (LogFileReaderException exception) { } finally { reader.close(); } reader = new ThriftLogFileReader(logger.getLogFile(), path, 0L, 16000, 16000); List<LogMessageAndPosition> messagesRead = Lists.newArrayListWithExpectedSize(3); try { reader.setByteOffset(startOffset); LogMessageAndPosition message = reader.readLogMessageAndPosition(); while (message != null) { messagesRead.add(message); message = reader.readLogMessageAndPosition(); } } finally { reader.close(); } assertThat(messagesRead, is(messagesWritten)); }
@Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader closed."); } try { LogMessage logMessage = thriftReader.read(); while (logMessage != null) { long newByteOffset = thriftReader.getByteOffset(); int messageSize = logMessage.getMessage().length; if (messageSize > maxMessageSize) { LOG.warn("Found a message at offset " + newByteOffset + "that exceeds the size limit in " + logFile.toString() + ": messageSize = " + messageSize); OpenTsdbMetricConverter.incr("singer.thrift_reader.skip_message", 1, "path=" + path); logMessage = thriftReader.read(); } else { LogPosition position = new LogPosition(logFile, newByteOffset); return new LogMessageAndPosition(logMessage, position); } } } catch (TException e) { LOG.error("Caught TException while reading " + logFile, e); OpenTsdbMetricConverter.incr("singer.reader.exception.texception", 1, "path=" + path); throw new LogFileReaderException("Cannot read a log message.", e); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); Stats.incr("singer.reader.exception.unexpected"); throw new LogFileReaderException("Cannot read a log message.", e); } return null; }
ThriftLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader closed."); } try { LogMessage logMessage = thriftReader.read(); while (logMessage != null) { long newByteOffset = thriftReader.getByteOffset(); int messageSize = logMessage.getMessage().length; if (messageSize > maxMessageSize) { LOG.warn("Found a message at offset " + newByteOffset + "that exceeds the size limit in " + logFile.toString() + ": messageSize = " + messageSize); OpenTsdbMetricConverter.incr("singer.thrift_reader.skip_message", 1, "path=" + path); logMessage = thriftReader.read(); } else { LogPosition position = new LogPosition(logFile, newByteOffset); return new LogMessageAndPosition(logMessage, position); } } } catch (TException e) { LOG.error("Caught TException while reading " + logFile, e); OpenTsdbMetricConverter.incr("singer.reader.exception.texception", 1, "path=" + path); throw new LogFileReaderException("Cannot read a log message.", e); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); Stats.incr("singer.reader.exception.unexpected"); throw new LogFileReaderException("Cannot read a log message.", e); } return null; } }
ThriftLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader closed."); } try { LogMessage logMessage = thriftReader.read(); while (logMessage != null) { long newByteOffset = thriftReader.getByteOffset(); int messageSize = logMessage.getMessage().length; if (messageSize > maxMessageSize) { LOG.warn("Found a message at offset " + newByteOffset + "that exceeds the size limit in " + logFile.toString() + ": messageSize = " + messageSize); OpenTsdbMetricConverter.incr("singer.thrift_reader.skip_message", 1, "path=" + path); logMessage = thriftReader.read(); } else { LogPosition position = new LogPosition(logFile, newByteOffset); return new LogMessageAndPosition(logMessage, position); } } } catch (TException e) { LOG.error("Caught TException while reading " + logFile, e); OpenTsdbMetricConverter.incr("singer.reader.exception.texception", 1, "path=" + path); throw new LogFileReaderException("Cannot read a log message.", e); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); Stats.incr("singer.reader.exception.unexpected"); throw new LogFileReaderException("Cannot read a log message.", e); } return null; } @SuppressWarnings({ "rawtypes", "unchecked" })  ThriftLogFileReader(
      LogFile logFile,
      String path,
      long byteOffset,
      int readBufferSize,
      int maxMessageSize); }
ThriftLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader closed."); } try { LogMessage logMessage = thriftReader.read(); while (logMessage != null) { long newByteOffset = thriftReader.getByteOffset(); int messageSize = logMessage.getMessage().length; if (messageSize > maxMessageSize) { LOG.warn("Found a message at offset " + newByteOffset + "that exceeds the size limit in " + logFile.toString() + ": messageSize = " + messageSize); OpenTsdbMetricConverter.incr("singer.thrift_reader.skip_message", 1, "path=" + path); logMessage = thriftReader.read(); } else { LogPosition position = new LogPosition(logFile, newByteOffset); return new LogMessageAndPosition(logMessage, position); } } } catch (TException e) { LOG.error("Caught TException while reading " + logFile, e); OpenTsdbMetricConverter.incr("singer.reader.exception.texception", 1, "path=" + path); throw new LogFileReaderException("Cannot read a log message.", e); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); Stats.incr("singer.reader.exception.unexpected"); throw new LogFileReaderException("Cannot read a log message.", e); } return null; } @SuppressWarnings({ "rawtypes", "unchecked" })  ThriftLogFileReader(
      LogFile logFile,
      String path,
      long byteOffset,
      int readBufferSize,
      int maxMessageSize); @Override LogMessageAndPosition readLogMessageAndPosition(); void close(); @Override LogFile getLogFile(); @Override long getByteOffset(); @Override void setByteOffset(long byteOffset); @Override boolean isClosed(); }
ThriftLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader closed."); } try { LogMessage logMessage = thriftReader.read(); while (logMessage != null) { long newByteOffset = thriftReader.getByteOffset(); int messageSize = logMessage.getMessage().length; if (messageSize > maxMessageSize) { LOG.warn("Found a message at offset " + newByteOffset + "that exceeds the size limit in " + logFile.toString() + ": messageSize = " + messageSize); OpenTsdbMetricConverter.incr("singer.thrift_reader.skip_message", 1, "path=" + path); logMessage = thriftReader.read(); } else { LogPosition position = new LogPosition(logFile, newByteOffset); return new LogMessageAndPosition(logMessage, position); } } } catch (TException e) { LOG.error("Caught TException while reading " + logFile, e); OpenTsdbMetricConverter.incr("singer.reader.exception.texception", 1, "path=" + path); throw new LogFileReaderException("Cannot read a log message.", e); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); Stats.incr("singer.reader.exception.unexpected"); throw new LogFileReaderException("Cannot read a log message.", e); } return null; } @SuppressWarnings({ "rawtypes", "unchecked" })  ThriftLogFileReader(
      LogFile logFile,
      String path,
      long byteOffset,
      int readBufferSize,
      int maxMessageSize); @Override LogMessageAndPosition readLogMessageAndPosition(); void close(); @Override LogFile getLogFile(); @Override long getByteOffset(); @Override void setByteOffset(long byteOffset); @Override boolean isClosed();  }