@Override
  public SamzaSqlRelMessage apply(SamzaSqlRelMessage message) {
    Instant endProcessing = Instant.now();
    Instant beginProcessing = Instant.ofEpochMilli(message.getSamzaSqlRelMsgMetadata().joinStartTimeMs);
    outputEvents.inc();
    processingTime.update(Duration.between(beginProcessing, endProcessing).toMillis());
    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/TranslatorOutputMetricsMapFunction.java