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 shouldSetStreamsConfigKsqlTopicPrefixedProperties() {
    final KsqlConfig ksqlConfig = new KsqlConfig(
        Collections.singletonMap(
            KsqlConfig.KSQL_STREAMS_PREFIX + StreamsConfig.TOPIC_PREFIX + TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, 2));

    assertThat(ksqlConfig.getKsqlStreamConfigProps()
            .get(StreamsConfig.TOPIC_PREFIX + TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG),
        equalTo(2));

    assertThat(ksqlConfig.getKsqlStreamConfigProps()
            .get(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG),
        is(nullValue()));
  }