protected IncomingMessageEnvelope handleNewRecord(ConsumerRecord<K, V> consumerRecord,
      SystemStreamPartition systemStreamPartition) {
    return new IncomingMessageEnvelope(systemStreamPartition, String.valueOf(consumerRecord.offset()),
        consumerRecord.key(), consumerRecord.value(), getRecordSize(consumerRecord), consumerRecord.timestamp(),
        Instant.now().toEpochMilli());
  }
Convert a {@link ConsumerRecord} to an {@link IncomingMessageEnvelope}. This may also execute some other custom
logic for each new {@link IncomingMessageEnvelope}.

This has a protected visibility so that {@link KafkaConsumerProxy} can be extended to add special handling logic
for custom Kafka systems.

@param consumerRecord {@link ConsumerRecord} from Kafka that was consumed
@param systemStreamPartition {@link SystemStreamPartition} corresponding to the record
@return {@link IncomingMessageEnvelope} corresponding to the {@code consumerRecord}
samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaConsumerProxy.java