SimpleKafkaProducerFactory implements KafkaProducerFactory<byte[], byte[]> { static Properties addProducerDefaultProperties(Properties properties) { properties.putIfAbsent(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, DEFAULT_KEY_SERIALIZER); properties.putIfAbsent(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DEFAULT_VAL_SERIALIZER); return properties; }  @Override Producer<byte[], byte[]> createProducer(Properties transportProps);  }
@Test public void testDefaultProducerConfig() { Properties properties = new Properties(); properties.put("bootstrap.servers", "MyBroker:10251"); properties.put("security.protocol", "SSL"); Properties actual = SimpleKafkaProducerFactory.addProducerDefaultProperties(properties); Properties expected = new Properties(); expected.putAll(properties); expected.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, DEFAULT_KEY_SERIALIZER); expected.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DEFAULT_VAL_SERIALIZER); Assert.assertEquals(actual, expected); }
@Test public void testOverrideDefaultProducerConfigs() { Properties properties = new Properties(); properties.put("bootstrap.servers", "MyBroker:10251"); properties.put("security.protocol", "SSL"); Properties additionalProperties = new Properties(); additionalProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "my.custom.serializer"); additionalProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "my.custom.serializer"); additionalProperties.putAll(properties); Properties actual = SimpleKafkaProducerFactory.addProducerDefaultProperties(additionalProperties); Properties expected = new Properties(); expected.putAll(properties); expected.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "my.custom.serializer"); expected.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "my.custom.serializer"); Assert.assertEquals(actual, expected); }