@Test(expected = IllegalArgumentException.class) public void testSetNoTimestampTypeNotAllowed() { MemoryRecords records = MemoryRecords.withRecords(RecordBatch.MAGIC_VALUE_V2, 0L, CompressionType.NONE, TimestampType.CREATE_TIME, new SimpleRecord(1L, "a".getBytes(), "1".getBytes()), new SimpleRecord(2L, "b".getBytes(), "2".getBytes()), new SimpleRecord(3L, "c".getBytes(), "3".getBytes())); DefaultRecordBatch batch = new DefaultRecordBatch(records.buffer()); batch.setMaxTimestamp(TimestampType.NO_TIMESTAMP_TYPE, RecordBatch.NO_TIMESTAMP); }
@Override public void setMaxTimestamp(TimestampType timestampType, long maxTimestamp) { long currentMaxTimestamp = maxTimestamp(); if (timestampType() == timestampType && currentMaxTimestamp == maxTimestamp) return; byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch()); buffer.putShort(ATTRIBUTES_OFFSET, attributes); buffer.putLong(MAX_TIMESTAMP_OFFSET, maxTimestamp); long crc = computeChecksum(); ByteUtils.writeUnsignedInt(buffer, CRC_OFFSET, crc); }
DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch { @Override public void setMaxTimestamp(TimestampType timestampType, long maxTimestamp) { long currentMaxTimestamp = maxTimestamp(); if (timestampType() == timestampType && currentMaxTimestamp == maxTimestamp) return; byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch()); buffer.putShort(ATTRIBUTES_OFFSET, attributes); buffer.putLong(MAX_TIMESTAMP_OFFSET, maxTimestamp); long crc = computeChecksum(); ByteUtils.writeUnsignedInt(buffer, CRC_OFFSET, crc); } }
DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch { @Override public void setMaxTimestamp(TimestampType timestampType, long maxTimestamp) { long currentMaxTimestamp = maxTimestamp(); if (timestampType() == timestampType && currentMaxTimestamp == maxTimestamp) return; byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch()); buffer.putShort(ATTRIBUTES_OFFSET, attributes); buffer.putLong(MAX_TIMESTAMP_OFFSET, maxTimestamp); long crc = computeChecksum(); ByteUtils.writeUnsignedInt(buffer, CRC_OFFSET, crc); } DefaultRecordBatch(ByteBuffer buffer); }
DefaultRecordBatch extends AbstractRecordBatch implements MutableRecordBatch { @Override public void setMaxTimestamp(TimestampType timestampType, long maxTimestamp) { long currentMaxTimestamp = maxTimestamp(); if (timestampType() == timestampType && currentMaxTimestamp == maxTimestamp) return; byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch()); buffer.putShort(ATTRIBUTES_OFFSET, attributes); buffer.putLong(MAX_TIMESTAMP_OFFSET, maxTimestamp); long crc = computeChecksum(); ByteUtils.writeUnsignedInt(buffer, CRC_OFFSET, crc); } 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 { @Override public void setMaxTimestamp(TimestampType timestampType, long maxTimestamp) { long currentMaxTimestamp = maxTimestamp(); if (timestampType() == timestampType && currentMaxTimestamp == maxTimestamp) return; byte attributes = computeAttributes(compressionType(), timestampType, isTransactional(), isControlBatch()); buffer.putShort(ATTRIBUTES_OFFSET, attributes); buffer.putLong(MAX_TIMESTAMP_OFFSET, maxTimestamp); long crc = computeChecksum(); ByteUtils.writeUnsignedInt(buffer, CRC_OFFSET, crc); } 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; }