@Test public void testIterator() { ByteBuffer buffer = ByteBuffer.allocate(1024); MemoryRecordsBuilder builder = new MemoryRecordsBuilder(buffer, magic, compression, TimestampType.CREATE_TIME, firstOffset, logAppendTime, pid, epoch, firstSequence, false, false, partitionLeaderEpoch, buffer.limit()); SimpleRecord[] records = new SimpleRecord[] { new SimpleRecord(1L, "a".getBytes(), "1".getBytes()), new SimpleRecord(2L, "b".getBytes(), "2".getBytes()), new SimpleRecord(3L, "c".getBytes(), "3".getBytes()), new SimpleRecord(4L, null, "4".getBytes()), new SimpleRecord(5L, "d".getBytes(), null), new SimpleRecord(6L, (byte[]) null, null) }; for (SimpleRecord record : records) builder.append(record); MemoryRecords memoryRecords = builder.build(); for (int iteration = 0; iteration < 2; iteration++) { int total = 0; for (RecordBatch batch : memoryRecords.batches()) { assertTrue(batch.isValid()); assertEquals(compression, batch.compressionType()); assertEquals(firstOffset + total, batch.baseOffset()); if (magic >= RecordBatch.MAGIC_VALUE_V2) { assertEquals(pid, batch.producerId()); assertEquals(epoch, batch.producerEpoch()); assertEquals(firstSequence + total, batch.baseSequence()); assertEquals(partitionLeaderEpoch, batch.partitionLeaderEpoch()); assertEquals(records.length, batch.countOrNull().intValue()); assertEquals(TimestampType.CREATE_TIME, batch.timestampType()); assertEquals(records[records.length - 1].timestamp(), batch.maxTimestamp()); } else { assertEquals(RecordBatch.NO_PRODUCER_ID, batch.producerId()); assertEquals(RecordBatch.NO_PRODUCER_EPOCH, batch.producerEpoch()); assertEquals(RecordBatch.NO_SEQUENCE, batch.baseSequence()); assertEquals(RecordBatch.NO_PARTITION_LEADER_EPOCH, batch.partitionLeaderEpoch()); assertNull(batch.countOrNull()); if (magic == RecordBatch.MAGIC_VALUE_V0) assertEquals(TimestampType.NO_TIMESTAMP_TYPE, batch.timestampType()); else assertEquals(TimestampType.CREATE_TIME, batch.timestampType()); } int recordCount = 0; for (Record record : batch) { assertTrue(record.isValid()); assertTrue(record.hasMagic(batch.magic())); assertFalse(record.isCompressed()); assertEquals(firstOffset + total, record.offset()); assertEquals(records[total].key(), record.key()); assertEquals(records[total].value(), record.value()); if (magic >= RecordBatch.MAGIC_VALUE_V2) assertEquals(firstSequence + total, record.sequence()); assertFalse(record.hasTimestampType(TimestampType.LOG_APPEND_TIME)); if (magic == RecordBatch.MAGIC_VALUE_V0) { assertEquals(RecordBatch.NO_TIMESTAMP, record.timestamp()); assertFalse(record.hasTimestampType(TimestampType.CREATE_TIME)); assertTrue(record.hasTimestampType(TimestampType.NO_TIMESTAMP_TYPE)); } else { assertEquals(records[total].timestamp(), record.timestamp()); assertFalse(record.hasTimestampType(TimestampType.NO_TIMESTAMP_TYPE)); if (magic < RecordBatch.MAGIC_VALUE_V2) assertTrue(record.hasTimestampType(TimestampType.CREATE_TIME)); else assertFalse(record.hasTimestampType(TimestampType.CREATE_TIME)); } total++; recordCount++; } assertEquals(batch.baseOffset() + recordCount - 1, batch.lastOffset()); } } }
@Override public Iterable<MutableRecordBatch> batches() { return batches; }
MemoryRecords extends AbstractRecords { @Override public Iterable<MutableRecordBatch> batches() { return batches; } }
MemoryRecords extends AbstractRecords { @Override public Iterable<MutableRecordBatch> batches() { return batches; } private  MemoryRecords(ByteBuffer buffer); }
MemoryRecords extends AbstractRecords { @Override public Iterable<MutableRecordBatch> batches() { return batches; } private  MemoryRecords(ByteBuffer buffer); @Override int sizeInBytes(); @Override long writeTo(GatheringByteChannel channel, long position, int length); int writeFullyTo(GatheringByteChannel channel); int validBytes(); @Override MemoryRecords downConvert(byte toMagic, long firstOffset); FilterResult filterTo(TopicPartition partition, RecordFilter filter, ByteBuffer destinationBuffer,
                                 int maxRecordBatchSize, BufferSupplier decompressionBufferSupplier); ByteBuffer buffer(); @Override Iterable<MutableRecordBatch> batches(); @Override String toString(); @Override boolean equals(Object o); @Override int hashCode(); static MemoryRecords readableRecords(ByteBuffer buffer); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset); static MemoryRecordsBuilder idempotentBuilder(ByteBuffer buffer,
                                                         CompressionType compressionType,
                                                         long baseOffset,
                                                         long producerId,
                                                         short producerEpoch,
                                                         int baseSequence); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               int partitionLeaderEpoch); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               CompressionType compressionType,
                                               long baseOffset,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence,
                                               boolean isTransactional); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence,
                                               boolean isTransactional,
                                               int partitionLeaderEpoch); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence,
                                               boolean isTransactional,
                                               boolean isControlBatch,
                                               int partitionLeaderEpoch); static MemoryRecords withRecords(CompressionType compressionType, SimpleRecord... records); static MemoryRecords withRecords(CompressionType compressionType, int partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withRecords(byte magic, CompressionType compressionType, SimpleRecord... records); static MemoryRecords withRecords(long initialOffset, CompressionType compressionType, SimpleRecord... records); static MemoryRecords withRecords(long initialOffset, CompressionType compressionType, Integer partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withIdempotentRecords(CompressionType compressionType, long producerId,
                                                      short producerEpoch, int baseSequence, SimpleRecord... records); static MemoryRecords withIdempotentRecords(byte magic, long initialOffset, CompressionType compressionType,
                                                      long producerId, short producerEpoch, int baseSequence,
                                                      int partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withIdempotentRecords(long initialOffset, CompressionType compressionType, long producerId,
                                                      short producerEpoch, int baseSequence, int partitionLeaderEpoch,
                                                      SimpleRecord... records); static MemoryRecords withTransactionalRecords(CompressionType compressionType, long producerId,
                                                         short producerEpoch, int baseSequence, SimpleRecord... records); static MemoryRecords withTransactionalRecords(byte magic, long initialOffset, CompressionType compressionType,
                                                         long producerId, short producerEpoch, int baseSequence,
                                                         int partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withTransactionalRecords(long initialOffset, CompressionType compressionType, long producerId,
                                                         short producerEpoch, int baseSequence, int partitionLeaderEpoch,
                                                         SimpleRecord... records); static MemoryRecords withRecords(byte magic, long initialOffset, CompressionType compressionType,
                                            TimestampType timestampType, SimpleRecord... records); static MemoryRecords withRecords(byte magic, long initialOffset, CompressionType compressionType,
                                            TimestampType timestampType, long producerId, short producerEpoch,
                                            int baseSequence, int partitionLeaderEpoch, boolean isTransactional,
                                            SimpleRecord ... records); static MemoryRecords withEndTransactionMarker(long producerId, short producerEpoch, EndTransactionMarker marker); static MemoryRecords withEndTransactionMarker(long timestamp, long producerId, short producerEpoch,
                                                         EndTransactionMarker marker); static MemoryRecords withEndTransactionMarker(long initialOffset, long timestamp, int partitionLeaderEpoch,
                                                         long producerId, short producerEpoch,
                                                         EndTransactionMarker marker); static void writeEndTransactionalMarker(ByteBuffer buffer, long initialOffset, long timestamp,
                                                   int partitionLeaderEpoch, long producerId, short producerEpoch,
                                                   EndTransactionMarker marker); }
MemoryRecords extends AbstractRecords { @Override public Iterable<MutableRecordBatch> batches() { return batches; } private  MemoryRecords(ByteBuffer buffer); @Override int sizeInBytes(); @Override long writeTo(GatheringByteChannel channel, long position, int length); int writeFullyTo(GatheringByteChannel channel); int validBytes(); @Override MemoryRecords downConvert(byte toMagic, long firstOffset); FilterResult filterTo(TopicPartition partition, RecordFilter filter, ByteBuffer destinationBuffer,
                                 int maxRecordBatchSize, BufferSupplier decompressionBufferSupplier); ByteBuffer buffer(); @Override Iterable<MutableRecordBatch> batches(); @Override String toString(); @Override boolean equals(Object o); @Override int hashCode(); static MemoryRecords readableRecords(ByteBuffer buffer); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset); static MemoryRecordsBuilder idempotentBuilder(ByteBuffer buffer,
                                                         CompressionType compressionType,
                                                         long baseOffset,
                                                         long producerId,
                                                         short producerEpoch,
                                                         int baseSequence); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               int partitionLeaderEpoch); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               CompressionType compressionType,
                                               long baseOffset,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence,
                                               boolean isTransactional); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence,
                                               boolean isTransactional,
                                               int partitionLeaderEpoch); static MemoryRecordsBuilder builder(ByteBuffer buffer,
                                               byte magic,
                                               CompressionType compressionType,
                                               TimestampType timestampType,
                                               long baseOffset,
                                               long logAppendTime,
                                               long producerId,
                                               short producerEpoch,
                                               int baseSequence,
                                               boolean isTransactional,
                                               boolean isControlBatch,
                                               int partitionLeaderEpoch); static MemoryRecords withRecords(CompressionType compressionType, SimpleRecord... records); static MemoryRecords withRecords(CompressionType compressionType, int partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withRecords(byte magic, CompressionType compressionType, SimpleRecord... records); static MemoryRecords withRecords(long initialOffset, CompressionType compressionType, SimpleRecord... records); static MemoryRecords withRecords(long initialOffset, CompressionType compressionType, Integer partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withIdempotentRecords(CompressionType compressionType, long producerId,
                                                      short producerEpoch, int baseSequence, SimpleRecord... records); static MemoryRecords withIdempotentRecords(byte magic, long initialOffset, CompressionType compressionType,
                                                      long producerId, short producerEpoch, int baseSequence,
                                                      int partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withIdempotentRecords(long initialOffset, CompressionType compressionType, long producerId,
                                                      short producerEpoch, int baseSequence, int partitionLeaderEpoch,
                                                      SimpleRecord... records); static MemoryRecords withTransactionalRecords(CompressionType compressionType, long producerId,
                                                         short producerEpoch, int baseSequence, SimpleRecord... records); static MemoryRecords withTransactionalRecords(byte magic, long initialOffset, CompressionType compressionType,
                                                         long producerId, short producerEpoch, int baseSequence,
                                                         int partitionLeaderEpoch, SimpleRecord... records); static MemoryRecords withTransactionalRecords(long initialOffset, CompressionType compressionType, long producerId,
                                                         short producerEpoch, int baseSequence, int partitionLeaderEpoch,
                                                         SimpleRecord... records); static MemoryRecords withRecords(byte magic, long initialOffset, CompressionType compressionType,
                                            TimestampType timestampType, SimpleRecord... records); static MemoryRecords withRecords(byte magic, long initialOffset, CompressionType compressionType,
                                            TimestampType timestampType, long producerId, short producerEpoch,
                                            int baseSequence, int partitionLeaderEpoch, boolean isTransactional,
                                            SimpleRecord ... records); static MemoryRecords withEndTransactionMarker(long producerId, short producerEpoch, EndTransactionMarker marker); static MemoryRecords withEndTransactionMarker(long timestamp, long producerId, short producerEpoch,
                                                         EndTransactionMarker marker); static MemoryRecords withEndTransactionMarker(long initialOffset, long timestamp, int partitionLeaderEpoch,
                                                         long producerId, short producerEpoch,
                                                         EndTransactionMarker marker); static void writeEndTransactionalMarker(ByteBuffer buffer, long initialOffset, long timestamp,
                                                   int partitionLeaderEpoch, long producerId, short producerEpoch,
                                                   EndTransactionMarker marker); static final MemoryRecords EMPTY; }