Status

Current state: Under DiscussionAccepted

Discussion thread: here

JIRA: here

...

We will add a public field in StreamsConfig  called streamsdefault.kafka.client.supplier to  to configure the KafkaClientSupplier  we use in KafkaStreams. The default value of the config will the classname

of DefaultKafkaClientSupplier .

We will also add a new public method in StreamsConfig  called getKafkaClientSupplier . It will return KafkaClientSupplier  based on the config.

Proposed Changes

In KafkaStreams  constructor, we will read the value of streams.kafkadefault.client.supplier from StreamsConfig  and if it's not null  and not empty, we . We will use reflection to create an object of given class and throw

an exception if the class type isn't  KafkaClientSupplier . If user provides both the config and supply KafkaClientSupplier  in KafkaStreams  constructor, the config will be ignored and supplied

KafkaClientSupplier  will be used.

Compatibility, Deprecation, and Migration Plan

...