private void updateMetrics(long beginProcessing, long endProcessing, SamzaSqlRelMsgMetadata metadata) {
      /* insert (SendToOutputStream) metrics */
      insertProcessingTime.update(endProcessing - beginProcessing);
      /* query metrics */
      queryOutputEvents.inc();
      /* TODO: remove scanTime validation once code to assign it is stable */
      Validate.isTrue(metadata.hasScanTime());
      queryLatencyNs.update(System.nanoTime() - metadata.getScanTimeNanos());
      /** TODO: change if hasArrivalTime to validation once arrivalTime is assigned,
       and later remove the check once code is stable */
      if (metadata.hasArrivalTime()) {
        queuingLatencyMS.update(metadata.getScanTimeMillis() - metadata.getArrivalTime());
      }
      /* since availability of eventTime depends on source, we need the following check */
      if (metadata.hasEventTime()) {
        totalLatencyMs.update(System.currentTimeMillis() - metadata.getEventTime());
      }
    }
Updates the Diagnostics Metrics (processing time and number of events)
@param beginProcessing when sendOutput Started processing this message
@param endProcessing when sendOutput finished processing this message
@param metadata the event's message metadata
samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java