GH-2298: Refactor Common Binder Code to Core

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2298

Polishing per review comments; javadocs.

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ConsumerConfigCustomizer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2020 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.kafka.config;
+package org.springframework.cloud.stream.binder.kafka.support;
 
 import java.util.Map;
 
@@ -29,4 +29,5 @@ import java.util.Map;
 public interface ConsumerConfigCustomizer {
 
 	void configure(Map<String, Object> consumerProperties, String bindingName, String destination);
+
 }

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/support/ProducerConfigCustomizer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2020 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.cloud.stream.binder.kafka.config;
+package org.springframework.cloud.stream.binder.kafka.support;
 
 import java.util.Map;
 
@@ -29,4 +29,5 @@ import java.util.Map;
 public interface ProducerConfigCustomizer {
 
 	void configure(Map<String, Object> producerProperties, String bindingName, String destination);
+
 }

binders/kafka-binder/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/utils/BindingUtils.java

@@ -0,0 +1,230 @@
+/*
+ * Copyright 2022-2022 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
+ *
+ *      https://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.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
+import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
+import org.springframework.kafka.support.KafkaHeaderMapper;
+import org.springframework.kafka.support.converter.MessageConverter;
+import org.springframework.kafka.support.converter.MessagingMessageConverter;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Binding Utilities.
+ *
+ * @author Gary Russell
+ * @since 4.0
+ *
+ */
+public final class BindingUtils {
+
+	private BindingUtils() {
+	}
+
+	/**
+	 * Get the message converter for consumer bindings from the application context. If
+	 * the binding properties do not contain a bean name, a default
+	 * {@link MessagingMessageConverter} is returned; if the binder properties contain a
+	 * header mapper bean name, it is used in the default converter, otherwise a
+	 * {@link DefaultKafkaHeaderMapper} is used.
+	 * @param applicationContext the application context.
+	 * @param extendedConsumerProperties the consumer binding properties.
+	 * @param configurationProperties the binder properties.
+	 * @return the converter
+	 * @throws IllegalStateException if a bean name is specified but not found.
+	 */
+	public static MessageConverter getConsumerMessageConverter(ApplicationContext applicationContext,
+			ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties,
+			KafkaBinderConfigurationProperties configurationProperties) {
+
+		MessageConverter messageConverter;
+		if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) {
+			MessagingMessageConverter mmc = new MessagingMessageConverter();
+			StandardHeaders standardHeaders = extendedConsumerProperties.getExtension()
+					.getStandardHeaders();
+			mmc.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders)
+							|| StandardHeaders.both.equals(standardHeaders));
+			mmc.setGenerateTimestamp(
+					StandardHeaders.timestamp.equals(standardHeaders)
+							|| StandardHeaders.both.equals(standardHeaders));
+			KafkaHeaderMapper headerMapper = getHeaderMapper(applicationContext, configurationProperties);
+			if (headerMapper == null) {
+				headerMapper = new DefaultKafkaHeaderMapper();
+			}
+			mmc.setHeaderMapper(headerMapper);
+			messageConverter = mmc;
+		}
+		else {
+			try {
+				messageConverter = applicationContext.getBean(
+						extendedConsumerProperties.getExtension().getConverterBeanName(), MessageConverter.class);
+			}
+			catch (NoSuchBeanDefinitionException ex) {
+				throw new IllegalStateException(
+						"Converter bean not present in application context", ex);
+			}
+		}
+		return messageConverter;
+	}
+
+	/**
+	 * Get the header mapper bean, if the binder properties contains a bean name; if not
+	 * look for a bean with name  {@code kafkaBinderHeaderMapper} is looked up; if that
+	 * doesn't exist, null is returned.
+	 * @param applicationContext the application context.
+	 * @param configurationProperties the binder properties.
+	 * @return the mapper.
+	 */
+	@Nullable
+	public static KafkaHeaderMapper getHeaderMapper(ApplicationContext applicationContext,
+			KafkaBinderConfigurationProperties configurationProperties) {
+
+		KafkaHeaderMapper mapper = null;
+		if (configurationProperties.getHeaderMapperBeanName() != null) {
+			mapper = applicationContext.getBean(
+					configurationProperties.getHeaderMapperBeanName(),
+					KafkaHeaderMapper.class);
+		}
+		if (mapper == null) {
+			//First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth)
+			try {
+				mapper = applicationContext.getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class);
+			}
+			catch (BeansException be) {
+			}
+		}
+		return mapper;
+	}
+
+	/**
+	 * Create the Kafka configuration map for a consumer binding. With anonymous bindings
+	 * (those without a {@code group} property, which are given a {@code UUID.toString()}
+	 * in the group id) consumption begins from the current end of the topic, otherwise
+	 * consumption starts from the beginning, the first time the binding consumes.
+	 * @param anonymous true if this is for an anonymous binding.
+	 * @param consumerGroup the group.
+	 * @param consumerProperties the binding properties.
+	 * @param configurationProperties the binder properties.
+	 * @return the config map.
+	 */
+	public static Map<String, Object> createConsumerConfigs(boolean anonymous, String consumerGroup,
+			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties,
+			KafkaBinderConfigurationProperties configurationProperties) {
+
+		Map<String, Object> props = new HashMap<>();
+		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+				ByteArrayDeserializer.class);
+		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+				ByteArrayDeserializer.class);
+		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
+		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
+		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
+				anonymous ? "latest" : "earliest");
+		props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
+
+		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,
+					configurationProperties.getKafkaConnectionString());
+		}
+		Map<String, String> config = consumerProperties.getExtension().getConfiguration();
+		if (!ObjectUtils.isEmpty(config)) {
+			Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
+					ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; "
+							+ "use multiple binders instead");
+			props.putAll(config);
+		}
+		if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) {
+			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
+					consumerProperties.getExtension().getStartOffset().name());
+		}
+		return props;
+	}
+
+	/**
+	 * Create the Kafka configuration map for a producer binding.
+	 * @param producerProperties the binding properties.
+	 * @param configurationProperties the binder properties.
+	 * @return the config map.
+	 */
+	public static Map<String, Object> createProducerConfigs(
+			ExtendedProducerProperties<KafkaProducerProperties> producerProperties,
+			KafkaBinderConfigurationProperties configurationProperties) {
+
+		Map<String, Object> props = new HashMap<>();
+		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(configurationProperties.getRequiredAcks()));
+		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());
+		}
+		final KafkaProducerProperties kafkaProducerProperties = producerProperties.getExtension();
+		if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) {
+			props.put(ProducerConfig.BATCH_SIZE_CONFIG,
+					String.valueOf(kafkaProducerProperties.getBufferSize()));
+		}
+		if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) {
+			props.put(ProducerConfig.LINGER_MS_CONFIG,
+					String.valueOf(kafkaProducerProperties.getBatchTimeout()));
+		}
+		if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) {
+			props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
+					kafkaProducerProperties.getCompressionType().toString());
+		}
+		Map<String, String> configs = producerProperties.getExtension().getConfiguration();
+		Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),
+				ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; "
+						+ "use multiple binders instead");
+		if (!ObjectUtils.isEmpty(configs)) {
+			props.putAll(configs);
+		}
+		if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) {
+			props.putAll(kafkaProducerProperties.getConfiguration());
+		}
+		return props;
+	}
+
+}

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java

