KafkaFactory { public Producer<Object, Object> producer( KafkaProducerOutputMeta meta, Function<String, String> variablesFunction ) { return producer( 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 );  }
@Test public void testMapsProducers() { outputMeta.setTopic( "topic" ); outputMeta.setClientId( "client" ); outputMeta.setKeyField( "key" ); outputMeta.setMessageField( "msg" ); outputMeta.setNamedClusterServiceLocator( namedClusterServiceLocator ); Map<String, String> advancedConfig = new LinkedHashMap<>(); advancedConfig.put( "advanced.config1", "advancedPropertyValue1" ); advancedConfig.put( "advanced.config2", "advancedPropertyValue2" ); outputMeta.setConfig( advancedConfig ); when( jaasConfigService.isKerberos() ).thenReturn( false ); new KafkaFactory( consumerFun, producerFun ).producer( outputMeta, Function.identity() ); Map<String, Object> expectedMap = new HashMap<>(); expectedMap.put( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "server:1234" ); expectedMap.put( ProducerConfig.CLIENT_ID_CONFIG, "client" ); expectedMap.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class ); expectedMap.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class ); expectedMap.put( "advanced.config1", "advancedPropertyValue1" ); expectedMap.put( "advanced.config2", "advancedPropertyValue2" ); Mockito.verify( producerFun ).apply( expectedMap ); }
@Test public void testMapsProducersWithSerializer() { outputMeta.setTopic( "topic" ); outputMeta.setClientId( "client" ); outputMeta.setKeyField( "key" ); outputMeta.setMessageField( "msg" ); outputMeta.setNamedClusterServiceLocator( namedClusterServiceLocator ); when( jaasConfigService.isKerberos() ).thenReturn( false ); new KafkaFactory( consumerFun, producerFun ).producer( outputMeta, Function.identity(), KafkaConsumerField.Type.Integer, KafkaConsumerField.Type.Number ); Map<String, Object> expectedMap = new HashMap<>(); expectedMap.put( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "server:1234" ); expectedMap.put( ProducerConfig.CLIENT_ID_CONFIG, "client" ); expectedMap.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DoubleSerializer.class ); expectedMap.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class ); Mockito.verify( producerFun ).apply( expectedMap ); }
@Test public void testMapsProducersWithVariables() { outputMeta.setClientId( "${client}" ); outputMeta.setNamedClusterServiceLocator( namedClusterServiceLocator ); Map<String, String> advancedConfig = new LinkedHashMap<>(); advancedConfig.put( "advanced.variable", "${advanced.var}" ); outputMeta.setConfig( advancedConfig ); when( jaasConfigService.isKerberos() ).thenReturn( false ); Variables variables = new Variables(); variables.setVariable( "server", "server:1234" ); variables.setVariable( "client", "myclient" ); variables.setVariable( "advanced.var", "advancedVarValue" ); new KafkaFactory( consumerFun, producerFun ).producer( outputMeta, variables::environmentSubstitute ); Map<String, Object> expectedMap = new HashMap<>(); expectedMap.put( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "server:1234" ); expectedMap.put( ProducerConfig.CLIENT_ID_CONFIG, "myclient" ); expectedMap.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class ); expectedMap.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class ); expectedMap.put( "advanced.variable", "advancedVarValue" ); Mockito.verify( producerFun ).apply( expectedMap ); }
@Test public void testNullMetaPropertiesResultInEmptyString() { outputMeta.setClusterName( null ); outputMeta.setNamedClusterServiceLocator( namedClusterServiceLocator ); when( jaasConfigService.isKerberos() ).thenReturn( false ); Variables variables = new Variables(); new KafkaFactory( consumerFun, producerFun ).producer( outputMeta, variables::environmentSubstitute ); Map<String, Object> expectedMap = new HashMap<>(); expectedMap.put( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "" ); expectedMap.put( ProducerConfig.CLIENT_ID_CONFIG, "" ); expectedMap.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class ); expectedMap.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class ); Mockito.verify( producerFun ).apply( expectedMap ); }