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 shouldFailOnProductionErrorByDefault() {
    final KsqlConfig ksqlConfig = new KsqlConfig(Collections.emptyMap());
    final Object result = ksqlConfig.getKsqlStreamConfigProps()
        .get(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG);
    assertThat(result, equalTo(LogAndFailProductionExceptionHandler.class));
  }