@@ -17,19 +17,14 @@
 package org.springframework.cloud.stream.binder.reactorkafka;
 
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.reactivestreams.Subscription;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -41,7 +36,6 @@ import reactor.kafka.sender.SenderOptions;
 import reactor.kafka.sender.SenderRecord;
 import reactor.kafka.sender.SenderResult;
 
-import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
 import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
@@ -49,24 +43,25 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
 import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.utils.BindingUtils;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.handler.AbstractMessageHandler;
-import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
+import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.support.converter.RecordMessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.Assert;
-import org.springframework.util.ObjectUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -86,6 +81,10 @@ public class ReactorKafkaBinder
 
 	private KafkaExtendedBindingProperties extendedBindingProperties = new KafkaExtendedBindingProperties();
 
+	private ConsumerConfigCustomizer consumerConfigCustomizer;
+
+	private ProducerConfigCustomizer producerConfigCustomizer;
+
 	public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties,
 			KafkaTopicProvisioner provisioner) {
 
@@ -93,21 +92,28 @@ public class ReactorKafkaBinder
 		this.configurationProperties = configurationProperties;
 	}
 
+	public void setConsumerConfigCustomizer(ConsumerConfigCustomizer consumerConfigCustomizer) {
+		this.consumerConfigCustomizer = consumerConfigCustomizer;
+	}
+
+	public void setProducerConfigCustomizer(ProducerConfigCustomizer producerConfigCustomizer) {
+		this.producerConfigCustomizer = producerConfigCustomizer;
+	}
 
 	@Override
 	protected MessageHandler createProducerMessageHandler(ProducerDestination destination,
 			ExtendedProducerProperties<KafkaProducerProperties> producerProperties, MessageChannel errorChannel)
 			throws Exception {
 
-		Map<String, Object> configs = createProducerConfigs(producerProperties);
-		// TODO: Move config customizers to core
-//		if (this.producerConfigCustomizer != null) {
-//			this.producerConfigCustomizer.configure(props, bindingNameHolder.get(), destination);
-//			bindingNameHolder.remove();
-//		}
+		Map<String, Object> configs = BindingUtils.createProducerConfigs(producerProperties,
+				this.configurationProperties);
+		if (this.producerConfigCustomizer != null) {
+			this.producerConfigCustomizer.configure(configs, producerProperties.getBindingName(),
+					destination.getName());
+		}
 
 		SenderOptions<Object, Object> opts = SenderOptions.create(configs);
-		// TODO bean for converter.
+		// TODO bean for converter; MCB doesn't use one on the producer side.
 		RecordMessageConverter converter = new MessagingMessageConverter();
 		return new ReactorMessageHandler(opts, converter, destination.getName());
 	}
