@Test(expected = IllegalArgumentException.class) public void testWriteIdempotentWithInvalidEpoch() { ByteBuffer buffer = ByteBuffer.allocate(128); buffer.position(bufferOffset); long pid = 9809; short epoch = RecordBatch.NO_PRODUCER_EPOCH; int sequence = 2342; MemoryRecordsBuilder builder = new MemoryRecordsBuilder(buffer, RecordBatch.CURRENT_MAGIC_VALUE, compressionType, TimestampType.CREATE_TIME, 0L, 0L, pid, epoch, sequence, true, false, RecordBatch.NO_PARTITION_LEADER_EPOCH, buffer.capacity()); builder.close(); }
public void close() { if (aborted) throw new IllegalStateException("Cannot close MemoryRecordsBuilder as it has already been aborted"); if (builtRecords != null) return; validateProducerState(); closeForRecordAppends(); if (numRecords == 0L) { buffer().position(initialPosition); builtRecords = MemoryRecords.EMPTY; } else { if (magic > RecordBatch.MAGIC_VALUE_V1) this.actualCompressionRatio = (float) writeDefaultBatchHeader() / this.writtenUncompressed; else if (compressionType != CompressionType.NONE) this.actualCompressionRatio = (float) writeLegacyCompressedWrapperHeader() / this.writtenUncompressed; ByteBuffer buffer = buffer().duplicate(); buffer.flip(); buffer.position(initialPosition); builtRecords = MemoryRecords.readableRecords(buffer.slice()); } }
MemoryRecordsBuilder { public void close() { if (aborted) throw new IllegalStateException("Cannot close MemoryRecordsBuilder as it has already been aborted"); if (builtRecords != null) return; validateProducerState(); closeForRecordAppends(); if (numRecords == 0L) { buffer().position(initialPosition); builtRecords = MemoryRecords.EMPTY; } else { if (magic > RecordBatch.MAGIC_VALUE_V1) this.actualCompressionRatio = (float) writeDefaultBatchHeader() / this.writtenUncompressed; else if (compressionType != CompressionType.NONE) this.actualCompressionRatio = (float) writeLegacyCompressedWrapperHeader() / this.writtenUncompressed; ByteBuffer buffer = buffer().duplicate(); buffer.flip(); buffer.position(initialPosition); builtRecords = MemoryRecords.readableRecords(buffer.slice()); } } }
MemoryRecordsBuilder { public void close() { if (aborted) throw new IllegalStateException("Cannot close MemoryRecordsBuilder as it has already been aborted"); if (builtRecords != null) return; validateProducerState(); closeForRecordAppends(); if (numRecords == 0L) { buffer().position(initialPosition); builtRecords = MemoryRecords.EMPTY; } else { if (magic > RecordBatch.MAGIC_VALUE_V1) this.actualCompressionRatio = (float) writeDefaultBatchHeader() / this.writtenUncompressed; else if (compressionType != CompressionType.NONE) this.actualCompressionRatio = (float) writeLegacyCompressedWrapperHeader() / this.writtenUncompressed; ByteBuffer buffer = buffer().duplicate(); buffer.flip(); buffer.position(initialPosition); builtRecords = MemoryRecords.readableRecords(buffer.slice()); } } MemoryRecordsBuilder(ByteBufferOutputStream bufferStream,
                                byte magic,
                                CompressionType compressionType,
                                TimestampType timestampType,
                                long baseOffset,
                                long logAppendTime,
                                long producerId,
                                short producerEpoch,
                                int baseSequence,
                                boolean isTransactional,
                                boolean isControlBatch,
                                int partitionLeaderEpoch,
                                int writeLimit);  MemoryRecordsBuilder(ByteBuffer buffer,
                                byte magic,
                                CompressionType compressionType,
                                TimestampType timestampType,
                                long baseOffset,
                                long logAppendTime,
                                long producerId,
                                short producerEpoch,
                                int baseSequence,
                                boolean isTransactional,
                                boolean isControlBatch,
                                int partitionLeaderEpoch,
                                int writeLimit); }
