Map<TopicPartition, Integer> partitionSizes() {
        if (partitionSizes == null) {
            // this method may be called by different thread (see the comment on data)
            synchronized (this) {
                if (partitionSizes == null) {
                    Map<TopicPartition, Integer> tmpPartitionSizes = new HashMap<>();
                    data.topicData().forEach(topicData ->
                        topicData.partitionData().forEach(partitionData ->
                            tmpPartitionSizes.compute(new TopicPartition(topicData.name(), partitionData.index()),
                                (ignored, previousValue) ->
                                    partitionData.records().sizeInBytes() + (previousValue == null ? 0 : previousValue))
                        )
                    );
                    partitionSizes = tmpPartitionSizes;
                }
            }
        }
        return partitionSizes;
    }
We have to copy acks, timeout, transactionalId and partitionSizes from data since data maybe reset to eliminate
the reference to ByteBuffer but those metadata are still useful.
clients/src/main/java/org/apache/kafka/common/requests/ProduceRequest.java