protected void setSource(OutgoingMessageEnvelope envelope, IndexRequest indexRequest) {
    Object message = envelope.getMessage();
    if (message instanceof byte[]) {
      indexRequest.source((byte[]) message);
    } else if (message instanceof Map) {
      indexRequest.source((Map) message);
    } else {
      throw new SamzaException("Unsupported message type: " + message.getClass().getCanonicalName());
    }
  }
The default {@link IndexRequestFactory}.

<p>Samza concepts are mapped to Elastic search concepts as follows:</p>

<ul>
  <li>
    The index and type are derived from the stream name using
    the following the pattern "{index-name}/{type-name}".
  </li>
  <li>
    The id of the document is the {@link String} representation of the
    {@link OutgoingMessageEnvelope#getKey()} from {@link Object#toString()} if provided.</li>
  <li>
    The source of the document is set from the {@link OutgoingMessageEnvelope#getMessage()}.
    Supported types are {@link byte[]} and {@link Map} which are both
    passed on without serialising.
  </li>
  <li>
    The routing key is set from {@link String} representation of the
    {@link OutgoingMessageEnvelope#getPartitionKey()} from {@link Object#toString()} if provided.
  </li>
</ul>
samza-elasticsearch/src/main/java/org/apache/samza/system/elasticsearch/indexrequest/DefaultIndexRequestFactory.java