void put(SystemStreamPartition ssp, IncomingMessageEnvelope envelope) {
    Preconditions.checkNotNull(envelope);
    Preconditions.checkNotNull(envelope.getOffset());
    List<IncomingMessageEnvelope> messages = bufferedMessages.get(ssp);
    String offset = String.valueOf(messages.size());
    if (!envelope.getOffset().equals(offset)) {
      throw new SamzaException(
          String.format("Offset mismatch for ssp %s, expected %s found %s, please set the correct offset", ssp, offset, envelope.getOffset()));
    }
    bufferedMessages.get(ssp).add(envelope);
  }
Handles produce request from {@link InMemorySystemProducer} for case where a job has a custom IME and
populates the underlying message queue with the IME.
Note: Offset in the envelope needs to be in the increasing order for envelopes in the same ssp and needs to
start at 0 for the first envelope, otherwise poll logic will be impacted

@param ssp system stream partition
@param envelope incoming message envelope
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java