@Override
  public Object apply(IncomingMessageEnvelope ime) {
    Assert.notNull(ime, "ime is null");
    KV<Object, Object> keyAndMessageKV = KV.of(ime.getKey(), ime.getMessage());
    SamzaSqlRelMsgMetadata metadata = new SamzaSqlRelMsgMetadata(ime.getEventTime(), ime.getArrivalTime());
    SamzaSqlInputMessage samzaMsg = SamzaSqlInputMessage.of(keyAndMessageKV, metadata);
    return  samzaMsg;
  }
SamzaSqlInputTransformer:
  Input Transformer for SamzaSQL that consumes {@link IncomingMessageEnvelope} (IME) and produces
  {@link SamzaSqlInputMessage} so that the event metadata (currently eventTime and arrivalTime) are copied
  from the IME to the SamzaSQL layer through the {@link SamzaSqlInputMessage}
samza-sql/src/main/java/org/apache/samza/sql/SamzaSqlInputTransformer.java