public Map<String, Object> getKsqlStreamConfigProps(final String applicationId) {
    final Map<String, Object> map = new HashMap<>(getKsqlStreamConfigProps());
    map.put(
        MetricCollectors.RESOURCE_LABEL_PREFIX
            + StreamsConfig.APPLICATION_ID_CONFIG,
        applicationId
    );

    // Streams client metrics aren't used in Confluent deployment
    possiblyConfigureConfluentTelemetry(map);
    return Collections.unmodifiableMap(map);
  }
@Test
  public void shouldSetMonitoringInterceptorConfigPropertiesByClientType() {
    // Given:
    final Map<String, String> props = ImmutableMap.of(
        "ksql.streams.consumer.confluent.monitoring.interceptor.topic", "foo",
        "producer.confluent.monitoring.interceptor.topic", "bar"
    );

    final KsqlConfig ksqlConfig = new KsqlConfig(props);

    // When:
    final Map<String, Object> result = ksqlConfig.getKsqlStreamConfigProps();

    // Then:
    assertThat(result.get("consumer.confluent.monitoring.interceptor.topic"), is("foo"));
    assertThat(result.get("producer.confluent.monitoring.interceptor.topic"), is("bar"));
  }