@Test(expected = IllegalArgumentException.class) public void testAppendAtInvalidOffset() { ByteBuffer buffer = ByteBuffer.allocate(1024); buffer.position(bufferOffset); long logAppendTime = System.currentTimeMillis(); MemoryRecordsBuilder builder = new MemoryRecordsBuilder(buffer, RecordBatch.MAGIC_VALUE_V1, compressionType, TimestampType.CREATE_TIME, 0L, logAppendTime, RecordBatch.NO_PRODUCER_ID, RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, false, false, RecordBatch.NO_PARTITION_LEADER_EPOCH, buffer.capacity()); builder.appendWithOffset(0L, System.currentTimeMillis(), "a".getBytes(), null); builder.appendWithOffset(0L, System.currentTimeMillis(), "b".getBytes(), null); }
private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) { try { if (isControlRecord != isControlBatch) throw new IllegalArgumentException("Control records can only be appended to control batches"); if (lastOffset != null && offset <= lastOffset) throw new IllegalArgumentException(String.format("Illegal offset %s following previous offset %s " + "(Offsets must increase monotonically).", offset, lastOffset)); if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP) throw new IllegalArgumentException("Invalid negative timestamp " + timestamp); if (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0) throw new IllegalArgumentException("Magic v" + magic + " does not support record headers"); if (baseTimestamp == null) baseTimestamp = timestamp; if (magic > RecordBatch.MAGIC_VALUE_V1) { appendDefaultRecord(offset, timestamp, key, value, headers); return null; } else { return appendLegacyRecord(offset, timestamp, key, value); } } catch (IOException e) { throw new KafkaException("I/O exception when writing to the append stream, closing", e); } }
MemoryRecordsBuilder { private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) { try { if (isControlRecord != isControlBatch) throw new IllegalArgumentException("Control records can only be appended to control batches"); if (lastOffset != null && offset <= lastOffset) throw new IllegalArgumentException(String.format("Illegal offset %s following previous offset %s " + "(Offsets must increase monotonically).", offset, lastOffset)); if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP) throw new IllegalArgumentException("Invalid negative timestamp " + timestamp); if (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0) throw new IllegalArgumentException("Magic v" + magic + " does not support record headers"); if (baseTimestamp == null) baseTimestamp = timestamp; if (magic > RecordBatch.MAGIC_VALUE_V1) { appendDefaultRecord(offset, timestamp, key, value, headers); return null; } else { return appendLegacyRecord(offset, timestamp, key, value); } } catch (IOException e) { throw new KafkaException("I/O exception when writing to the append stream, closing", e); } } }
MemoryRecordsBuilder { private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) { try { if (isControlRecord != isControlBatch) throw new IllegalArgumentException("Control records can only be appended to control batches"); if (lastOffset != null && offset <= lastOffset) throw new IllegalArgumentException(String.format("Illegal offset %s following previous offset %s " + "(Offsets must increase monotonically).", offset, lastOffset)); if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP) throw new IllegalArgumentException("Invalid negative timestamp " + timestamp); if (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0) throw new IllegalArgumentException("Magic v" + magic + " does not support record headers"); if (baseTimestamp == null) baseTimestamp = timestamp; if (magic > RecordBatch.MAGIC_VALUE_V1) { appendDefaultRecord(offset, timestamp, key, value, headers); return null; } else { return appendLegacyRecord(offset, timestamp, key, value); } } catch (IOException e) { throw new KafkaException("I/O exception when writing to the append stream, closing", e); } } 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 { private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) { try { if (isControlRecord != isControlBatch) throw new IllegalArgumentException("Control records can only be appended to control batches"); if (lastOffset != null && offset <= lastOffset) throw new IllegalArgumentException(String.format("Illegal offset %s following previous offset %s " + "(Offsets must increase monotonically).", offset, lastOffset)); if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP) throw new IllegalArgumentException("Invalid negative timestamp " + timestamp); if (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0) throw new IllegalArgumentException("Magic v" + magic + " does not support record headers"); if (baseTimestamp == null) baseTimestamp = timestamp; if (magic > RecordBatch.MAGIC_VALUE_V1) { appendDefaultRecord(offset, timestamp, key, value, headers); return null; } else { return appendLegacyRecord(offset, timestamp, key, value); } } catch (IOException e) { throw new KafkaException("I/O exception when writing to the append stream, closing", e); } } 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 { private Long appendWithOffset(long offset, boolean isControlRecord, long timestamp, ByteBuffer key, ByteBuffer value, Header[] headers) { try { if (isControlRecord != isControlBatch) throw new IllegalArgumentException("Control records can only be appended to control batches"); if (lastOffset != null && offset <= lastOffset) throw new IllegalArgumentException(String.format("Illegal offset %s following previous offset %s " + "(Offsets must increase monotonically).", offset, lastOffset)); if (timestamp < 0 && timestamp != RecordBatch.NO_TIMESTAMP) throw new IllegalArgumentException("Invalid negative timestamp " + timestamp); if (magic < RecordBatch.MAGIC_VALUE_V2 && headers != null && headers.length > 0) throw new IllegalArgumentException("Magic v" + magic + " does not support record headers"); if (baseTimestamp == null) baseTimestamp = timestamp; if (magic > RecordBatch.MAGIC_VALUE_V1) { appendDefaultRecord(offset, timestamp, key, value, headers); return null; } else { return appendLegacyRecord(offset, timestamp, key, value); } } catch (IOException e) { throw new KafkaException("I/O exception when writing to the append stream, closing", e); } } 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();  }