@Test(expected = IllegalArgumentException.class) public void testWriteEndTxnMarkerNonControlBatch() { ByteBuffer buffer = ByteBuffer.allocate(128); buffer.position(bufferOffset); long pid = 9809; short epoch = 15; int sequence = RecordBatch.NO_SEQUENCE; 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.appendEndTxnMarker(RecordBatch.NO_TIMESTAMP, new EndTransactionMarker(ControlRecordType.ABORT, 0)); }
public Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker) { if (producerId == RecordBatch.NO_PRODUCER_ID) throw new IllegalArgumentException("End transaction marker requires a valid producerId"); if (!isTransactional) throw new IllegalArgumentException("End transaction marker depends on batch transactional flag being enabled"); ByteBuffer value = marker.serializeValue(); return appendControlRecord(timestamp, marker.controlType(), value); }
MemoryRecordsBuilder { public Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker) { if (producerId == RecordBatch.NO_PRODUCER_ID) throw new IllegalArgumentException("End transaction marker requires a valid producerId"); if (!isTransactional) throw new IllegalArgumentException("End transaction marker depends on batch transactional flag being enabled"); ByteBuffer value = marker.serializeValue(); return appendControlRecord(timestamp, marker.controlType(), value); } }
MemoryRecordsBuilder { public Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker) { if (producerId == RecordBatch.NO_PRODUCER_ID) throw new IllegalArgumentException("End transaction marker requires a valid producerId"); if (!isTransactional) throw new IllegalArgumentException("End transaction marker depends on batch transactional flag being enabled"); ByteBuffer value = marker.serializeValue(); return appendControlRecord(timestamp, marker.controlType(), value); } 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 Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker) { if (producerId == RecordBatch.NO_PRODUCER_ID) throw new IllegalArgumentException("End transaction marker requires a valid producerId"); if (!isTransactional) throw new IllegalArgumentException("End transaction marker depends on batch transactional flag being enabled"); ByteBuffer value = marker.serializeValue(); return appendControlRecord(timestamp, marker.controlType(), value); } 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 Long appendEndTxnMarker(long timestamp, EndTransactionMarker marker) { if (producerId == RecordBatch.NO_PRODUCER_ID) throw new IllegalArgumentException("End transaction marker requires a valid producerId"); if (!isTransactional) throw new IllegalArgumentException("End transaction marker depends on batch transactional flag being enabled"); ByteBuffer value = marker.serializeValue(); return appendControlRecord(timestamp, marker.controlType(), value); } 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();  }