...

We might also consider to extend `KafkaClientSupplier` and add a `getGlobalConsumer()` method.

Public Interfaces

The changes affected classes including `StreamsConfig` and `KafkaClientSupplier`

// Current KafkaClientSupplier consumer APIs: 
Consumer<byte[], byte[]> getConsumer(final Map<String, Object> config);
Consumer<byte[], byte[]> getRestoreConsumer(final Map<String, Object> config);

Proposed Changes

Describe the new thing you want to do in appropriate detail. This may be fairly extensive and have large subsections of its own. Or it may be a few sentences. Use judgement based on the scope of the change.We shall add a new public API in the KafkaClientSupplier class to get global consumer during init. 

// New API for global consumer:
Consumer<byte[], byte[]> getGlobalConsumer(final Map<String, Object> config); 

...