Use pooled direct memory allocator when decoding Pulsar entry to Kafka records

[BewareMyPower]

Motivation

When a Pulsar entry is decoded to Kafka record in ByteBufUtils#decodePulsarEntryToKafkaRecords, a NIO buffer whose initial capacity is 1 MB will be allocated from heap memory. Therefore, each time an entry is read, 1 MB heap memory will be allocated. Then the heap memory will increase very quickly and GC will happen frequently.

Kafka MemoryRecordsBuilder uses its underlying ByteBufferOutputStream field as the internal buffer whose capacity can be increased in write method. Even if a direct buffer was allocated by Netty's pooled direct memory allocator and its underlying ByteBuffer was passed to ByteBufferOutputStream's constructor, if the reallocation happened, the new buffer could still be allocated from heap memory.

Modification

This PR adds a DirectBufferOutputStream class that inherits from ByteBufferOutputStream and overrides some methods that can be called in MemoryRecordsBuilder. This class uses Pulsar's default ByteBufAllocator to allocate memory. The other methods' behaviors are the same with ByteBufferOutputStream.

A unit test is added to verify that the MemoryRecordsBuilder will build the same records no matter the underlying ByteBufferOutputStream is ByteBufferOutputStream or DirectBufferOutputStream. Three cases are tested in this test:

1. The initial capacity is less than the size of records header, in this case, position(int) method will be called to increase the capacity.
2. The initial capacity is greater than both the size of records header and the total size of records.
3. The initial capacity is greater than the size of records header but less than the total size of records, in this case, write() method will increase the capacity automatically.

Then, a DirectBufferOutputStream instance is passed to MemoryRecordsBuilder's constructor in ByteBufUtils#decodePulsarEntryToKafkaRecords and the return value's type is changed to DecodeResult because we need to release the ByteBuf later.

[BewareMyPower]

Here're the comparison between KoP before this PR and KoP after this PR