GH-739: Custom (De)Serializer Doc Improvement

Resolves https://github.com/spring-projects/spring-kafka/issues/739

* Polishing - `<Foo, Bar>` instead of `<String, String>`

* Fix typo

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

@@ -25,6 +25,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.serialization.Deserializer;
 
+import org.springframework.lang.Nullable;
 import org.springframework.util.StringUtils;
 
 /**
@@ -48,13 +49,23 @@ public class DefaultKafkaConsumerFactory<K, V> implements ConsumerFactory<K, V>
 
 	private Deserializer<V> valueDeserializer;
 
+	/**
+	 * Construct a factory with the provided configuration.
+	 * @param configs the configuration.
+	 */
 	public DefaultKafkaConsumerFactory(Map<String, Object> configs) {
 		this(configs, null, null);
 	}
 
+	/**
+	 * Construct a factory with the provided configuration and deserializers.
+	 * @param configs the configuration.
+	 * @param keyDeserializer the key {@link Deserializer}.
+	 * @param valueDeserializer the value {@link Deserializer}.
+	 */
 	public DefaultKafkaConsumerFactory(Map<String, Object> configs,
-			Deserializer<K> keyDeserializer,
-			Deserializer<V> valueDeserializer) {
+			@Nullable Deserializer<K> keyDeserializer,
+			@Nullable Deserializer<V> valueDeserializer) {
 		this.configs = new HashMap<>(configs);
 		this.keyDeserializer = keyDeserializer;
 		this.valueDeserializer = valueDeserializer;

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

@@ -44,6 +44,7 @@ import org.apache.kafka.common.serialization.Serializer;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.Lifecycle;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
 /**
@@ -93,12 +94,23 @@ public class DefaultKafkaProducerFactory<K, V> implements ProducerFactory<K, V>,
 
 	private volatile boolean running;
 
+	/**
+	 * Construct a factory with the provided configuration.
+	 * @param configs the configuration.
+	 */
 	public DefaultKafkaProducerFactory(Map<String, Object> configs) {
 		this(configs, null, null);
 	}
 
-	public DefaultKafkaProducerFactory(Map<String, Object> configs, Serializer<K> keySerializer,
-			Serializer<V> valueSerializer) {
+	/**
+	 * Construct a factory with the provided configuration and {@link Serializer}s.
+	 * @param configs the configuration.
+	 * @param keySerializer the key {@link Serializer}.
+	 * @param valueSerializer the value {@link Serializer}.
+	 */
+	public DefaultKafkaProducerFactory(Map<String, Object> configs,
+			@Nullable Serializer<K> keySerializer,
+			@Nullable Serializer<V> valueSerializer) {
 		this.configs = new HashMap<>(configs);
 		this.keySerializer = keySerializer;
 		this.valueSerializer = valueSerializer;

src/reference/asciidoc/kafka.adoc

@@ -1557,10 +1557,9 @@ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
 for more complex or particular cases, the `KafkaConsumer`, and therefore `KafkaProducer`, provides overloaded
 constructors to accept `(De)Serializer` instances for `keys` and/or `values`, respectively.
 
-To meet this API, the `DefaultKafkaProducerFactory` and `DefaultKafkaConsumerFactory` also provide properties to allow
-to inject a custom `(De)Serializer` to target `Producer`/`Consumer`.
+Using this API, the `DefaultKafkaProducerFactory` and `DefaultKafkaConsumerFactory` also provide properties (via constructors or setter methods) to inject custom `(De)Serializer` s to the target `Producer`/`Consumer`.
 
-For this purpose, Spring for Apache Kafka also provides `JsonSerializer`/`JsonDeserializer` implementations based on the
+Spring for Apache Kafka also provides `JsonSerializer`/`JsonDeserializer` implementations based on the
 Jackson JSON object mapper.
 The `JsonSerializer` is quite simple and just allows writing any Java object as a JSON `byte[]`, the `JsonDeserializer`
 requires an additional `Class<?> targetType` argument to allow the deserialization of a consumed `byte[]` to the proper target
@@ -1583,6 +1582,31 @@ In addition, the serializer/deserializer can be configured using Kafka propertie
 - `JsonDeserializer.VALUE_DEFAULT_TYPE`; fallback type for deserialization of values if no header information is present.
 - `JsonDeserializer.TRUSTED_PACKAGES` (default `java.util`, `java.lang`); comma-delimited list of package patterns allowed for deserialization; `*` means deserialize all.
 
+IMPORTANT: Only simple configuration can be performed with properties; for more advanced configuration (such as using a custom `ObjectMapper` in the serializer/deserializer), you should use the producer/consumer factory constructors that accept a pre-built serializer and deserializer. For example, with Spring Boot, to override the default factories:
+
+[source, java]
+----
+@Bean
+public ConsumerFactory<Foo, Bar> kafkaConsumerFactory(KafkaProperties properties,
+    JsonDeserializer customDeserializer) {
+
+    return new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties(),
+        customDeserializer, customDeserializer);
+}
+
+@Bean
+public ProducererFactory<Foo, Bar> kafkaProducerFactory(KafkaProperties properties,
+    JsonSserializer customSerializer) {
+
+    return new DefaultKafkaConsumerFactory<>(properties.buildProducerProperties(),
+        customSerializer, customSerializer);
+}
+----
+
+Setters are also provided, as an alternative to using these constructors.
+
+===== Spring Messaging Message Conversion
+
 Although the `Serializer`/`Deserializer` API is quite simple and flexible from the low-level Kafka `Consumer` and
 `Producer` perspective, you might need more flexibility at the Spring Messaging level, either when using `@KafkaListener` or <<si-kafka,Spring Integration>>.
 To easily convert to/from `org.springframework.messaging.Message`, Spring for Apache Kafka provides a `MessageConverter`
@@ -1617,7 +1641,7 @@ With a class-level `@KafkaListener`, the payload type is used to select which `@
 ====
 
 NOTE: When using the `StringJsonMessageConverter`, you should use a `StringDeserializer` in the kafka consumer configuration and `StringSerializer` in the kafka producer configuration, when using Spring Integration or the `KafkaTemplate.send(Message<?> message)` method.
-When using the `BytesJsonMessageConverter`, you should use a `BytesDeserializer` in the kafka consumer configuration and `BytesSerializer` in the kafka producer configuration, when using Spring Integration or the `KafkaTemplate.send(Message<?> message)` method.
+When using the `BytesJsonMessageConverter`, you should use a `BytesDeserializer` in the kafka consumer configuration and `BytesSerializer` in the kafka producer configuration, when using Spring Integration or the `KafkaTemplate.send(Message<?> message)` method (see <<kafka-template>>).
 Generally, the `BytesJsonMessageConverter` is more efficient because it avoids a `String` to/from `byte[]` conversion.
 
 [[error-handling-deserializer]]