...

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 .

Proposed Changes

In KafkaStreams  constructor, we will read the value of streamsdefault.kafka.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

...