@@ -119,14 +125,16 @@ public class ReactorKafkaBinder
 
 		boolean anonymous = !StringUtils.hasText(group);
 		String consumerGroup = anonymous ? "anonymous." + UUID.randomUUID().toString() : group;
-		Map<String, Object> configs = createConsumerConfigs(anonymous, consumerGroup, properties);
+		Map<String, Object> configs = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties,
+				this.configurationProperties);
 
-		// TODO: Move config customizers to core
-//		if (this.consumerConfigCustomizer != null) {
-//			this.consumerConfigCustomizer.configure(configs, bindingNameHolder.get(), destination);
-//		}
+		if (this.consumerConfigCustomizer != null) {
+			this.consumerConfigCustomizer.configure(configs, properties.getBindingName(), destination.getName());
+		}
 
-		RecordMessageConverter converter = getMessageConverter(properties);
+		MessageConverter converter = BindingUtils.getConsumerMessageConverter(getApplicationContext(), properties,
+				this.configurationProperties);
+		Assert.isInstanceOf(RecordMessageConverter.class, converter);
 		ReceiverOptions<Object, Object> opts = ReceiverOptions.create(configs)
 			.addAssignListener(parts -> System.out.println("Assigned: " + parts))
 			.subscription(Collections.singletonList(destination.getName()));
@@ -143,7 +151,8 @@ public class ReactorKafkaBinder
 				Flux<Message<Object>> flux = receiver
 						.receive()
 						.doOnSubscribe(subs -> this.subscription = subs)
-						.map(record -> (Message<Object>) converter.toMessage(record, null, null, null));
+						.map(record -> (Message<Object>) ((RecordMessageConverter) converter)
+								.toMessage(record, null, null, null));
 				subscribeToPublisher(flux);
 			}
 
@@ -158,126 +167,6 @@ public class ReactorKafkaBinder
 		};
 	}
 
