@Test public void testBatchIterationWithMultipleRecordsPerBatch() throws IOException { if (magic < MAGIC_VALUE_V2 && compression == CompressionType.NONE) return; try (FileRecords fileRecords = FileRecords.open(tempFile())) { SimpleRecord[] firstBatchRecords = new SimpleRecord[]{ new SimpleRecord(3241324L, "a".getBytes(), "1".getBytes()), new SimpleRecord(234280L, "b".getBytes(), "2".getBytes()) }; SimpleRecord[] secondBatchRecords = new SimpleRecord[]{ new SimpleRecord(238423489L, "c".getBytes(), "3".getBytes()), new SimpleRecord(897839L, null, "4".getBytes()), new SimpleRecord(8234020L, "e".getBytes(), null) }; fileRecords.append(MemoryRecords.withRecords(magic, 0L, compression, CREATE_TIME, firstBatchRecords)); fileRecords.append(MemoryRecords.withRecords(magic, 1L, compression, CREATE_TIME, secondBatchRecords)); fileRecords.flush(); FileLogInputStream logInputStream = new FileLogInputStream(fileRecords.channel(), 0, fileRecords.sizeInBytes()); FileChannelRecordBatch firstBatch = logInputStream.nextBatch(); assertNoProducerData(firstBatch); assertGenericRecordBatchData(firstBatch, 0L, 3241324L, firstBatchRecords); FileChannelRecordBatch secondBatch = logInputStream.nextBatch(); assertNoProducerData(secondBatch); assertGenericRecordBatchData(secondBatch, 1L, 238423489L, secondBatchRecords); assertNull(logInputStream.nextBatch()); } }
@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();  }