KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams #12988

ashmeet13 · 2022-12-14T07:36:36Z

Streams hard-codes a few configurations, currently the documentation refers to 5 such configs.

The four mentioned within - Parameters controlled by Kafka Streams + enable.auto.commit.
Three out of the 4 mentioned within the Parameters controlled by Kafka Streams are also present within Default Values which state the configurations that can be configured but will have a default value if not set.

This PR makes changes to warn the user when a configuration set by them is being overridden.
Due to the overlapping documentation I have gone through the code and have separated the configs in a few categories -

Non configurable consumer configs when EOS is disabled -

enable.auto.commit
allow.auto.create.topics

Non configurable consumer configs when EOS is enabled -

isolation.level

Non configurable producer configs when EOS is enabled -

enable.idempotence
max.in.flight.requests.per.connection
transactional.id

Default consumer configs which can be configured -

auto.offset.reset
max.poll.records

Default producer configs which can be configured -

linger.ms

StreamsConfig already had code to log warnings when a non-configurable property was being set.
It was missing logging warnings when allow.auto.create.topics was configured. I have added this change and removed the code that was setting allow.auto.create.topics within getMainConsumerConfigs.

Now it follows the existing code execution path to check for -

Missing default configurations
Logging warnings and overriding settings for non-configurable properties.

Permalink to checkIfUnexpectedUserSpecifiedConsumerConfig

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

ashmeet13 · 2022-12-14T07:38:14Z

Since I am fairly new to the code base - I will be spending some more time identifying any other configs that I might have missed that can fall under any of these buckets.
@ableegoldman could you please have a look at this PR.

ableegoldman

Thanks for the PR! One config in particular that is kind of a special case is the ProducerConfig default.partitioner

It's not that Streams overrides this (as is the case for other configs, like allow.auto.create.topics) but you just can't use it within Streams without breaking your app. We have a special StreamsPartitioner interface for Streams users to plug in rather than using this config.

I guess rather than adding special handling, we could just add it to the PRODUCER_DEFAULT_OVERRIDES similar to what you did here already, and just override it to null

ashmeet13 · 2022-12-28T14:45:48Z

Hi @ableegoldman thank you for the review!
I went through the producer configs and was not able to find any config with the name default.partitioner. Were you referring to the partitioner.class config?
In case this is not the config could you please help me by pointing where I can find default.partitioner within code?

ableegoldman · 2022-12-30T05:16:50Z

Ah, yeah, sorry -- it's partitioner.class

mjsax · 2023-02-14T00:57:10Z

What's the status of this PR?

github-actions · 2023-06-23T03:39:03Z

This PR is being marked as stale since it has not had any activity in 90 days. If you would like to keep this PR alive, please ask a committer for review. If the PR has merge conflicts, please update it with the latest from trunk (or appropriate release branch)

If this PR is no longer valid or desired, please feel free to close it. If no activity occurrs in the next 30 days, it will be automatically closed.

ashmeet13 · 2023-07-22T17:22:18Z

Will be re-picking this. My bad for dropping this in the middle.
Will update soon.
Issue - https://issues.apache.org/jira/browse/KAFKA-14405

mjsax · 2023-07-25T22:59:00Z

Thanks! Great to hear.

ashmeet13 · 2023-08-16T12:28:48Z

Hi @mjsax, a small doubt -
From what I understand there are three different types of consumer configs - global, main and restore

If a user has set the property consumer.allow.auto.create.topics to True we override it to False because stream controls this property value.
But this can be bypassed by setting the property as main.consumer.allow.auto.create.topics = True and this property would be set as True.

Is this expected behaviour?

mjsax · 2023-09-08T02:18:25Z

Sorry for late reply. I was OOO. Well, if it can be by-passed, it sounds like a bug.

Because there are multiple consumer, we use consumer. to allow users to change a config for all consumer. If you want to change a config for a specific consumer (or if you want to configure two consumer differently), you would use main.consumer. et al. If both prefix (generic and specific) are use the consumer-specific prefix has preference over the general consumer. prefix. Does this make sense? It's basically a "config hierarchy" (flexible and powerful, but maybe a little hard to understand on first encounter...)

But no matter what prefix is used, we should not allow users to by-pass what Streams tries to overwrite. (Btw: just setting allow.auto.create.topic = true, ie, without consumer. prefix is technically also valid, and we would pass config without prefix into the consumer, too, so this case must also be covered.)

ashmeet13 · 2023-10-08T18:48:35Z