MemoryRecordsBuilder { public void close() { if (aborted) throw new IllegalStateException("Cannot close MemoryRecordsBuilder as it has already been aborted"); if (builtRecords != null) return; validateProducerState(); closeForRecordAppends(); if (numRecords == 0L) { buffer().position(initialPosition); builtRecords = MemoryRecords.EMPTY; } else { if (magic > RecordBatch.MAGIC_VALUE_V1) this.actualCompressionRatio = (float) writeDefaultBatchHeader() / this.writtenUncompressed; else if (compressionType != CompressionType.NONE) this.actualCompressionRatio = (float) writeLegacyCompressedWrapperHeader() / this.writtenUncompressed; ByteBuffer buffer = buffer().duplicate(); buffer.flip(); buffer.position(initialPosition); builtRecords = MemoryRecords.readableRecords(buffer.slice()); } } MemoryRecordsBuilder(ByteBufferOutputStream bufferStream,
                                byte magic,
                                CompressionType compressionType,
                                TimestampType timestampType,
                                long baseOffset,
                                long logAppendTime,
                                long producerId,
                                short producerEpoch,
                                int baseSequence,
                                boolean isTransactional,
                                boolean isControlBatch,
                                int partitionLeaderEpoch,
                                int writeLimit);  MemoryRecordsBuilder(ByteBuffer buffer,
                                byte magic,
                                CompressionType compressionType,
                                TimestampType timestampType,
                                long baseOffset,
                                long logAppendTime,
                                long producerId,
                                short producerEpoch,
                                int baseSequence,
                                boolean isTransactional,
                                boolean isControlBatch,
                                int partitionLeaderEpoch,
                                int writeLimit); ByteBuffer buffer(); int initialCapacity(); double compressionRatio(); CompressionType compressionType(); boolean isControlBatch(); boolean isTransactional(); MemoryRecords build(); RecordsInfo info(); void setProducerState(long producerId, short producerEpoch, int baseSequence, boolean isTransactional); void overrideLastOffset(long lastOffset); void closeForRecordAppends(); void abort(); void close(); Long appendWithOffset(long offset, long timestamp, byte[] key, byte[] value, Header[] headers); Long appendWithOffset(long offset, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers); Long appendWithOffset(long offset, long timestamp, byte[] key, byte[] value); Long appendWithOffset(long offset, long timestamp, ByteBuffer key, ByteBuffer value); Long appendWithOffset(long offset, SimpleRecord record); Long append(long timestamp, ByteBuffer key, ByteBuffer value); Long append(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers); Long append(long timestamp, byte[] key, byte[] value); Long append(long timestamp, byte[] key, byte[] value, Header[] headers); Long append(SimpleRecord record); Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker); void appendUncheckedWithOffset(long offset, LegacyRecord record); void append(Record record); void appendWithOffset(long offset, Record record); void appendWithOffset(long offset, LegacyRecord record); void append(LegacyRecord record); void setEstimatedCompressionRatio(float estimatedCompressionRatio); boolean hasRoomFor(long timestamp, byte[] key, byte[] value, Header[] headers); boolean hasRoomFor(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers); boolean isClosed(); boolean isFull(); int sizeInBytes(); byte magic(); long producerId(); short producerEpoch(); }
MemoryRecordsBuilder { public void close() { if (aborted) throw new IllegalStateException("Cannot close MemoryRecordsBuilder as it has already been aborted"); if (builtRecords != null) return; validateProducerState(); closeForRecordAppends(); if (numRecords == 0L) { buffer().position(initialPosition); builtRecords = MemoryRecords.EMPTY; } else { if (magic > RecordBatch.MAGIC_VALUE_V1) this.actualCompressionRatio = (float) writeDefaultBatchHeader() / this.writtenUncompressed; else if (compressionType != CompressionType.NONE) this.actualCompressionRatio = (float) writeLegacyCompressedWrapperHeader() / this.writtenUncompressed; ByteBuffer buffer = buffer().duplicate(); buffer.flip(); buffer.position(initialPosition); builtRecords = MemoryRecords.readableRecords(buffer.slice()); } } MemoryRecordsBuilder(ByteBufferOutputStream bufferStream,
                                byte magic,
                                CompressionType compressionType,
                                TimestampType timestampType,
                                long baseOffset,
                                long logAppendTime,
                                long producerId,
                                short producerEpoch,
                                int baseSequence,
                                boolean isTransactional,
                                boolean isControlBatch,
                                int partitionLeaderEpoch,
                                int writeLimit);  MemoryRecordsBuilder(ByteBuffer buffer,
                                byte magic,
                                CompressionType compressionType,
                                TimestampType timestampType,
                                long baseOffset,
                                long logAppendTime,
                                long producerId,
                                short producerEpoch,
                                int baseSequence,
                                boolean isTransactional,
                                boolean isControlBatch,
                                int partitionLeaderEpoch,
                                int writeLimit); ByteBuffer buffer(); int initialCapacity(); double compressionRatio(); CompressionType compressionType(); boolean isControlBatch(); boolean isTransactional(); MemoryRecords build(); RecordsInfo info(); void setProducerState(long producerId, short producerEpoch, int baseSequence, boolean isTransactional); void overrideLastOffset(long lastOffset); void closeForRecordAppends(); void abort(); void close(); Long appendWithOffset(long offset, long timestamp, byte[] key, byte[] value, Header[] headers); Long appendWithOffset(long offset, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers); Long appendWithOffset(long offset, long timestamp, byte[] key, byte[] value); Long appendWithOffset(long offset, long timestamp, ByteBuffer key, ByteBuffer value); Long appendWithOffset(long offset, SimpleRecord record); Long append(long timestamp, ByteBuffer key, ByteBuffer value); Long append(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers); Long append(long timestamp, byte[] key, byte[] value); Long append(long timestamp, byte[] key, byte[] value, Header[] headers); Long append(SimpleRecord record); Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker); void appendUncheckedWithOffset(long offset, LegacyRecord record); void append(Record record); void appendWithOffset(long offset, Record record); void appendWithOffset(long offset, LegacyRecord record); void append(LegacyRecord record); void setEstimatedCompressionRatio(float estimatedCompressionRatio); boolean hasRoomFor(long timestamp, byte[] key, byte[] value, Header[] headers); boolean hasRoomFor(long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers); boolean isClosed(); boolean isFull(); int sizeInBytes(); byte magic(); long producerId(); short producerEpoch();  }