...

Currently we use Deserializer#deserialize(String topic, Headers headers, byte[] data) in Fetcher#parseRecord(TopicPartition, RecordBatch, Record) to deserialize key&value, we first call Utils.toArray(ByteBuffer) to convert ByteBuffer into byte[] and then call Deserializer#deserialize(String topic, Headers headers, byte[] data) which will cause memory allocation and memory copying. Actually, we can directly

The default implementation of this method would still call Utils.toArray(ByteBuffer) and then leverage on the existing method. But for the following cases we can use ByteBuffer instead of byte[] for deserialization, which will reduce memory allocation and memory copying in some cases.If we add the default method Deserializer#deserialize(String, Headers, ByteBuffer) and use it in Fetcher#parseRecord(TopicPartition, RecordBatch, Record), we can reduce the memory allocation and memory copy of StringDeserializer and ByteBufferDeserializer, of course if :

• For built-in StringDeserializer and ByteBufferDeserializer, we will do the deserialization directly on with the overloaded method with ByteBuffer.
• If user-customized Deserializers

...

• override this overloaded method's default implementation, they also can reduce memory allocation and memory copying.

Public Interfaces

We propose adding default method Deserializer#deserialize(String, Headers, ByteBuffer).

...