KAFKA-16738: Returns BaseRecords instead of MemoryRecords

[luozhenyu]

Related to https://issues.apache.org/jira/browse/KAFKA-16738

We can write a record which is a subtype of BaseRecords, but we can not read a record which is a subtype of BaseRecords. If we change the return type of Readable#readRecords from MemoryRecords to BaseRecords, we can override the implementation of readRecords and returns a subtype of BaseRecords easily.

We known that the MemoryRecords is based on JDK's ByteBuffer. We are developing a netty project(kroxylicious) and we want to create a subtype of BaseRecords like MemoryRecords based on netty's ByteBuf.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[chia7712]

in kafak serialization, we assume the impl of BaseRecords is MemoryRecords. For example:

        @Override
        public PartitionProduceData duplicate() {
            PartitionProduceData _duplicate = new PartitionProduceData();
            _duplicate.index = index;
            if (records == null) {
                _duplicate.records = null;
            } else {
                _duplicate.records = MemoryRecords.readableRecords(((MemoryRecords) records).buffer().duplicate());
            }
            return _duplicate;
        }

Hence, I'm not sure how this change works if you introduce a non-MemoryRecords impl.

[luozhenyu]

Hence, I'm not sure how this change works if you introduce a non-MemoryRecords impl.

Hard-coded duplicating of MemoryRecords is not elegant. So I add a duplicate method to BaseRecords. How do you think?

        @Override
        public PartitionProduceData duplicate() {
            PartitionProduceData _duplicate = new PartitionProduceData();
            _duplicate.index = index;
            if (records == null) {
                _duplicate.records = null;
            } else {
                _duplicate.records = records.duplicate();
            }
            return _duplicate;
        }

[chia7712]

Hard-coded duplicating of MemoryRecords is not elegant. So I add a duplicate method to BaseRecords. How do you think?

Could you file a KIP to let us know the whole picture? MemoryRecords is used in code base, so we need to be careful of changing that.