public Map<String, Object> getProducerClientConfigProps() {
    final Map<String, Object> map = new HashMap<>();
    map.putAll(getConfigsFor(ProducerConfig.configNames()));
    // producer client metrics aren't used in Confluent deployment
    possiblyConfigureConfluentTelemetry(map);
    return Collections.unmodifiableMap(map);
  }
@Test
  public void shouldFilterProducerConfigs() {
    // Given:
    final Map<String, Object> configs = new HashMap<>();
    configs.put(ProducerConfig.ACKS_CONFIG, "all");
    configs.put(ProducerConfig.CLIENT_ID_CONFIG, null);
    configs.put("not.a.config", "123");

    final KsqlConfig ksqlConfig = new KsqlConfig(configs);

    // When:
    assertThat(ksqlConfig.getProducerClientConfigProps(), hasEntry(ProducerConfig.ACKS_CONFIG, "all"));
    assertThat(ksqlConfig.getProducerClientConfigProps(), hasEntry(ProducerConfig.CLIENT_ID_CONFIG, null));
    assertThat(ksqlConfig.getProducerClientConfigProps(), not(hasKey("not.a.config")));
  }