@Test public void testMapsConsumersWithVariables() { inputMeta.setConsumerGroup( "${consumerGroup}" ); inputMeta.setNamedClusterServiceLocator( namedClusterServiceLocator ); Map<String, String> advancedConfig = new LinkedHashMap<>(); advancedConfig.put( "advanced.variable", "${advanced.var}" ); inputMeta.setConfig( advancedConfig ); when( jaasConfigService.isKerberos() ).thenReturn( false ); Variables variables = new Variables(); variables.setVariable( "server", "server:1234" ); variables.setVariable( "consumerGroup", "cg" ); variables.setVariable( "advanced.var", "advancedVarValue" ); new KafkaFactory( consumerFun, producerFun ).consumer( inputMeta, variables::environmentSubstitute ); 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( "advanced.variable", "advancedVarValue" ); expectedMap.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true ); 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 );  }