@Test public void testMapsConsumers() { ArrayList<String> topicList = new ArrayList<>(); topicList.add( "topic" ); inputMeta.setTopics( topicList ); inputMeta.setConsumerGroup( "cg" ); inputMeta.setKeyField( new KafkaConsumerField( KafkaConsumerField.Name.KEY, "key" ) ); inputMeta.setMessageField( new KafkaConsumerField( KafkaConsumerField.Name.MESSAGE, "msg" ) ); inputMeta.setNamedClusterServiceLocator( namedClusterServiceLocator ); inputMeta.setAutoCommit( false ); Map<String, String> advancedConfig = new LinkedHashMap<>(); advancedConfig.put( "advanced.config1", "advancedPropertyValue1" ); advancedConfig.put( "advanced.config2", "advancedPropertyValue2" ); inputMeta.setConfig( advancedConfig ); when( jaasConfigService.isKerberos() ).thenReturn( false ); new KafkaFactory( consumerFun, producerFun ).consumer( inputMeta, Function.identity() ); Map<String, Object> expectedMap = new HashMap<>(); expectedMap.put( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "server:1234" ); expectedMap.put( ConsumerConfig.GROUP_ID_CONFIG, "cg" ); expectedMap.put( ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class ); expectedMap.put( ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class ); expectedMap.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false ); expectedMap.put( "advanced.config1", "advancedPropertyValue1" ); expectedMap.put( "advanced.config2", "advancedPropertyValue2" ); Mockito.verify( consumerFun ).apply( expectedMap ); }
public Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction ) { return consumer( meta, variablesFunction, KafkaConsumerField.Type.String, KafkaConsumerField.Type.String ); }
KafkaFactory { public Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction ) { return consumer( meta, variablesFunction, KafkaConsumerField.Type.String, KafkaConsumerField.Type.String ); } }
KafkaFactory { public Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction ) { return consumer( meta, variablesFunction, KafkaConsumerField.Type.String, KafkaConsumerField.Type.String ); } KafkaFactory(
    Function<Map<String, Object>, Consumer> consumerFunction,
    Function<Map<String, Object>, Producer<Object, Object>> producerFunction ); }
KafkaFactory { public Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction ) { return consumer( meta, variablesFunction, KafkaConsumerField.Type.String, KafkaConsumerField.Type.String ); } KafkaFactory(
    Function<Map<String, Object>, Consumer> consumerFunction,
    Function<Map<String, Object>, Producer<Object, Object>> producerFunction ); static KafkaFactory defaultFactory(); Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction ); Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction,
    KafkaConsumerField.Type keyDeserializerType, KafkaConsumerField.Type msgDeserializerType ); void putKerberosConfig( Map<String, Object> kafkaConfig, JaasConfigService jaasConfigService ); Producer<Object, Object> producer(
    KafkaProducerOutputMeta meta, Function<String, String> variablesFunction ); Producer<Object, Object> producer(
    KafkaProducerOutputMeta meta, Function<String, String> variablesFunction,
    KafkaConsumerField.Type keySerializerType, KafkaConsumerField.Type msgSerializerType ); }
KafkaFactory { public Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction ) { return consumer( meta, variablesFunction, KafkaConsumerField.Type.String, KafkaConsumerField.Type.String ); } KafkaFactory(
    Function<Map<String, Object>, Consumer> consumerFunction,
    Function<Map<String, Object>, Producer<Object, Object>> producerFunction ); static KafkaFactory defaultFactory(); Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction ); Consumer consumer( KafkaConsumerInputMeta meta, Function<String, String> variablesFunction,
    KafkaConsumerField.Type keyDeserializerType, KafkaConsumerField.Type msgDeserializerType ); void putKerberosConfig( Map<String, Object> kafkaConfig, JaasConfigService jaasConfigService ); Producer<Object, Object> producer(
    KafkaProducerOutputMeta meta, Function<String, String> variablesFunction ); Producer<Object, Object> producer(
    KafkaProducerOutputMeta meta, Function<String, String> variablesFunction,
    KafkaConsumerField.Type keySerializerType, KafkaConsumerField.Type msgSerializerType );  }