Got it @mjsax -
Sharing the code that seems to be causing this bypass. Currently to fetch any consumer config i.e. main, restore or global we use a common function getCommonConsumerConfigs

It's within the getCommonConsumerConfigs function where we check and override the configs preferred by streams -

    private Map<String, Object> getCommonConsumerConfigs() {
        // Fetch all consumer props starting with "consumer."
        clientProvidedProps = getClientPropsWithPrefix(CONSUMER_PREFIX, ConsumerConfig.configNames());

        // CLean out any properties that were set but need to be controlled by streams
        checkIfUnexpectedUserSpecifiedConfig(clientProvidedProps, NON_CONFIGURABLE_CONSUMER_DEFAULT_CONFIGS);
        checkIfUnexpectedUserSpecifiedConfig(clientProvidedProps, NON_CONFIGURABLE_CONSUMER_EOS_CONFIGS);

        // Create a config map of the preferred props and merge it with the cleaned props from above  
        final consumerProps =new (eosEnabled ? CONSUMER_EOS_OVERRIDES : CONSUMER_DEFAULT_OVERRIDES);
        consumerProps.putAll(clientProvidedProps);
    }

And the logic within getMainConsumerConfigs is -

    public Map<String, Object> getMainConsumerConfigs(...) {
        // Fetch the props starting with "consumer." after cleaning
        // any props that needed to be overwritten
        final consumerProps = getCommonConsumerConfigs();

        // Get main consumer override props i.e. the ones 
        // starting with "main.consumer." and merge the two maps.
        final mainConsumerProps = originalsWithPrefix(MAIN_CONSUMER_PREFIX);
        for (final entry: mainConsumerProps.entrySet()) {
            consumerProps.put(entry.getKey(), entry.getValue());

        // Continue processing and filling in other required configs
        }

Do you think I've understood this piece correct?
If so should a fix go for this within this PR itself?

mjsax · 2023-10-13T02:38:52Z

Thanks for digging into this -- I think you are spot on -- seem we should extract a method that will set KS controlled config, and refactor getMainConsumerConfigs to first call getCommonConsumerConfigs(), than apply main.consumer configs, and in a last step call the new method to set KS controlled configs.

I assume we need to do something similar for restore and global consumer? -- To be fair, I was actually aware that something is off and still have a (old and stale) local branch adding corresponding testing to StreamsConfigTest to verify that overwrite hierarchy works as expected... Would be great if you could also look into this test...

ashmeet13 · 2023-10-13T20:07:58Z

Got it! I'll make this change - for now I have gone through the code and the following two references and compiled a list of configs that are somehow "controlled" by KS. For now sharing the Producer Configs here and soon Consumer Configs too

Producer Configs with EoS Disabled

1. [Editable] [CustomDefault] linger.ms = 100
2. [Fixed] partitioner.class = StreamsPartitioner

Producer Configs with EoS Disabled

1. [Editable] [CustomDefault] linger.ms = 100
2. [Fixed] partitioner.class = StreamsPartitioner
3. [Fixed] enable.idempotence = true
4. [Validate] max.in.flight.requests.per.connection <= 5
5. [Fixed] [NoDefault] transactional.id = <appId>-<generatedSuffix>
6. [Editable] [CustomDefault] delivery.timeout.ms = Integer.MAX
7. [Editable] [CustomDefault] transaction.timeout.ms = 10000

mjsax · 2023-11-29T04:30:48Z

@ashmeet13 -- Any update on this PR? We are coming up to 3.7 release code freeze deadline. Might be nice to finish this on time?

ashmeet13 · 2023-12-06T08:34:06Z

Hi @mjsax apologies for the delay. Pushing this soon.

mjsax · 2024-02-17T01:33:44Z

@ashmeet13 -- do you still have interest to finish this PR?

mjsax · 2024-03-01T23:33:13Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

@@ -1140,6 +1145,7 @@ public class StreamsConfig extends AbstractConfig {
    static {
        final Map<String, Object> tempProducerDefaultOverrides = new HashMap<>();
        tempProducerDefaultOverrides.put(ProducerConfig.LINGER_MS_CONFIG, "100");
+        tempProducerDefaultOverrides.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "none");


Default is already null -- why do we need to set it?

mjsax · 2024-03-01T23:36:30Z

streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java

@@ -1883,3 +1886,20 @@ public static void main(final String[] args) {
        System.out.println(CONFIG.toHtml(4, config -> "streamsconfigs_" + config));
    }
 }
+
+
+    public Map<String, Object> getMainConsumerConfigs(...) {


StreamsConfig is public API and we cannot just modify it w/o a KIP. -- Also, why do we need this new method to begin with? We already have getMainConsumerConfigs(...).

mjsax · 2024-03-01T23:38:14Z

streams/src/test/java/org/apache/kafka/streams/StreamsConfigTest.java

@@ -530,6 +530,14 @@ public void shouldSetInternalLeaveGroupOnCloseConfigToFalseInConsumer() {
        assertThat(consumerConfigs.get("internal.leave.group.on.close"), is(false));
    }

+    @Test
+    public void shouldResetToDefaultIfConsumerAllowAutoCreateTopicsIsOverridden() {


This should apply to all consumers, right? Should we extend the test accordingly?

Should we also capture the logs and verify that the WARN is printed (not sure if necessary)?

mjsax · 2024-04-30T03:26:00Z

@ashmeet13 -- any updates on this PR?

ashmeet13 added 2 commits December 12, 2022 21:38

Add ALLOW_AUTO_CREATE_TOPICS_CONFIG to not allowed to override configs

10fbe05

Remove debug log

42649c6

Set the value of allow.auto.create.topics when config is created

bc90755

ableegoldman reviewed Dec 21, 2022

View reviewed changes

mjsax added the streams label Dec 28, 2022

github-actions bot added the stale Stale PRs label Jun 23, 2023

github-actions bot removed the stale Stale PRs label Jul 23, 2023

Merge remote-tracking branch 'origin/trunk' into KAFKA-14405

6c0ca56

Add NON_CONFIGURABLE_PRODUCER_DEFAULT_CONFIGS

9d3567e

Merge remote-tracking branch 'origin/trunk' into KAFKA-14405

7f8bfda

Staging codE

21e01f9

mjsax reviewed Mar 1, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams #12988

KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams #12988

ashmeet13 commented Dec 14, 2022 •

edited

ashmeet13 commented Dec 14, 2022 •

edited

ableegoldman left a comment

ashmeet13 commented Dec 28, 2022

ableegoldman commented Dec 30, 2022

mjsax commented Feb 14, 2023

github-actions bot commented Jun 23, 2023

ashmeet13 commented Jul 22, 2023 •

edited

mjsax commented Jul 25, 2023

ashmeet13 commented Aug 16, 2023 •

edited

mjsax commented Sep 8, 2023 •

edited

ashmeet13 commented Oct 8, 2023

mjsax commented Oct 13, 2023

ashmeet13 commented Oct 13, 2023

mjsax commented Nov 29, 2023

ashmeet13 commented Dec 6, 2023

mjsax commented Feb 17, 2024

mjsax Mar 1, 2024

mjsax Mar 1, 2024

mjsax Mar 1, 2024

mjsax commented Apr 30, 2024

KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams #12988

Are you sure you want to change the base?

KAFKA-14405: Log a warning when users attempt to set a config controlled by Streams #12988

Conversation

ashmeet13 commented Dec 14, 2022 • edited

Committer Checklist (excluded from commit message)

ashmeet13 commented Dec 14, 2022 • edited

ableegoldman left a comment

Choose a reason for hiding this comment

ashmeet13 commented Dec 28, 2022

ableegoldman commented Dec 30, 2022

mjsax commented Feb 14, 2023

github-actions bot commented Jun 23, 2023

ashmeet13 commented Jul 22, 2023 • edited

mjsax commented Jul 25, 2023

ashmeet13 commented Aug 16, 2023 • edited

mjsax commented Sep 8, 2023 • edited

ashmeet13 commented Oct 8, 2023

mjsax commented Oct 13, 2023

ashmeet13 commented Oct 13, 2023

mjsax commented Nov 29, 2023

ashmeet13 commented Dec 6, 2023

mjsax commented Feb 17, 2024

mjsax Mar 1, 2024

Choose a reason for hiding this comment

mjsax Mar 1, 2024

Choose a reason for hiding this comment

mjsax Mar 1, 2024

Choose a reason for hiding this comment

mjsax commented Apr 30, 2024

ashmeet13 commented Dec 14, 2022 •

edited

ashmeet13 commented Dec 14, 2022 •

edited

ashmeet13 commented Jul 22, 2023 •

edited

ashmeet13 commented Aug 16, 2023 •

edited

mjsax commented Sep 8, 2023 •

edited