public <K, V> KafkaConsumer<K, V> createConsumer(Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer,
                                                     Properties overrideConfig) {
        Properties conf = consumerConfig();
        if (overrideConfig != null) {
            conf.putAll(overrideConfig);
        }
        keyDeserializer.configure(Maps.fromProperties(conf), true);
        valueDeserializer.configure(Maps.fromProperties(conf), false);
        return new KafkaConsumer<>(conf, keyDeserializer, valueDeserializer);
    }
public < K , V > KafkaConsumer < K , V > createConsumer ( Deserializer < K > keyDeserializer , Deserializer < V > valueDeserializer , Properties overrideConfig ) { Properties conf = consumerConfig ( ) ; if ( overrideConfig != null ) { conf . putAll ( overrideConfig ) ; } keyDeserializer . configure ( Maps . fromProperties ( conf ) , true ) ; valueDeserializer . configure ( Maps . fromProperties ( conf ) , false ) ; return new KafkaConsumer < > ( conf , keyDeserializer , valueDeserializer ) ; }
Create a consumer that can read from this broker

@param keyDeserializer   Key deserializer
@param valueDeserializer Value deserializer
@param overrideConfig    Consumer config to override. Pass null if there aren't any
@param <K>               Type of Key
@param <V>               Type of Value
@return KafkaConsumer