KsqlConfig fails to propagate monitoring interceptor configs prefixed with "producer." or "consumer."

[vcrfxia]

Configs prefixed with confluent.monitoring.interceptor. are successfully passed by KSQL to KStreams, as evidenced by this test, but configs prefixed with producer.confluent.monitoring.interceptor. or consumer.confluent.monitoring.interceptor. fail to be passed through.

Thus, users with the same producer and consumer monitoring interceptor configs have the workaround of using the confluent.monitoring.interceptor. prefix, rather than producer.confluent.monitoring.interceptor. and consumer.confluent.monitoring.interceptor. separately, but this bug should be fixed so users may set different configs for the producer and consumer monitoring interceptors.

[apurvam]

Note this worked in 5.0.x, but regressed in 5.1.x.

[stevenpyzhang]

I think the regression occurred as a result of changes in the resolveStreamConfig that filtered out unresolved producer. and consumer. prefixed configs.
https://github.com/confluentinc/ksql/blob/5.0.3-post/ksql-common/src/main/java/io/confluent/ksql/util/KsqlConfig.java#L317

https://github.com/confluentinc/ksql/blob/5.1.0-post/ksql-common/src/main/java/io/confluent/ksql/config/KsqlConfigResolver.java#L74

Are there any examples of configs using producer.confluent.monitoring.interceptor. and consumer.confluent.monitoring.interceptor. as a prefix? I couldn't find any in the documentation for those. I did find docs on producer.interceptor.classes and consumer.interceptor.classes
Is this something separate from this issue or are these the actual config values that should be used?
https://docs.confluent.io/current/control-center/installation/clients.html#kstreams

confluent.monitoring.interceptor. has some examples here
https://docs.confluent.io/current/control-center/installation/clients.html#general-options

On another note: confluent.monitoring.interceptor. prefixed configs are actually not resolved properly in the resolveStreamConfig function, but will still be passed through if the strict flag is false.
https://github.com/confluentinc/ksql/blob/5.1.0-post/ksql-common/src/test/java/io/confluent/ksql/config/KsqlConfigResolverTest.java#L166

[slepkin]

This still seems to be a problem in 5.2.2. When starting the server, we see KsqlConfig and StreamsConfig, but no ProducerConfig or ConsumerConfig. This makes it impossible to change the value of acks, which makes it difficult to guarantee at-least-once processing.

[big-andy-coates]

This still seems to be a problem in 5.2.2. When starting the server, we see KsqlConfig and StreamsConfig, but no ProducerConfig or ConsumerConfig. This makes it impossible to change the value of acks, which makes it difficult to guarantee at-least-once processing.

hi @slepkin

I'm not sure this is the case. We actually have a test to ensure you can set the acks, see

ksql/ksql-common/src/test/java/io/confluent/ksql/util/KsqlConfigTest.java

Lines 581 to 594 in 97fc956

	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")));
	}

If after double checking you're still having issues setting the ack level, can you raise a new Github issue please, as I don't believe your issue and this issue are related - unless you're prefixing the setting?. Please include relevant info, e.g. KSQL version, ksql-server.properties, commands youi're running, logs etc.

Thanks!