protected int getRecordSize(ConsumerRecord<K, V> r) {
    int keySize = (r.key() == null) ? 0 : r.serializedKeySize();
    return keySize + r.serializedValueSize();
  }
Protected to help extensions of this class build {@link IncomingMessageEnvelope}s.
@param r consumer record to size
@return the size of the serialized record
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaConsumerProxy.java