@Override
    public KV<Object, Object> apply(SamzaSqlRelMessage message) {
      long beginProcessing = System.nanoTime();
      KV<Object, Object> retKV = this.samzaMsgConverter.convertToSamzaMessage(message);
      if (message.getSamzaSqlRelRecord().containsField(SamzaSqlRelMessage.OP_NAME)
          && ((String) message.getSamzaSqlRelRecord().getField(SamzaSqlRelMessage.OP_NAME).get()).equalsIgnoreCase(
          SamzaSqlRelMessage.DELETE_OP)) {
        // If it is a delete op. Set the payload to null so that the record gets deleted.
        retKV = new KV<>(retKV.key, null);
      }
      updateMetrics(beginProcessing, System.nanoTime(), message.getSamzaSqlRelMsgMetadata());
      return retKV;
    }
TODO: [SAMZA-2031]: the time-based metrics here for insert and query are
currently not accurate because they don't include the time of sendTo() call
It is not feasible to include it because sendTo operator does not return
a stream to process its messages to update hte metrics.
samza-sql/src/main/java/org/apache/samza/sql/translator/QueryTranslator.java