@Test public void testWriteEmptyHeader() { long producerId = 23423L; short producerEpoch = 145; int baseSequence = 983; long baseOffset = 15L; long lastOffset = 37; int partitionLeaderEpoch = 15; long timestamp = System.currentTimeMillis(); for (TimestampType timestampType : Arrays.asList(TimestampType.CREATE_TIME, TimestampType.LOG_APPEND_TIME)) { for (boolean isTransactional : Arrays.asList(true, false)) { for (boolean isControlBatch : Arrays.asList(true, false)) { ByteBuffer buffer = ByteBuffer.allocate(2048); DefaultRecordBatch.writeEmptyHeader(buffer, RecordBatch.CURRENT_MAGIC_VALUE, producerId, producerEpoch, baseSequence, baseOffset, lastOffset, partitionLeaderEpoch, timestampType, timestamp, isTransactional, isControlBatch); buffer.flip(); DefaultRecordBatch batch = new DefaultRecordBatch(buffer); assertEquals(producerId, batch.producerId()); assertEquals(producerEpoch, batch.producerEpoch()); assertEquals(baseSequence, batch.baseSequence()); assertEquals(baseSequence + ((int) (lastOffset - baseOffset)), batch.lastSequence()); assertEquals(baseOffset, batch.baseOffset()); assertEquals(lastOffset, batch.lastOffset()); assertEquals(partitionLeaderEpoch, batch.partitionLeaderEpoch()); assertEquals(isTransactional, batch.isTransactional()); assertEquals(timestampType, batch.timestampType()); assertEquals(timestamp, batch.maxTimestamp()); assertEquals(isControlBatch, batch.isControlBatch()); } } } }
public static void writeEmptyHeader(ByteBuffer buffer, byte magic, long producerId, short producerEpoch, int baseSequence, long baseOffset, long lastOffset, int partitionLeaderEpoch, TimestampType timestampType, long timestamp, boolean isTransactional, boolean isControlRecord) { int offsetDelta = (int) (lastOffset - baseOffset); writeHeader(buffer, baseOffset, offsetDelta, DefaultRecordBatch.RECORD_BATCH_OVERHEAD, magic, CompressionType.NONE, timestampType, timestamp, timestamp, producerId, producerEpoch, baseSequence, isTransactional, isControlRecord, partitionLeaderEpoch, 0); }
DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch { public static void writeEmptyHeader(ByteBuffer buffer, byte magic, long producerId, short producerEpoch, int baseSequence, long baseOffset, long lastOffset, int partitionLeaderEpoch, TimestampType timestampType, long timestamp, boolean isTransactional, boolean isControlRecord) { int offsetDelta = (int) (lastOffset - baseOffset); writeHeader(buffer, baseOffset, offsetDelta, DefaultRecordBatch.RECORD_BATCH_OVERHEAD, magic, CompressionType.NONE, timestampType, timestamp, timestamp, producerId, producerEpoch, baseSequence, isTransactional, isControlRecord, partitionLeaderEpoch, 0); } }
DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch { public static void writeEmptyHeader(ByteBuffer buffer, byte magic, long producerId, short producerEpoch, int baseSequence, long baseOffset, long lastOffset, int partitionLeaderEpoch, TimestampType timestampType, long timestamp, boolean isTransactional, boolean isControlRecord) { int offsetDelta = (int) (lastOffset - baseOffset); writeHeader(buffer, baseOffset, offsetDelta, DefaultRecordBatch.RECORD_BATCH_OVERHEAD, magic, CompressionType.NONE, timestampType, timestamp, timestamp, producerId, producerEpoch, baseSequence, isTransactional, isControlRecord, partitionLeaderEpoch, 0); } DefaultRecordBatch(ByteBuffer buffer); }
DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch { public static void writeEmptyHeader(ByteBuffer buffer, byte magic, long producerId, short producerEpoch, int baseSequence, long baseOffset, long lastOffset, int partitionLeaderEpoch, TimestampType timestampType, long timestamp, boolean isTransactional, boolean isControlRecord) { int offsetDelta = (int) (lastOffset - baseOffset); writeHeader(buffer, baseOffset, offsetDelta, DefaultRecordBatch.RECORD_BATCH_OVERHEAD, magic, CompressionType.NONE, timestampType, timestamp, timestamp, producerId, producerEpoch, baseSequence, isTransactional, isControlRecord, partitionLeaderEpoch, 0); } DefaultRecordBatch(ByteBuffer buffer); @Override byte magic(); @Override void ensureValid(); long baseTimestamp(); @Override long maxTimestamp(); @Override TimestampType timestampType(); @Override long baseOffset(); @Override long lastOffset(); @Override long producerId(); @Override short producerEpoch(); @Override int baseSequence(); @Override int lastSequence(); @Override CompressionType compressionType(); @Override int sizeInBytes(); @Override Integer countOrNull(); @Override void writeTo(ByteBuffer buffer); @Override void writeTo(ByteBufferOutputStream outputStream); @Override boolean isTransactional(); @Override boolean isControlBatch(); @Override int partitionLeaderEpoch(); @Override Iterator<Record> iterator(); @Override CloseableIterator<Record> streamingIterator(BufferSupplier bufferSupplier); @Override void setLastOffset(long offset); @Override void setMaxTimestamp(TimestampType timestampType, long maxTimestamp); @Override void setPartitionLeaderEpoch(int epoch); @Override long checksum(); boolean isValid(); @Override boolean equals(Object o); @Override int hashCode(); static void writeEmptyHeader(ByteBuffer buffer,
                                        byte magic,
                                        long producerId,
                                        short producerEpoch,
                                        int baseSequence,
                                        long baseOffset,
                                        long lastOffset,
                                        int partitionLeaderEpoch,
                                        TimestampType timestampType,
                                        long timestamp,
                                        boolean isTransactional,
                                        boolean isControlRecord); @Override String toString(); static int sizeInBytes(long baseOffset, Iterable<Record> records); static int sizeInBytes(Iterable<SimpleRecord> records); }
DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch { public static void writeEmptyHeader(ByteBuffer buffer, byte magic, long producerId, short producerEpoch, int baseSequence, long baseOffset, long lastOffset, int partitionLeaderEpoch, TimestampType timestampType, long timestamp, boolean isTransactional, boolean isControlRecord) { int offsetDelta = (int) (lastOffset - baseOffset); writeHeader(buffer, baseOffset, offsetDelta, DefaultRecordBatch.RECORD_BATCH_OVERHEAD, magic, CompressionType.NONE, timestampType, timestamp, timestamp, producerId, producerEpoch, baseSequence, isTransactional, isControlRecord, partitionLeaderEpoch, 0); } DefaultRecordBatch(ByteBuffer buffer); @Override byte magic(); @Override void ensureValid(); long baseTimestamp(); @Override long maxTimestamp(); @Override TimestampType timestampType(); @Override long baseOffset(); @Override long lastOffset(); @Override long producerId(); @Override short producerEpoch(); @Override int baseSequence(); @Override int lastSequence(); @Override CompressionType compressionType(); @Override int sizeInBytes(); @Override Integer countOrNull(); @Override void writeTo(ByteBuffer buffer); @Override void writeTo(ByteBufferOutputStream outputStream); @Override boolean isTransactional(); @Override boolean isControlBatch(); @Override int partitionLeaderEpoch(); @Override Iterator<Record> iterator(); @Override CloseableIterator<Record> streamingIterator(BufferSupplier bufferSupplier); @Override void setLastOffset(long offset); @Override void setMaxTimestamp(TimestampType timestampType, long maxTimestamp); @Override void setPartitionLeaderEpoch(int epoch); @Override long checksum(); boolean isValid(); @Override boolean equals(Object o); @Override int hashCode(); static void writeEmptyHeader(ByteBuffer buffer,
                                        byte magic,
                                        long producerId,
                                        short producerEpoch,
                                        int baseSequence,
                                        long baseOffset,
                                        long lastOffset,
                                        int partitionLeaderEpoch,
                                        TimestampType timestampType,
                                        long timestamp,
                                        boolean isTransactional,
                                        boolean isControlRecord); @Override String toString(); static int sizeInBytes(long baseOffset, Iterable<Record> records); static int sizeInBytes(Iterable<SimpleRecord> records); static final int RECORD_BATCH_OVERHEAD; }