@Override
    public void init(Context context) {
      TranslatorContext translatorContext =
          ((SamzaSqlApplicationContext) context.getApplicationTaskContext()).getTranslatorContexts().get(queryId);
      this.samzaMsgConverter = translatorContext.getMsgConverter(outputTopic);
      ContainerContext containerContext = context.getContainerContext();
      metricsRegistry = containerContext.getContainerMetricsRegistry();
      /* insert (SendToOutputStream) metrics */
      insertProcessingTime =
          new SamzaHistogram(metricsRegistry, insertLogicalId, TranslatorConstants.TOTAL_LATENCY_MILLIS_NAME);

      /* query metrics */
      totalLatencyMs = new SamzaHistogram(metricsRegistry, queryLogicalId, TranslatorConstants.TOTAL_LATENCY_MILLIS_NAME);

      queryLatencyNs = new SamzaHistogram(metricsRegistry, queryLogicalId, TranslatorConstants.QUERY_LATENCY_NANOS_NAME);
      queuingLatencyMS = new SamzaHistogram(metricsRegistry, queryLogicalId,
          TranslatorConstants.QUEUEING_LATENCY_MILLIS_NAME);
      queryOutputEvents = metricsRegistry.newCounter(queryLogicalId, TranslatorConstants.OUTPUT_EVENTS_NAME);
      queryOutputEvents.clear();
    }
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