From cadd61d3e520eedfcd40ce2933cd9e8e998eb8e5 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 19 Jul 2018 12:04:57 -0400 Subject: [PATCH] GH-739: Custom (De)Serializer Doc Improvement Resolves https://github.com/spring-projects/spring-kafka/issues/739 * Polishing - `` instead of `` * Fix typo --- .../core/DefaultKafkaConsumerFactory.java | 15 +++++++-- .../core/DefaultKafkaProducerFactory.java | 16 ++++++++-- src/reference/asciidoc/kafka.adoc | 32 ++++++++++++++++--- 3 files changed, 55 insertions(+), 8 deletions(-) diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java index 634d310f..8310171b 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaConsumerFactory.java +++ b/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 implements ConsumerFactory private Deserializer valueDeserializer; + /** + * Construct a factory with the provided configuration. + * @param configs the configuration. + */ public DefaultKafkaConsumerFactory(Map 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 configs, - Deserializer keyDeserializer, - Deserializer valueDeserializer) { + @Nullable Deserializer keyDeserializer, + @Nullable Deserializer valueDeserializer) { this.configs = new HashMap<>(configs); this.keyDeserializer = keyDeserializer; this.valueDeserializer = valueDeserializer; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java index 8b53b6ba..b119251c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/DefaultKafkaProducerFactory.java +++ b/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 implements ProducerFactory, private volatile boolean running; + /** + * Construct a factory with the provided configuration. + * @param configs the configuration. + */ public DefaultKafkaProducerFactory(Map configs) { this(configs, null, null); } - public DefaultKafkaProducerFactory(Map configs, Serializer keySerializer, - Serializer 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 configs, + @Nullable Serializer keySerializer, + @Nullable Serializer valueSerializer) { this.configs = new HashMap<>(configs); this.keySerializer = keySerializer; this.valueSerializer = valueSerializer; diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 3d3740a4..fa5c9073 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/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 kafkaConsumerFactory(KafkaProperties properties, + JsonDeserializer customDeserializer) { + + return new DefaultKafkaConsumerFactory<>(properties.buildConsumerProperties(), + customDeserializer, customDeserializer); +} + +@Bean +public ProducererFactory 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 <>. 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 <>). Generally, the `BytesJsonMessageConverter` is more efficient because it avoids a `String` to/from `byte[]` conversion. [[error-handling-deserializer]]