-	/*
-	 * TODO: Copied (and modified) from Kafka binder - refactor to core
-	 */
-	private RecordMessageConverter getMessageConverter(
-			final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
-
-		RecordMessageConverter messageConverter;
-		if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) {
-			MessagingMessageConverter mmc = new MessagingMessageConverter();
-			StandardHeaders standardHeaders = extendedConsumerProperties.getExtension()
-					.getStandardHeaders();
-			mmc.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders)
-							|| StandardHeaders.both.equals(standardHeaders));
-			mmc.setGenerateTimestamp(
-					StandardHeaders.timestamp.equals(standardHeaders)
-							|| StandardHeaders.both.equals(standardHeaders));
-			mmc.setHeaderMapper(new DefaultKafkaHeaderMapper()); //TODO
-			messageConverter = mmc;
-		}
-		else {
-			try {
-				messageConverter = getApplicationContext().getBean(
-						extendedConsumerProperties.getExtension().getConverterBeanName(),
-						RecordMessageConverter.class);
-			}
-			catch (NoSuchBeanDefinitionException ex) {
-				throw new IllegalStateException(
-						"Converter bean not present in application context", ex);
-			}
-		}
-		return messageConverter;
-	}
-
-	/*
-	 * TODO: Copied from Kafka binder - refactor to core
-	 */
-	private Map<String, Object> createConsumerConfigs(boolean anonymous, String consumerGroup,
-			ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties) {
-
-		Map<String, Object> props = new HashMap<>();
-		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
-				ByteArrayDeserializer.class);
-		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
-				ByteArrayDeserializer.class);
-		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
-		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
-		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
-				anonymous ? "latest" : "earliest");
-		props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
-
-		Map<String, Object> mergedConfig = this.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());
-		}
-		Map<String, String> config = consumerProperties.getExtension().getConfiguration();
-		if (!ObjectUtils.isEmpty(config)) {
-			Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
-					ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; "
-							+ "use multiple binders instead");
-			props.putAll(config);
-		}
-		if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) {
-			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
-					consumerProperties.getExtension().getStartOffset().name());
-		}
-		return props;
-	}
-
-	/*
-	 * TODO: Copied from Kafka binder - refactor to core
-	 */
-	private Map<String, Object> createProducerConfigs(
-			ExtendedProducerProperties<KafkaProducerProperties> producerProperties) {
-
-		Map<String, Object> props = new HashMap<>();
-		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()));
-		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());
-		}
-		final KafkaProducerProperties kafkaProducerProperties = producerProperties.getExtension();
-		if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) {
-			props.put(ProducerConfig.BATCH_SIZE_CONFIG,
-					String.valueOf(kafkaProducerProperties.getBufferSize()));
-		}
-		if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) {
-			props.put(ProducerConfig.LINGER_MS_CONFIG,
-					String.valueOf(kafkaProducerProperties.getBatchTimeout()));
-		}
-		if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) {
-			props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
-					kafkaProducerProperties.getCompressionType().toString());
-		}
-		Map<String, String> configs = producerProperties.getExtension().getConfiguration();
-		Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),
-				ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; "
-						+ "use multiple binders instead");
-		if (!ObjectUtils.isEmpty(configs)) {
-			props.putAll(configs);
-		}
-		if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) {
-			props.putAll(kafkaProducerProperties.getConfiguration());
-		}
-		return props;
-	}
-
 	@Override
 	public KafkaConsumerProperties getExtendedConsumerProperties(String channelName) {
 		return this.extendedBindingProperties.getExtendedConsumerProperties(channelName);

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java

@@ -25,6 +25,8 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfi
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.kafka.provisioning.AdminClientConfigCustomizer;
 import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -56,10 +58,14 @@ public class ReactorKafkaBinderConfiguration {
 	@Bean
 	ReactorKafkaBinder reactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties,
 			KafkaTopicProvisioner provisioningProvider,
-			KafkaExtendedBindingProperties extendedBindingProperties) {
+			KafkaExtendedBindingProperties extendedBindingProperties,
+			ObjectProvider<ConsumerConfigCustomizer> consumerConfigCustomizer,
+			ObjectProvider<ProducerConfigCustomizer> producerConfigCustomizer) {
 
 		ReactorKafkaBinder reactorKafkaBinder = new ReactorKafkaBinder(configurationProperties, provisioningProvider);
 		reactorKafkaBinder.setExtendedBindingProperties(extendedBindingProperties);
+		reactorKafkaBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique());
+		reactorKafkaBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique());
 		return reactorKafkaBinder;
 	}
 

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

@@ -24,7 +24,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -49,12 +48,8 @@ import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.header.internals.RecordHeaders;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
 
-import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.DisposableBean;
-import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
 import org.springframework.cloud.stream.binder.BinderHeaders;
 import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
@@ -66,14 +61,14 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.HeaderMode;
 import org.springframework.cloud.stream.binder.MessageValues;
 import org.springframework.cloud.stream.binder.kafka.config.ClientFactoryCustomizer;
-import org.springframework.cloud.stream.binder.kafka.config.ConsumerConfigCustomizer;
-import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties;
-import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties.StandardHeaders;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaProducerProperties;
 import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.utils.BindingUtils;
 import org.springframework.cloud.stream.binder.kafka.utils.DlqDestinationResolver;
 import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction;
 import org.springframework.cloud.stream.binding.DefaultPartitioningInterceptor;
@@ -448,21 +443,7 @@ public class KafkaMessageChannelBinder extends
 		AbstractApplicationContext applicationContext = getApplicationContext();
 		handler.setApplicationContext(applicationContext);
 
