GH-2274: Option to Not Configure (De)serializers

Resolves #2274

Provide an option to suppress configuring programmatically provided
`Serializer`s and `Deserializer`s.

Also fix the DKPF, which did not configure serializers provided by the
supplier setter.

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-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.
@@ -89,6 +89,7 @@ public class DefaultKafkaConsumerFactory<K, V> extends KafkaResourceFactory
 
 	private String beanName = "not.managed.by.Spring";
 
+	private boolean configureDeserializers = true;
 
 	/**
 	 * Construct a factory with the provided configuration.
@@ -113,6 +114,23 @@ public DefaultKafkaConsumerFactory(Map<String, Object> configs,
 		this(configs, () -> keyDeserializer, () -> valueDeserializer);
 	}
 
+	/**
+	 * Construct a factory with the provided configuration and deserializers.
+	 * The deserializers' {@code configure()} methods will be called with the
+	 * configuration map unless {@code configureDeserializers} is false.
+	 * @param configs the configuration.
+	 * @param keyDeserializer the key {@link Deserializer}.
+	 * @param valueDeserializer the value {@link Deserializer}.
+	 * @param configureDeserializers false to not configure the deserializers.
+	 * @since 2.8.7
+	 */
+	public DefaultKafkaConsumerFactory(Map<String, Object> configs,
+			@Nullable Deserializer<K> keyDeserializer,
+			@Nullable Deserializer<V> valueDeserializer, boolean configureDeserializers) {
+
+		this(configs, () -> keyDeserializer, () -> valueDeserializer, configureDeserializers);
+	}
+
 	/**
 	 * Construct a factory with the provided configuration and deserializer suppliers.
 	 * When the suppliers are invoked to get an instance, the deserializers'
@@ -126,14 +144,36 @@ public DefaultKafkaConsumerFactory(Map<String, Object> configs,
 			@Nullable Supplier<Deserializer<K>> keyDeserializerSupplier,
 			@Nullable Supplier<Deserializer<V>> valueDeserializerSupplier) {
 
+		this(configs, keyDeserializerSupplier, valueDeserializerSupplier, true);
+	}
+
+	/**
+	 * Construct a factory with the provided configuration and deserializer suppliers.
+	 * When the suppliers are invoked to get an instance, the deserializers'
+	 * {@code configure()} methods will be called with the configuration map unless
+	 * {@code configureDeserializers} is false.
+	 * @param configs the configuration.
+	 * @param keyDeserializerSupplier the key {@link Deserializer} supplier function.
+	 * @param valueDeserializerSupplier the value {@link Deserializer} supplier function.
+	 * @param configureDeserializers false to not configure the deserializers.
+	 * @since 2.8.7
+	 */
+	public DefaultKafkaConsumerFactory(Map<String, Object> configs,
+			@Nullable Supplier<Deserializer<K>> keyDeserializerSupplier,
+			@Nullable Supplier<Deserializer<V>> valueDeserializerSupplier, boolean configureDeserializers) {
+
 		this.configs = new ConcurrentHashMap<>(configs);
+		this.configureDeserializers = configureDeserializers;
 		this.keyDeserializerSupplier = keyDeserializerSupplier(keyDeserializerSupplier);
 		this.valueDeserializerSupplier = valueDeserializerSupplier(valueDeserializerSupplier);
 	}
 
 	private Supplier<Deserializer<K>> keyDeserializerSupplier(
 			@Nullable Supplier<Deserializer<K>> keyDeserializerSupplier) {
 
+		if (!this.configureDeserializers) {
+			return keyDeserializerSupplier;
+		}
 		return keyDeserializerSupplier == null
 				? () -> null
 				: () -> {
@@ -148,6 +188,9 @@ private Supplier<Deserializer<K>> keyDeserializerSupplier(
 	private Supplier<Deserializer<V>> valueDeserializerSupplier(
 			@Nullable Supplier<Deserializer<V>> valueDeserializerSupplier) {
 
+		if (!this.configureDeserializers) {
+			return valueDeserializerSupplier;
+		}
 		return valueDeserializerSupplier == null
 				? () -> null
 				: () -> {
@@ -165,23 +208,29 @@ public void setBeanName(String name) {
 	}
 
 	/**
-	 * Set the key deserializer.
+	 * Set the key deserializer. The deserializer will be configured using the consumer
+	 * configuration, unless {@link #setConfigureDeserializers(boolean)
+	 * configureDeserializers} is false.
 	 * @param keyDeserializer the deserializer.
 	 */
 	public void setKeyDeserializer(@Nullable Deserializer<K> keyDeserializer) {
 		this.keyDeserializerSupplier = keyDeserializerSupplier(() -> keyDeserializer);
 	}
 
 	/**
-	 * Set the value deserializer.
+	 * Set the value deserializer. The deserializer will be configured using the consumer
+	 * configuration, unless {@link #setConfigureDeserializers(boolean)
+	 * configureDeserializers} is false.
 	 * @param valueDeserializer the value deserializer.
 	 */
 	public void setValueDeserializer(@Nullable Deserializer<V> valueDeserializer) {
 		this.valueDeserializerSupplier = valueDeserializerSupplier(() -> valueDeserializer);
 	}
 
 	/**
-	 * Set a supplier to supply instances of the key deserializer.
+	 * Set a supplier to supply instances of the key deserializer. The deserializer will
+	 * be configured using the consumer configuration, unless
+	 * {@link #setConfigureDeserializers(boolean) configureDeserializers} is false.
 	 * @param keyDeserializerSupplier the supplier.
 	 * @since 2.8
 	 */
@@ -190,14 +239,32 @@ public void setKeyDeserializerSupplier(Supplier<Deserializer<K>> keyDeserializer
 	}
 
 	/**
-	 * Set a supplier to supply instances of the value deserializer.
+	 * Set a supplier to supply instances of the value deserializer. The deserializer will
+	 * be configured using the consumer configuration, unless
+	 * {@link #setConfigureDeserializers(boolean) configureDeserializers} is false.
 	 * @param valueDeserializerSupplier the supplier.
 	 * @since 2.8
 	 */
 	public void setValueDeserializerSupplier(Supplier<Deserializer<V>> valueDeserializerSupplier) {
 		this.valueDeserializerSupplier = valueDeserializerSupplier(valueDeserializerSupplier);
 	}
 
+
+	/**
+	 * Set to false (default true) to prevent programmatically provided deserializers (via
+	 * constructor or setters) from being configured using the producer configuration,
+	 * e.g. if the deserializers are already fully configured.
+	 * @param configureDeserializers false to not configure.
+	 * @since 2.8.7
+	 * @see #setKeyDeserializer(Deserializer)
+	 * @see #setKeyDeserializerSupplier(Supplier)
+	 * @see #setValueDeserializer(Deserializer)
+	 * @see #setValueDeserializerSupplier(Supplier)
+	 **/
+	public void setConfigureDeserializers(boolean configureDeserializers) {
+		this.configureDeserializers = configureDeserializers;
+	}
+
 	@Override
 	public Map<String, Object> getConfigurationProperties() {
 		Map<String, Object> configs2 = new HashMap<>(this.configs);

spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java

@@ -149,6 +149,8 @@ public class DefaultKafkaProducerFactory<K, V> extends KafkaResourceFactory
 
 	private long maxAge;
 
+	private boolean configureSerializers = true;
+
 	private volatile String transactionIdPrefix;
 
 	private volatile String clientIdPrefix;
@@ -178,16 +180,37 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 			@Nullable Serializer<K> keySerializer,
 			@Nullable Serializer<V> valueSerializer) {
 
-		this(configs, () -> keySerializer, () -> valueSerializer);
+		this(configs, () -> keySerializer, () -> valueSerializer, true);
 	}
 
 	/**
-	 * Construct a factory with the provided configuration and {@link Serializer} Suppliers.
-	 * Also configures a {@link #transactionIdPrefix} as a value from the
-	 * {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} if provided.
-	 * This config is going to be overridden with a suffix for target {@link Producer} instance.
-	 * When the suppliers are invoked to get an instance, the serializers'
-	 * {@code configure()} methods will be called with the configuration map.
+	 * Construct a factory with the provided configuration and {@link Serializer}s. Also
+	 * configures a {@link #transactionIdPrefix} as a value from the
+	 * {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} if provided. This config is going to
+	 * be overridden with a suffix for target {@link Producer} instance. The serializers'
+	 * {@code configure()} methods will be called with the configuration map unless
+	 * {@code configureSerializers} is false..
+	 * @param configs the configuration.
+	 * @param keySerializer the key {@link Serializer}.
+	 * @param valueSerializer the value {@link Serializer}.
+	 * @param configureSerializers set to false if serializers are already fully
+	 * configured.
+	 * @since 2.8.7
+	 */
+	public DefaultKafkaProducerFactory(Map<String, Object> configs,
+			@Nullable Serializer<K> keySerializer,
+			@Nullable Serializer<V> valueSerializer, boolean configureSerializers) {
+
+		this(configs, () -> keySerializer, () -> valueSerializer, configureSerializers);
+	}
+
+	/**
+	 * Construct a factory with the provided configuration and {@link Serializer}
+	 * Suppliers. Also configures a {@link #transactionIdPrefix} as a value from the
+	 * {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} if provided. This config is going to
+	 * be overridden with a suffix for target {@link Producer} instance. When the
+	 * suppliers are invoked to get an instance, the serializers' {@code configure()}
+	 * methods will be called with the configuration map.
 	 * @param configs the configuration.
 	 * @param keySerializerSupplier the key {@link Serializer} supplier function.
 	 * @param valueSerializerSupplier the value {@link Serializer} supplier function.
@@ -197,7 +220,30 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 			@Nullable Supplier<Serializer<K>> keySerializerSupplier,
 			@Nullable Supplier<Serializer<V>> valueSerializerSupplier) {
 
+		this(configs, keySerializerSupplier, valueSerializerSupplier, true);
+	}
+
+	/**
+	 * Construct a factory with the provided configuration and {@link Serializer}
+	 * Suppliers. Also configures a {@link #transactionIdPrefix} as a value from the
+	 * {@link ProducerConfig#TRANSACTIONAL_ID_CONFIG} if provided. This config is going to
+	 * be overridden with a suffix for target {@link Producer} instance. When the
+	 * suppliers are invoked to get an instance, the serializers' {@code configure()}
+	 * methods will be called with the configuration map unless
+	 * {@code configureSerializers} is false.
+	 * @param configs the configuration.
+	 * @param keySerializerSupplier the key {@link Serializer} supplier function.
+	 * @param valueSerializerSupplier the value {@link Serializer} supplier function.
+	 * @param configureSerializers set to false if serializers are already fully
+	 * configured.
+	 * @since 2.8.7
+	 */
+	public DefaultKafkaProducerFactory(Map<String, Object> configs,
+			@Nullable Supplier<Serializer<K>> keySerializerSupplier,
+			@Nullable Supplier<Serializer<V>> valueSerializerSupplier, boolean configureSerializers) {
+
 		this.configs = new ConcurrentHashMap<>(configs);
+		this.configureSerializers = configureSerializers;
 		this.keySerializerSupplier = keySerializerSupplier(keySerializerSupplier);
 		this.valueSerializerSupplier = valueSerializerSupplier(valueSerializerSupplier);
 		if (this.clientIdPrefix == null && configs.get(ProducerConfig.CLIENT_ID_CONFIG) instanceof String) {
@@ -212,6 +258,9 @@ public DefaultKafkaProducerFactory(Map<String, Object> configs,
 
 	private Supplier<Serializer<K>> keySerializerSupplier(@Nullable Supplier<Serializer<K>> keySerializerSupplier) {
 		this.rawKeySerializerSupplier = keySerializerSupplier;
+		if (!this.configureSerializers) {
+			return keySerializerSupplier;
+		}
 		return keySerializerSupplier == null
 				? () -> null
 				: () -> {
@@ -225,6 +274,9 @@ private Supplier<Serializer<K>> keySerializerSupplier(@Nullable Supplier<Seriali
 
 	private Supplier<Serializer<V>> valueSerializerSupplier(@Nullable Supplier<Serializer<V>> valueSerializerSupplier) {
 		this.rawValueSerializerSupplier = valueSerializerSupplier;
+		if (!this.configureSerializers) {
+			return valueSerializerSupplier;
+		}
 		return valueSerializerSupplier == null
 				? () -> null
 				: () -> {
@@ -247,37 +299,79 @@ public void setBeanName(String name) {
 	}
 
 	/**
-	 * Set a key serializer.
+	 * Set a key serializer. The serializer will be configured using the producer
+	 * configuration, unless {@link #setConfigureSerializers(boolean)
+	 * configureSerializers} is false.
 	 * @param keySerializer the key serializer.
+	 * @see #setConfigureSerializers(boolean)
 	 */
 	public void setKeySerializer(@Nullable Serializer<K> keySerializer) {
 		this.keySerializerSupplier = keySerializerSupplier(() -> keySerializer);
 	}
 
 	/**
-	 * Set a value serializer.
+	 * Set a value serializer. The serializer will be configured using the producer
+	 * configuration, unless {@link #setConfigureSerializers(boolean)
+	 * configureSerializers} is false.
 	 * @param valueSerializer the value serializer.
+	 * @see #setConfigureSerializers(boolean)
 	 */
 	public void setValueSerializer(@Nullable Serializer<V> valueSerializer) {
 		this.valueSerializerSupplier = valueSerializerSupplier(() -> valueSerializer);
 	}
 
 	/**
-	 * Set a supplier to supply instances of the key serializer.
+	 * Set a supplier to supply instances of the key serializer. The serializer will be
+	 * configured using the producer configuration, unless
+	 * {@link #setConfigureSerializers(boolean) configureSerializers} is false.
 	 * @param keySerializerSupplier the supplier.
 	 * @since 2.8
+	 * @see #setConfigureSerializers(boolean)
 	 */
 	public void setKeySerializerSupplier(Supplier<Serializer<K>> keySerializerSupplier) {
-		this.keySerializerSupplier = keySerializerSupplier;
+		this.keySerializerSupplier = keySerializerSupplier(keySerializerSupplier);
 	}
 
 	/**
 	 * Set a supplier to supply instances of the value serializer.
-	 * @param valueSerializerSupplier the supplier.
+	 * @param valueSerializerSupplier the supplier. The serializer will be configured
+	 * using the producer configuration, unless {@link #setConfigureSerializers(boolean)
+	 * configureSerializers} is false.
 	 * @since 2.8
+	 * @see #setConfigureSerializers(boolean)
 	 */
 	public void setValueSerializerSupplier(Supplier<Serializer<V>> valueSerializerSupplier) {
-		this.valueSerializerSupplier = valueSerializerSupplier;
+		this.valueSerializerSupplier = valueSerializerSupplier(valueSerializerSupplier);
+	}
+
+	/**
+	 * If true (default), programmatically provided serializers (via constructor or
+	 * setters) will be configured using the producer configuration. Set to false if the
+	 * serializers are already fully configured.
+	 * @return true to configure.
+	 * @since 2.8.7
+	 * @see #setKeySerializer(Serializer)
+	 * @see #setKeySerializerSupplier(Supplier)
+	 * @see #setValueSerializer(Serializer)
+	 * @see #setValueSerializerSupplier(Supplier)
+	 */
+	public boolean isConfigureSerializers() {
+		return this.configureSerializers;
+	}
+
+	/**
+	 * Set to false (default true) to prevent programmatically provided serializers (via
+	 * constructor or setters) from being configured using the producer configuration,
+	 * e.g. if the serializers are already fully configured.
+	 * @param configureSerializers false to not configure.
+	 * @since 2.8.7
+	 * @see #setKeySerializer(Serializer)
+	 * @see #setKeySerializerSupplier(Supplier)
+	 * @see #setValueSerializer(Serializer)
+	 * @see #setValueSerializerSupplier(Supplier)
+	 */
+	public void setConfigureSerializers(boolean configureSerializers) {
+		this.configureSerializers = configureSerializers;
 	}
 
 	/**
@@ -441,10 +535,10 @@ public ProducerFactory<K, V> copyWithConfigurationOverride(Map<String, Object> o
 		Map<String, Object> producerProperties = new HashMap<>(getConfigurationProperties());
 		producerProperties.putAll(overrideProperties);
 		producerProperties = ensureExistingTransactionIdPrefixInProperties(producerProperties);
-		DefaultKafkaProducerFactory<K, V> newFactory =
-				new DefaultKafkaProducerFactory<>(producerProperties,
+		DefaultKafkaProducerFactory<K, V> newFactory = new DefaultKafkaProducerFactory<>(producerProperties,
 						getKeySerializerSupplier(),
-						getValueSerializerSupplier());
+						getValueSerializerSupplier(),
+						isConfigureSerializers());
 		newFactory.setPhysicalCloseTimeout((int) getPhysicalCloseTimeout().getSeconds());
 		newFactory.setProducerPerThread(isProducerPerThread());
 		for (ProducerPostProcessor<K, V> templatePostProcessor : getPostProcessors()) {