private void moveMessagesToTheirQueue(SystemStreamPartition ssp, List<IncomingMessageEnvelope> envelopes) {
    long nextOffset = nextOffsets.get(ssp);

    for (IncomingMessageEnvelope env : envelopes) {
      sink.addMessage(ssp, env);  // move message to the BlockingEnvelopeMap's queue

      LOG.trace("IncomingMessageEnvelope. got envelope with offset:{} for ssp={}", env.getOffset(), ssp);
      nextOffset = Long.valueOf(env.getOffset()) + 1;
    }

    nextOffsets.put(ssp, nextOffset);
  }
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