-		KafkaHeaderMapper mapper = null;
-		if (this.configurationProperties.getHeaderMapperBeanName() != null) {
-			mapper = applicationContext.getBean(
-					this.configurationProperties.getHeaderMapperBeanName(),
-					KafkaHeaderMapper.class);
-		}
-		if (mapper == null) {
-			//First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth)
-			try {
-				mapper = applicationContext.getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class);
-			}
-			catch (BeansException be) {
-				// Pass through
-			}
-		}
+		KafkaHeaderMapper mapper = BindingUtils.getHeaderMapper(applicationContext, this.configurationProperties);
 
 		/*
 		 * Even if the user configures a bean, we must not use it if the header mode is
@@ -530,44 +511,10 @@ public class KafkaMessageChannelBinder extends
 	protected DefaultKafkaProducerFactory<byte[], byte[]> getProducerFactory(
 			String transactionIdPrefix,
 			ExtendedProducerProperties<KafkaProducerProperties> producerProperties, String beanName, String destination) {
-		Map<String, Object> props = new HashMap<>();
-		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()));
-		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());
-		}
+
+		Map<String, Object> props = BindingUtils.createProducerConfigs(producerProperties,
+				this.configurationProperties);
 		final KafkaProducerProperties kafkaProducerProperties = producerProperties.getExtension();
-		if (ObjectUtils.isEmpty(props.get(ProducerConfig.BATCH_SIZE_CONFIG))) {
-			props.put(ProducerConfig.BATCH_SIZE_CONFIG,
-					String.valueOf(kafkaProducerProperties.getBufferSize()));
-		}
-		if (ObjectUtils.isEmpty(props.get(ProducerConfig.LINGER_MS_CONFIG))) {
-			props.put(ProducerConfig.LINGER_MS_CONFIG,
-					String.valueOf(kafkaProducerProperties.getBatchTimeout()));
-		}
-		if (ObjectUtils.isEmpty(props.get(ProducerConfig.COMPRESSION_TYPE_CONFIG))) {
-			props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,
-					kafkaProducerProperties.getCompressionType().toString());
-		}
-		Map<String, String> configs = producerProperties.getExtension().getConfiguration();
-		Assert.state(!configs.containsKey(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),
-				ProducerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; "
-						+ "use multiple binders instead");
-		if (!ObjectUtils.isEmpty(configs)) {
-			props.putAll(configs);
-		}
-		if (!ObjectUtils.isEmpty(kafkaProducerProperties.getConfiguration())) {
-			props.putAll(kafkaProducerProperties.getConfiguration());
-		}
 		if (this.producerConfigCustomizer != null) {
 			this.producerConfigCustomizer.configure(props, producerProperties.getBindingName(), destination);
 		}
@@ -1082,29 +1029,8 @@ public class KafkaMessageChannelBinder extends
 	private MessageConverter getMessageConverter(
 			final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
 
-		MessageConverter messageConverter;
-		if (extendedConsumerProperties.getExtension().getConverterBeanName() == null) {
-			MessagingMessageConverter mmc = new MessagingMessageConverter();
-			StandardHeaders standardHeaders = extendedConsumerProperties.getExtension()
-					.getStandardHeaders();
-			mmc.setGenerateMessageId(StandardHeaders.id.equals(standardHeaders)
-							|| StandardHeaders.both.equals(standardHeaders));
-			mmc.setGenerateTimestamp(
-					StandardHeaders.timestamp.equals(standardHeaders)
-							|| StandardHeaders.both.equals(standardHeaders));
-			messageConverter = mmc;
-		}
-		else {
-			try {
-				messageConverter = getApplicationContext().getBean(
-						extendedConsumerProperties.getExtension().getConverterBeanName(),
-						MessageConverter.class);
-			}
-			catch (NoSuchBeanDefinitionException ex) {
-				throw new IllegalStateException(
-						"Converter bean not present in application context", ex);
-			}
-		}
+		MessageConverter messageConverter = BindingUtils.getConsumerMessageConverter(getApplicationContext(),
+				extendedConsumerProperties, this.configurationProperties);
 		if (messageConverter instanceof MessagingMessageConverter) {
 			((MessagingMessageConverter) messageConverter).setHeaderMapper(getHeaderMapper(extendedConsumerProperties));
 		}
@@ -1113,36 +1039,26 @@ public class KafkaMessageChannelBinder extends
 
 	private KafkaHeaderMapper getHeaderMapper(
 			final ExtendedConsumerProperties<KafkaConsumerProperties> extendedConsumerProperties) {
-		KafkaHeaderMapper mapper = null;
-		if (this.configurationProperties.getHeaderMapperBeanName() != null) {
-			mapper = getApplicationContext().getBean(
-					this.configurationProperties.getHeaderMapperBeanName(),
-					KafkaHeaderMapper.class);
-		}
+
+		KafkaHeaderMapper mapper = BindingUtils.getHeaderMapper(getApplicationContext(), this.configurationProperties);
 		if (mapper == null) {
-			//First, try to see if there is a bean named headerMapper registered by other frameworks using the binder (for e.g. spring cloud sleuth)
-			try {
-				mapper = getApplicationContext().getBean("kafkaBinderHeaderMapper", KafkaHeaderMapper.class);
-			}
-			catch (BeansException be) {
-				BinderHeaderMapper headerMapper = new BinderHeaderMapper() {
+			BinderHeaderMapper headerMapper = new BinderHeaderMapper() {
 
-					@Override
-					public void toHeaders(Headers source, Map<String, Object> headers) {
-						super.toHeaders(source, headers);
-						if (headers.size() > 0) {
-							headers.put(BinderHeaders.NATIVE_HEADERS_PRESENT, Boolean.TRUE);
-						}
+				@Override
+				public void toHeaders(Headers source, Map<String, Object> headers) {
+					super.toHeaders(source, headers);
+					if (headers.size() > 0) {
+						headers.put(BinderHeaders.NATIVE_HEADERS_PRESENT, Boolean.TRUE);
 					}
-
-				};
-				String[] trustedPackages = extendedConsumerProperties.getExtension()
-						.getTrustedPackages();
-				if (!StringUtils.isEmpty(trustedPackages)) {
-					headerMapper.addTrustedPackages(trustedPackages);
 				}
-				mapper = headerMapper;
+
+			};
+			String[] trustedPackages = extendedConsumerProperties.getExtension()
+					.getTrustedPackages();
+			if (!ObjectUtils.isEmpty(trustedPackages)) {
+				headerMapper.addTrustedPackages(trustedPackages);
 			}
+			mapper = headerMapper;
 		}
 		return mapper;
 	}
@@ -1418,37 +1334,9 @@ public class KafkaMessageChannelBinder extends
 	protected ConsumerFactory<?, ?> createKafkaConsumerFactory(boolean anonymous,
 			String consumerGroup, ExtendedConsumerProperties<KafkaConsumerProperties> consumerProperties,
 			String beanName, String destination) {
-		Map<String, Object> props = new HashMap<>();
-		props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
-				ByteArrayDeserializer.class);
-		props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
-				ByteArrayDeserializer.class);
-		props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
-		props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);
-		props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
-				anonymous ? "latest" : "earliest");
-		props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
 
-		Map<String, Object> mergedConfig = this.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());
-		}
-		Map<String, String> config = consumerProperties.getExtension().getConfiguration();
-		if (!ObjectUtils.isEmpty(config)) {
-			Assert.state(!config.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
-					ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG + " cannot be overridden at the binding level; "
-							+ "use multiple binders instead");
-			props.putAll(config);
-		}
-		if (!ObjectUtils.isEmpty(consumerProperties.getExtension().getStartOffset())) {
-			props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
-					consumerProperties.getExtension().getStartOffset().name());
-		}
+		Map<String, Object> props = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, consumerProperties,
+				this.configurationProperties);
 
 		if (this.consumerConfigCustomizer != null) {
 			this.consumerConfigCustomizer.configure(props, consumerProperties.getBindingName(), destination);

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

@@ -37,6 +37,8 @@ import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfi
 import org.springframework.cloud.stream.binder.kafka.properties.KafkaExtendedBindingProperties;
 import org.springframework.cloud.stream.binder.kafka.provisioning.AdminClientConfigCustomizer;
 import org.springframework.cloud.stream.binder.kafka.provisioning.KafkaTopicProvisioner;
+import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer;
 import org.springframework.cloud.stream.binder.kafka.utils.DlqDestinationResolver;
 import org.springframework.cloud.stream.binder.kafka.utils.DlqPartitionFunction;
 import org.springframework.cloud.stream.config.ConsumerEndpointCustomizer;

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/integration/KafkaConfigCustomizationTests.java

@@ -36,8 +36,8 @@ import org.junit.jupiter.api.extension.ExtendWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.cloud.stream.binder.kafka.config.ConsumerConfigCustomizer;
-import org.springframework.cloud.stream.binder.kafka.config.ProducerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.support.ConsumerConfigCustomizer;
+import org.springframework.cloud.stream.binder.kafka.support.ProducerConfigCustomizer;
 import org.springframework.context.annotation.Bean;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;