void put(SystemStreamPartition ssp, Object key, Object message) {
    List<IncomingMessageEnvelope> messages = bufferedMessages.get(ssp);
    String offset = String.valueOf(messages.size());

    if (shouldUseEndOfStreamOffset(message)) {
      offset = IncomingMessageEnvelope.END_OF_STREAM_OFFSET;
    }

    IncomingMessageEnvelope messageEnvelope = new IncomingMessageEnvelope(ssp, offset, key, message,
        0, 0L, Instant.now().toEpochMilli());
    bufferedMessages.get(ssp)
        .add(messageEnvelope);
  }
Handles the produce request from {@link InMemorySystemProducer} and populates the underlying message queue.

@param ssp system stream partition
@param key key for message produced
@param message actual payload
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemoryManager.java