@Override
  public void onCreate(
      final ServiceContext serviceContext,
      final MetaStore metaStore,
      final QueryMetadata queryMetadata) {
    if (perQuery.containsKey(queryMetadata.getQueryId())) {
      return;
    }
    perQuery.put(
        queryMetadata.getQueryId(),
        new PerQueryListener(
            metrics,
            metricsPrefix,
            queryMetadata.getQueryId().toString(),
            metricsTags
        )
    );
  }
@Test
  public void shouldAddMetricWithSuppliedPrefix() {
    // Given:
    final String groupPrefix = "some-prefix-";
    final Map<String, String> tags = new HashMap<>(metricsTags);
    tags.put("status", TAG);

    clearInvocations(metrics);

    // When:
    listener = new QueryStateMetricsReportingListener(metrics, groupPrefix, metricsTags);
    listener.onCreate(serviceContext, metaStore, query);

    // Then:
    verify(metrics).metricName("query-status", groupPrefix + "ksql-queries",
        "The current Kafka Streams status of the given query.",
        tags);
    verify(metrics).metricName("error-status", groupPrefix + "ksql-queries",
        "The current error status of the given query, if the state is in ERROR state",
        tags);
    verify(metrics).metricName("ksql-query-status", groupPrefix + "ksql-queries",
            "The current ksqlDB status of the given query.",
            tags);
    tags.put("query-id", QUERY_ID.toString());
    verify(metrics).metricName(QueryStateMetricsReportingListener.QUERY_RESTART_METRIC_NAME, "some-prefix-ksql-queries",
        QueryStateMetricsReportingListener.QUERY_RESTART_METRIC_DESCRIPTION,
        tags);
  }