GH-360: Improve Binder Producer/Consumer Config

Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/360

`producer-configuration` and `consumer-configuration` improperly appear in content-assist.

These are methods used by the binders to get merged configuration data (boot and binder).

Rename the methods and add `producerProperties` and `consumerProperties` to allow
configuration.

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaBinderConfigurationProperties.java

@@ -24,10 +24,10 @@ import java.util.Map;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.ProducerConfig;
 
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.DeprecatedConfigurationProperty;
+import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
@@ -46,13 +46,25 @@ public class KafkaBinderConfigurationProperties {
 
 	private final Transaction transaction = new Transaction();
 
-	@Autowired
-	private KafkaProperties kafkaProperties;
+	private final KafkaProperties kafkaProperties;
 
 	private String[] zkNodes = new String[] { "localhost" };
 
+	/**
+	 * Arbitrary kafka properties that apply to both producers and consumers.
+	 */
 	private Map<String, String> configuration = new HashMap<>();
 
+	/**
+	 * Arbitrary kafka consumer properties.
+	 */
+	private Map<String, String> consumerProperties = new HashMap<>();
+
+	/**
+	 * Arbitrary kafka producer properties.
+	 */
+	private Map<String, String> producerProperties = new HashMap<>();
+
 	private String defaultZkPort = "2181";
 
 	private String[] brokers = new String[] { "localhost" };
@@ -107,6 +119,12 @@ public class KafkaBinderConfigurationProperties {
 	 */
 	private String headerMapperBeanName;
 
+
+	public KafkaBinderConfigurationProperties(KafkaProperties kafkaProperties) {
+		Assert.notNull(kafkaProperties, "'kafkaProperties' cannot be null");
+		this.kafkaProperties = kafkaProperties;
+	}
+
 	public Transaction getTransaction() {
 		return this.transaction;
 	}
@@ -461,18 +479,40 @@ public class KafkaBinderConfigurationProperties {
 		this.configuration = configuration;
 	}
 
-	public Map<String, Object> getConsumerConfiguration() {
+	public Map<String, String> getConsumerProperties() {
+		return this.consumerProperties;
+	}
+
+	public void setConsumerProperties(Map<String, String> consumerProperties) {
+		Assert.notNull(consumerProperties, "'consumerProperties' cannot be null");
+		this.consumerProperties = consumerProperties;
+	}
+
+	public Map<String, String> getProducerProperties() {
+		return this.producerProperties;
+	}
+
+	public void setProducerProperties(Map<String, String> producerProperties) {
+		Assert.notNull(producerProperties, "'producerProperties' cannot be null");
+		this.producerProperties = producerProperties;
+	}
+
+	/**
+	 * Merge boot consumer properties, general properties from
+	 * {@link #setConfiguration(Map)} that apply to consumers, properties from
+	 * {@link #setConsumerProperties(Map)}, in that order.
+	 * @return the merged properties.
+	 */
+	public Map<String, Object> mergedConsumerConfiguration() {
 		Map<String, Object> consumerConfiguration = new HashMap<>();
-		// If Spring Boot Kafka properties are present, add them with lowest precedence
-		if (this.kafkaProperties != null) {
-			consumerConfiguration.putAll(this.kafkaProperties.buildConsumerProperties());
-		}
-		// Copy configured binder properties
+		consumerConfiguration.putAll(this.kafkaProperties.buildConsumerProperties());
+		// Copy configured binder properties that apply to consumers
 		for (Map.Entry<String, String> configurationEntry : this.configuration.entrySet()) {
 			if (ConsumerConfig.configNames().contains(configurationEntry.getKey())) {
 				consumerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue());
 			}
 		}
+		consumerConfiguration.putAll(this.consumerProperties);
 		// Override Spring Boot bootstrap server setting if left to default with the value
 		// configured in the binder
 		if (ObjectUtils.isEmpty(consumerConfiguration.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
@@ -492,18 +532,22 @@ public class KafkaBinderConfigurationProperties {
 		return Collections.unmodifiableMap(consumerConfiguration);
 	}
 
-	public Map<String, Object> getProducerConfiguration() {
+	/**
+	 * Merge boot producer properties, general properties from
+	 * {@link #setConfiguration(Map)} that apply to producers, properties from
+	 * {@link #setProducerProperties(Map)}, in that order.
+	 * @return the merged properties.
+	 */
+	public Map<String, Object> mergedProducerConfiguration() {
 		Map<String, Object> producerConfiguration = new HashMap<>();
-		// If Spring Boot Kafka properties are present, add them with lowest precedence
-		if (this.kafkaProperties != null) {
-			producerConfiguration.putAll(this.kafkaProperties.buildProducerProperties());
-		}
-		// Copy configured binder properties
+		producerConfiguration.putAll(this.kafkaProperties.buildProducerProperties());
+		// Copy configured binder properties that apply to producers
 		for (Map.Entry<String, String> configurationEntry : configuration.entrySet()) {
 			if (ProducerConfig.configNames().contains(configurationEntry.getKey())) {
 				producerConfiguration.put(configurationEntry.getKey(), configurationEntry.getValue());
 			}
 		}
+		producerConfiguration.putAll(this.producerProperties);
 		// Override Spring Boot bootstrap server setting if left to default with the value
 		// configured in the binder
 		if (ObjectUtils.isEmpty(producerConfiguration.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) {

spring-cloud-stream-binder-kafka-core/src/test/java/org/springframework/cloud/stream/binder/kafka/provisioning/KafkaTopicProvisionerTests.java

@@ -49,7 +49,7 @@ public class KafkaTopicProvisionerTests {
 		KafkaProperties bootConfig = new KafkaProperties();
 		bootConfig.getProperties().put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
 		bootConfig.setBootstrapServers(Collections.singletonList("localhost:1234"));
-		KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties();
+		KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(bootConfig);
 		binderConfig.getConfiguration().put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SSL");
 		ClassPathResource ts = new ClassPathResource("test.truststore.ks");
 		binderConfig.getConfiguration().put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, ts.getFile().getAbsolutePath());
@@ -68,7 +68,7 @@ public class KafkaTopicProvisionerTests {
 		KafkaProperties bootConfig = new KafkaProperties();
 		bootConfig.getProperties().put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT");
 		bootConfig.setBootstrapServers(Collections.singletonList("localhost:9092"));
-		KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties();
+		KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(bootConfig);
 		binderConfig.getConfiguration().put(AdminClientConfig.SECURITY_PROTOCOL_CONFIG, "SSL");
 		ClassPathResource ts = new ClassPathResource("test.truststore.ks");
 		binderConfig.getConfiguration().put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, ts.getFile().getAbsolutePath());
@@ -84,7 +84,7 @@ public class KafkaTopicProvisionerTests {
 	@Test
 	public void brokersInvalid() throws Exception {
 		KafkaProperties bootConfig = new KafkaProperties();
-		KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties();
+		KafkaBinderConfigurationProperties binderConfig = new KafkaBinderConfigurationProperties(bootConfig);
 		binderConfig.getConfiguration().put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:1234");
 		try {
 			new KafkaTopicProvisioner(binderConfig, bootConfig);

spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc

@@ -63,6 +63,12 @@ Default: `9092`.
 spring.cloud.stream.kafka.binder.configuration::
 Key/Value map of client properties (both producers and consumer) passed to all clients created by the binder.
 Due to the fact that these properties are used by both producers and consumers, usage should be restricted to common properties -- for example, security settings.
+Properties here supersede any properties set in boot.
++
+Default: Empty map.
+spring.cloud.stream.kafka.binder.consumerProperties::
+Key/Value map of arbitrary Kafka client consumer properties.
+Properties here supersede any properties set in boot and in the `configuration` property above.
 +
 Default: Empty map.
 spring.cloud.stream.kafka.binder.headers::
@@ -86,6 +92,11 @@ The global minimum number of partitions that the binder configures on topics on
 It can be superseded by the `partitionCount` setting of the producer or by the value of `instanceCount * concurrency` settings of the producer (if either is larger).
 +
 Default: `1`.
+spring.cloud.stream.kafka.binder.producerProperties::
+Key/Value map of arbitrary Kafka client producer properties.
+Properties here supersede any properties set in boot and in the `configuration` property above.
++
+Default: Empty map.
 spring.cloud.stream.kafka.binder.replicationFactor::
 The replication factor of auto-created topics if `autoCreateTopics` is active.
 Can be overridden on each binding.

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java

@@ -27,6 +27,7 @@ import org.apache.kafka.streams.errors.LogAndFailExceptionHandler;
 
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsBinderConfigurationProperties;
@@ -41,6 +42,7 @@ import org.springframework.util.ObjectUtils;
 /**
  * @author Marius Bogoevici
  * @author Soby Chacko
+ * @author Gary Russell
  */
 @EnableConfigurationProperties(KafkaStreamsExtendedBindingProperties.class)
 @ConditionalOnBean(BindingService.class)
@@ -48,8 +50,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration {
 
 	@Bean
 	@ConfigurationProperties(prefix = "spring.cloud.stream.kafka.streams.binder")
-	public KafkaStreamsBinderConfigurationProperties binderConfigurationProperties() {
-		return new KafkaStreamsBinderConfigurationProperties();
+	public KafkaStreamsBinderConfigurationProperties binderConfigurationProperties(KafkaProperties kafkaProperties) {
+		return new KafkaStreamsBinderConfigurationProperties(kafkaProperties);
 	}
 
 	@Bean("streamConfigGlobalProperties")

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsDlqDispatch.java

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.kafka.streams;
 /**
  * @author Soby Chacko
  * @author Rafal Zukowski
+ * @author Gary Russell
  */
 import java.util.HashMap;
 import java.util.Map;
@@ -105,8 +106,9 @@ class KafkaStreamsDlqDispatch {
 		props.put(ProducerConfig.RETRIES_CONFIG, 0);
 		props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
 		props.put(ProducerConfig.ACKS_CONFIG, configurationProperties.getRequiredAcks());
-		if (!ObjectUtils.isEmpty(configurationProperties.getProducerConfiguration())) {
-			props.putAll(configurationProperties.getProducerConfiguration());
+		Map<String, Object> mergedConfig = configurationProperties.mergedProducerConfiguration();
+		if (!ObjectUtils.isEmpty(mergedConfig)) {
+			props.putAll(mergedConfig);
 		}
 		if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
 			props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString());

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/properties/KafkaStreamsBinderConfigurationProperties.java

@@ -16,13 +16,19 @@
 
 package org.springframework.cloud.stream.binder.kafka.streams.properties;
 
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 
 /**
  * @author Soby Chacko
+ * @author Gary Russell
  */
 public class KafkaStreamsBinderConfigurationProperties extends KafkaBinderConfigurationProperties {
 
+	public KafkaStreamsBinderConfigurationProperties(KafkaProperties kafkaProperties) {
+		super(kafkaProperties);
+	}
+
 	public enum SerdeError {
 		logAndContinue,
 		logAndFail,

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderMetrics.java

@@ -134,8 +134,9 @@ public class KafkaBinderMetrics implements MeterBinder, ApplicationListener<Bind
 			Map<String, Object> props = new HashMap<>();
 			props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
 			props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
-			if (!ObjectUtils.isEmpty(binderConfigurationProperties.getConsumerConfiguration())) {
-				props.putAll(binderConfigurationProperties.getConsumerConfiguration());
+			Map<String, Object> mergedConfig = this.binderConfigurationProperties.mergedConsumerConfiguration();
+			if (!ObjectUtils.isEmpty(mergedConfig)) {
+				props.putAll(mergedConfig);
 			}
 			if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
 				props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -286,8 +286,9 @@ public class KafkaMessageChannelBinder extends
 		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
 		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
 		props.put(ProducerConfig.ACKS_CONFIG, String.valueOf(this.configurationProperties.getRequiredAcks()));
-		if (!ObjectUtils.isEmpty(configurationProperties.getProducerConfiguration())) {
-			props.putAll(configurationProperties.getProducerConfiguration());
+		Map<String, Object> mergedConfig = this.configurationProperties.mergedProducerConfiguration();
+		if (!ObjectUtils.isEmpty(mergedConfig)) {
+			props.putAll(mergedConfig);
 		}
 		if (ObjectUtils.isEmpty(props.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
 			props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());
@@ -723,8 +724,9 @@ public class KafkaMessageChannelBinder extends
 		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, anonymous ? "latest" : "earliest");
 		props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
 
-		if (!ObjectUtils.isEmpty(configurationProperties.getConsumerConfiguration())) {
-			props.putAll(configurationProperties.getConsumerConfiguration());
+		Map<String, Object> mergedConfig = configurationProperties.mergedConsumerConfiguration();
+		if (!ObjectUtils.isEmpty(mergedConfig)) {
+			props.putAll(mergedConfig);
 		}
 		if (ObjectUtils.isEmpty(props.get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG))) {
 			props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString());

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java

@@ -70,8 +70,8 @@ public class KafkaBinderConfiguration {
 	private KafkaProperties kafkaProperties;
 
 	@Bean
-	KafkaBinderConfigurationProperties configurationProperties() {
-		return new KafkaBinderConfigurationProperties();
+	KafkaBinderConfigurationProperties configurationProperties(KafkaProperties kafkaProperties) {
+		return new KafkaBinderConfigurationProperties(kafkaProperties);
 	}
 
 	@Bean

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderHealthIndicatorConfiguration.java

@@ -33,7 +33,7 @@ import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.util.ObjectUtils;
 
 /**
- * 
+ *
  * @author Oleg Zhurakousky
  *
  */
@@ -48,8 +48,9 @@ class KafkaBinderHealthIndicatorConfiguration {
 		Map<String, Object> props = new HashMap<>();
 		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
 		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
-		if (!ObjectUtils.isEmpty(configurationProperties.getConsumerConfiguration())) {
-			props.putAll(configurationProperties.getConsumerConfiguration());
+		Map<String, Object> mergedConfig = configurationProperties.mergedConsumerConfiguration();
+		if (!ObjectUtils.isEmpty(mergedConfig)) {
+			props.putAll(mergedConfig);
 		}
 		if (!props.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)) {
 			props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, configurationProperties.getKafkaConnectionString());

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -63,7 +63,6 @@ import org.junit.Test;
 import org.junit.rules.ExpectedException;
 
 import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
 import org.springframework.cloud.stream.binder.Binder;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.Binding;
@@ -182,7 +181,7 @@ public class KafkaBinderTests extends
 	protected KafkaTestBinder getBinder() {
 		if (binder == null) {
 			KafkaBinderConfigurationProperties binderConfiguration = createConfigurationProperties();
-			KafkaTopicProvisioner kafkaTopicProvisioner = new KafkaTopicProvisioner(binderConfiguration, new KafkaProperties());
+			KafkaTopicProvisioner kafkaTopicProvisioner = new KafkaTopicProvisioner(binderConfiguration, new TestKafkaProperties());
 			try {
 				kafkaTopicProvisioner.afterPropertiesSet();
 			}
@@ -196,7 +195,7 @@ public class KafkaBinderTests extends
 
 	private Binder getBinder(KafkaBinderConfigurationProperties kafkaBinderConfigurationProperties) {
 		KafkaTopicProvisioner provisioningProvider =
-				new KafkaTopicProvisioner(kafkaBinderConfigurationProperties, new KafkaProperties());
+				new KafkaTopicProvisioner(kafkaBinderConfigurationProperties, new TestKafkaProperties());
 		try {
 			provisioningProvider.afterPropertiesSet();
 		}
@@ -207,7 +206,8 @@ public class KafkaBinderTests extends
 	}
 
 	private KafkaBinderConfigurationProperties createConfigurationProperties() {
-		KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties();
+		KafkaBinderConfigurationProperties binderConfiguration = new KafkaBinderConfigurationProperties(
+				new TestKafkaProperties());
 		BrokerAddress[] brokerAddresses = embeddedKafka.getBrokerAddresses();
 		List<String> bAddresses = new ArrayList<>();
 		for (BrokerAddress bAddress : brokerAddresses) {
@@ -215,7 +215,6 @@ public class KafkaBinderTests extends
 		}
 		String[] foo = new String[bAddresses.size()];
 		binderConfiguration.setBrokers(bAddresses.toArray(foo));
-		binderConfiguration.setZkNodes(embeddedKafka.getZookeeperConnectionString());
 		return binderConfiguration;
 	}
 

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderUnitTests.java

@@ -30,6 +30,7 @@ import java.util.stream.Collectors;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.PartitionInfo;
 import org.apache.kafka.common.TopicPartition;
 import org.junit.Test;
@@ -67,8 +68,10 @@ public class KafkaBinderUnitTests {
 
 	@Test
 	public void testPropertyOverrides() throws Exception {
-		KafkaBinderConfigurationProperties binderConfigurationProperties = new KafkaBinderConfigurationProperties();
-		KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(binderConfigurationProperties, new KafkaProperties());
+		KafkaProperties kafkaProperties = new TestKafkaProperties();
+		KafkaBinderConfigurationProperties binderConfigurationProperties =
+				new KafkaBinderConfigurationProperties(kafkaProperties);
+		KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(binderConfigurationProperties, kafkaProperties);
 		KafkaMessageChannelBinder binder = new KafkaMessageChannelBinder(binderConfigurationProperties,
 				provisioningProvider);
 		KafkaConsumerProperties consumerProps = new KafkaConsumerProperties();
@@ -102,6 +105,30 @@ public class KafkaBinderUnitTests {
 		assertThat(configs.get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("earliest");
 	}
 
+	@Test
+	public void testMergedConsumerProperties() {
+		KafkaProperties bootProps = new TestKafkaProperties();
+		bootProps.getConsumer().getProperties().put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "bar");
+		KafkaBinderConfigurationProperties props = new KafkaBinderConfigurationProperties(bootProps);
+		assertThat(props.mergedConsumerConfiguration().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("bar");
+		props.getConfiguration().put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "baz");
+		assertThat(props.mergedConsumerConfiguration().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("baz");
+		props.getConsumerProperties().put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "qux");
+		assertThat(props.mergedConsumerConfiguration().get(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG)).isEqualTo("qux");
+	}
+
+	@Test
+	public void testMergedProducerProperties() {
+		KafkaProperties bootProps = new TestKafkaProperties();
+		bootProps.getProducer().getProperties().put(ProducerConfig.RETRIES_CONFIG, "bar");
+		KafkaBinderConfigurationProperties props = new KafkaBinderConfigurationProperties(bootProps);
+		assertThat(props.mergedProducerConfiguration().get(ProducerConfig.RETRIES_CONFIG)).isEqualTo("bar");
+		props.getConfiguration().put(ProducerConfig.RETRIES_CONFIG, "baz");
+		assertThat(props.mergedProducerConfiguration().get(ProducerConfig.RETRIES_CONFIG)).isEqualTo("baz");
+		props.getProducerProperties().put(ProducerConfig.RETRIES_CONFIG, "qux");
+		assertThat(props.mergedProducerConfiguration().get(ProducerConfig.RETRIES_CONFIG)).isEqualTo("qux");
+	}
+
 	@Test
 	public void testOffsetResetWithGroupManagementEarliest() throws Exception {
 		testOffsetResetWithGroupManagement(true, true);
@@ -126,7 +153,8 @@ public class KafkaBinderUnitTests {
 		final List<TopicPartition> partitions = new ArrayList<>();
 		partitions.add(new TopicPartition("foo", 0));
 		partitions.add(new TopicPartition("foo", 1));
-		KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties();
+		KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties(
+				new TestKafkaProperties());
 		KafkaTopicProvisioner provisioningProvider = mock(KafkaTopicProvisioner.class);
 		ConsumerDestination dest = mock(ConsumerDestination.class);
 		given(dest.getName()).willReturn("foo");

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaTransactionTests.java

@@ -58,9 +58,10 @@ public class KafkaTransactionTests {
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	public void testProducerRunsInTx() {
-		KafkaProperties kafkaProperties = new KafkaProperties();
+		KafkaProperties kafkaProperties = new TestKafkaProperties();
 		kafkaProperties.setBootstrapServers(Collections.singletonList(embeddedKafka.getBrokersAsString()));
-		KafkaBinderConfigurationProperties configurationProperties = new KafkaBinderConfigurationProperties();
+		KafkaBinderConfigurationProperties configurationProperties =
+				new KafkaBinderConfigurationProperties(kafkaProperties);
 		configurationProperties.getTransaction().setTransactionIdPrefix("foo-");
 		KafkaTopicProvisioner provisioningProvider = new KafkaTopicProvisioner(configurationProperties, kafkaProperties);
 		provisioningProvider.setMetadataRetryOperations(new RetryTemplate());

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/TestKafkaProperties.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.binder.kafka;
+
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+
+import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
+
+/**
+ * Test {@link KafkaProperties} initialized in the same way as the
+ * {@code KafkaBinderEnvironmentPostProcessor} initializes the properties.
+ *
+ * @author Gary Russell
+ * @since 2.1
+ *
+ */
+public class TestKafkaProperties extends KafkaProperties {
+
+	public TestKafkaProperties() {
+		getConsumer().setKeyDeserializer(ByteArrayDeserializer.class);
+		getConsumer().setValueDeserializer(ByteArrayDeserializer.class);
+		getProducer().setKeySerializer(ByteArraySerializer.class);
+		getProducer().setValueSerializer(ByteArraySerializer.class);
+	}
+
+}