@Override
  public void send(String source, OutgoingMessageEnvelope envelope) {
    Object key = envelope.getKey();
    Object message = envelope.getMessage();

    Object partitionKey;
    // We use the partition key from message if available, if not fallback to message key or use message as partition
    // key as the final resort.
    if (envelope.getPartitionKey() != null) {
      partitionKey = envelope.getPartitionKey();
    } else if (key != null) {
      partitionKey = key;
    } else {
      partitionKey = message;
    }

    Preconditions.checkNotNull(partitionKey, "Failed to compute partition key for the message: " + envelope);

    int partition =
        Math.abs(hashCode(partitionKey)) % memoryManager.getPartitionCountForSystemStream(envelope.getSystemStream());

    SystemStreamPartition ssp = new SystemStreamPartition(envelope.getSystemStream(), new Partition(partition));
    memoryManager.put(ssp, key, message);
  }
Sends a specified message envelope from a specified Samza source.
@param source String representing the source of the message.
@param envelope Aggregate object representing the serialized message to send from the source.
samza-core/src/main/java/org/apache/samza/system/inmemory/InMemorySystemProducer.java