@Override
  public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) {
    inputEvents.inc();
    message.getSamzaSqlRelMsgMetadata().joinStartTimeMs = Instant.now().toEpochMilli();
    return message;
  }
update metrics given a message
@param message  the input message
@return the same message
samza-sql/src/main/java/org/apache/samza/sql/translator/TranslatorInputMetricsMapFunction.java