@Test public void testWriteTo() throws IOException { try (FileRecords fileRecords = FileRecords.open(tempFile())) { fileRecords.append(MemoryRecords.withRecords(magic, compression, new SimpleRecord("foo".getBytes()))); fileRecords.flush(); FileLogInputStream logInputStream = new FileLogInputStream(fileRecords.channel(), 0, fileRecords.sizeInBytes()); FileChannelRecordBatch batch = logInputStream.nextBatch(); assertNotNull(batch); assertEquals(magic, batch.magic()); ByteBuffer buffer = ByteBuffer.allocate(128); batch.writeTo(buffer); buffer.flip(); MemoryRecords memRecords = MemoryRecords.readableRecords(buffer); List<Record> records = Utils.toList(memRecords.records().iterator()); assertEquals(1, records.size()); Record record0 = records.get(0); assertTrue(record0.hasMagic(magic)); assertEquals("foo", Utils.utf8(record0.value(), record0.valueSize())); } }
@Override public FileChannelRecordBatch nextBatch() throws IOException { if (position + HEADER_SIZE_UP_TO_MAGIC >= end) return null; logHeaderBuffer.rewind(); Utils.readFullyOrFail(channel, logHeaderBuffer, position, "log header"); logHeaderBuffer.rewind(); long offset = logHeaderBuffer.getLong(OFFSET_OFFSET); int size = logHeaderBuffer.getInt(SIZE_OFFSET); if (size < LegacyRecord.RECORD_OVERHEAD_V0) throw new CorruptRecordException(String.format("Record size is smaller than minimum record overhead (%d).", LegacyRecord.RECORD_OVERHEAD_V0)); if (position + LOG_OVERHEAD + size > end) return null; byte magic = logHeaderBuffer.get(MAGIC_OFFSET); final FileChannelRecordBatch batch; if (magic < RecordBatch.MAGIC_VALUE_V2) batch = new LegacyFileChannelRecordBatch(offset, magic, channel, position, size); else batch = new DefaultFileChannelRecordBatch(offset, magic, channel, position, size); position += batch.sizeInBytes(); return batch; }
FileLogInputStream implements LogInputStream<FileLogInputStream.FileChannelRecordBatch> { @Override public FileChannelRecordBatch nextBatch() throws IOException { if (position + HEADER_SIZE_UP_TO_MAGIC >= end) return null; logHeaderBuffer.rewind(); Utils.readFullyOrFail(channel, logHeaderBuffer, position, "log header"); logHeaderBuffer.rewind(); long offset = logHeaderBuffer.getLong(OFFSET_OFFSET); int size = logHeaderBuffer.getInt(SIZE_OFFSET); if (size < LegacyRecord.RECORD_OVERHEAD_V0) throw new CorruptRecordException(String.format("Record size is smaller than minimum record overhead (%d).", LegacyRecord.RECORD_OVERHEAD_V0)); if (position + LOG_OVERHEAD + size > end) return null; byte magic = logHeaderBuffer.get(MAGIC_OFFSET); final FileChannelRecordBatch batch; if (magic < RecordBatch.MAGIC_VALUE_V2) batch = new LegacyFileChannelRecordBatch(offset, magic, channel, position, size); else batch = new DefaultFileChannelRecordBatch(offset, magic, channel, position, size); position += batch.sizeInBytes(); return batch; } }
FileLogInputStream implements LogInputStream<FileLogInputStream.FileChannelRecordBatch> { @Override public FileChannelRecordBatch nextBatch() throws IOException { if (position + HEADER_SIZE_UP_TO_MAGIC >= end) return null; logHeaderBuffer.rewind(); Utils.readFullyOrFail(channel, logHeaderBuffer, position, "log header"); logHeaderBuffer.rewind(); long offset = logHeaderBuffer.getLong(OFFSET_OFFSET); int size = logHeaderBuffer.getInt(SIZE_OFFSET); if (size < LegacyRecord.RECORD_OVERHEAD_V0) throw new CorruptRecordException(String.format("Record size is smaller than minimum record overhead (%d).", LegacyRecord.RECORD_OVERHEAD_V0)); if (position + LOG_OVERHEAD + size > end) return null; byte magic = logHeaderBuffer.get(MAGIC_OFFSET); final FileChannelRecordBatch batch; if (magic < RecordBatch.MAGIC_VALUE_V2) batch = new LegacyFileChannelRecordBatch(offset, magic, channel, position, size); else batch = new DefaultFileChannelRecordBatch(offset, magic, channel, position, size); position += batch.sizeInBytes(); return batch; } FileLogInputStream(FileChannel channel,
                       int start,
                       int end); }
FileLogInputStream implements LogInputStream<FileLogInputStream.FileChannelRecordBatch> { @Override public FileChannelRecordBatch nextBatch() throws IOException { if (position + HEADER_SIZE_UP_TO_MAGIC >= end) return null; logHeaderBuffer.rewind(); Utils.readFullyOrFail(channel, logHeaderBuffer, position, "log header"); logHeaderBuffer.rewind(); long offset = logHeaderBuffer.getLong(OFFSET_OFFSET); int size = logHeaderBuffer.getInt(SIZE_OFFSET); if (size < LegacyRecord.RECORD_OVERHEAD_V0) throw new CorruptRecordException(String.format("Record size is smaller than minimum record overhead (%d).", LegacyRecord.RECORD_OVERHEAD_V0)); if (position + LOG_OVERHEAD + size > end) return null; byte magic = logHeaderBuffer.get(MAGIC_OFFSET); final FileChannelRecordBatch batch; if (magic < RecordBatch.MAGIC_VALUE_V2) batch = new LegacyFileChannelRecordBatch(offset, magic, channel, position, size); else batch = new DefaultFileChannelRecordBatch(offset, magic, channel, position, size); position += batch.sizeInBytes(); return batch; } FileLogInputStream(FileChannel channel,
                       int start,
                       int end); @Override FileChannelRecordBatch nextBatch(); }
FileLogInputStream implements LogInputStream<FileLogInputStream.FileChannelRecordBatch> { @Override public FileChannelRecordBatch nextBatch() throws IOException { if (position + HEADER_SIZE_UP_TO_MAGIC >= end) return null; logHeaderBuffer.rewind(); Utils.readFullyOrFail(channel, logHeaderBuffer, position, "log header"); logHeaderBuffer.rewind(); long offset = logHeaderBuffer.getLong(OFFSET_OFFSET); int size = logHeaderBuffer.getInt(SIZE_OFFSET); if (size < LegacyRecord.RECORD_OVERHEAD_V0) throw new CorruptRecordException(String.format("Record size is smaller than minimum record overhead (%d).", LegacyRecord.RECORD_OVERHEAD_V0)); if (position + LOG_OVERHEAD + size > end) return null; byte magic = logHeaderBuffer.get(MAGIC_OFFSET); final FileChannelRecordBatch batch; if (magic < RecordBatch.MAGIC_VALUE_V2) batch = new LegacyFileChannelRecordBatch(offset, magic, channel, position, size); else batch = new DefaultFileChannelRecordBatch(offset, magic, channel, position, size); position += batch.sizeInBytes(); return batch; } FileLogInputStream(FileChannel channel,
                       int start,
                       int end); @Override FileChannelRecordBatch nextBatch();  }