From 7cd53119d3841a0e964c6f42186a3affcf9a4ea2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 27 Nov 2017 13:51:32 -0500 Subject: [PATCH] GH-492: JSON Improvements Resolves https://github.com/spring-projects/spring-kafka/issues/492 For `StringJsonMessageConverter` and `JsonSerializer` - Convey type information in Headers using a Jackson type mapper - Setting `addTypeInfo` to `false` disables this feature For `StringJsonMessageConverter` (inbound) - If configured with with the default type mapper, no changes - use the inferred type from the method signature - If configured with a type mapper where the `TypePrecedence` is not `INFERRED`, use the headers if available. For `JsonDeserializer` - use type information from headers, falling back to a default if present For `JsonSerializer` and `JsonDeserializer` - provide configuration (e.g. default type) via kafka properties Polishing - @since * Fix `DefaultJackson2JavaTypeMapper` JavaDoc do not mention Spring AMQP --- .../converter/AbstractJavaTypeMapper.java | 136 ++++++++++++ .../kafka/support/converter/ClassMapper.java | 37 ++++ .../DefaultJackson2JavaTypeMapper.java | 199 ++++++++++++++++++ .../converter/Jackson2JavaTypeMapper.java | 53 +++++ .../converter/MessagingMessageConverter.java | 13 +- .../converter/StringJsonMessageConverter.java | 35 ++- .../support/serializer/JsonDeserializer.java | 115 +++++++++- .../support/serializer/JsonSerializer.java | 71 ++++++- .../EnableKafkaIntegrationTests.java | 48 ++++- .../KafkaMessageListenerContainerTests.java | 114 +++++++++- src/reference/asciidoc/appendix.adoc | 2 +- src/reference/asciidoc/changes-since-1.0.adoc | 26 +++ src/reference/asciidoc/kafka.adoc | 13 ++ src/reference/asciidoc/whats-new.adoc | 28 +-- 14 files changed, 848 insertions(+), 42 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/converter/AbstractJavaTypeMapper.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/converter/ClassMapper.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/converter/DefaultJackson2JavaTypeMapper.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/converter/Jackson2JavaTypeMapper.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/AbstractJavaTypeMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/AbstractJavaTypeMapper.java new file mode 100644 index 00000000..18f92452 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/AbstractJavaTypeMapper.java @@ -0,0 +1,136 @@ +/* + * Copyright 2017 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.kafka.support.converter; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; + +import org.apache.kafka.common.header.Header; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeader; + +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.util.ClassUtils; + +/** + * Abstract type mapper. + * + * @author Mark Pollack + * @author Sam Nelson + * @author Andreas Asplund + * @author Gary Russell + * + * @since 2.1 + */ +public abstract class AbstractJavaTypeMapper implements BeanClassLoaderAware { + + /** + * Default header name for type information. + */ + public static final String DEFAULT_CLASSID_FIELD_NAME = "__TypeId__"; + + /** + * Default header name for container object contents type information. + */ + public static final String DEFAULT_CONTENT_CLASSID_FIELD_NAME = "__ContentTypeId__"; + + /** + * Default header name for map key type information. + */ + public static final String DEFAULT_KEY_CLASSID_FIELD_NAME = "__KeyTypeId__"; + + private final Map> idClassMapping = new HashMap>(); + + private final Map, byte[]> classIdMapping = new HashMap, byte[]>(); + + private ClassLoader classLoader = ClassUtils.getDefaultClassLoader(); + + public String getClassIdFieldName() { + return DEFAULT_CLASSID_FIELD_NAME; + } + + public String getContentClassIdFieldName() { + return DEFAULT_CONTENT_CLASSID_FIELD_NAME; + } + + public String getKeyClassIdFieldName() { + return DEFAULT_KEY_CLASSID_FIELD_NAME; + } + + public void setIdClassMapping(Map> idClassMapping) { + this.idClassMapping.putAll(idClassMapping); + createReverseMap(); + } + + @Override + public void setBeanClassLoader(ClassLoader classLoader) { + this.classLoader = classLoader; + } + + protected ClassLoader getClassLoader() { + return this.classLoader; + } + + protected void addHeader(Headers headers, String headerName, Class clazz) { + if (this.classIdMapping.containsKey(clazz)) { + headers.add(new RecordHeader(headerName, this.classIdMapping.get(clazz))); + } + else { + headers.add(new RecordHeader(headerName, clazz.getName().getBytes(StandardCharsets.UTF_8))); + } + } + + protected String retrieveHeader(Headers headers, String headerName) { + String classId = retrieveHeaderAsString(headers, headerName); + if (classId == null) { + throw new MessageConversionException( + "failed to convert Message content. Could not resolve " + headerName + " in header"); + } + return classId; + } + + protected String retrieveHeaderAsString(Headers headers, String headerName) { + Iterator
headerValues = headers.headers(headerName).iterator(); + if (headerValues.hasNext()) { + Header headerValue = headerValues.next(); + String classId = null; + if (headerValue.value() != null) { + classId = new String(headerValue.value(), StandardCharsets.UTF_8); + } + return classId; + } + return null; + } + + private void createReverseMap() { + this.classIdMapping.clear(); + for (Map.Entry> entry : this.idClassMapping.entrySet()) { + String id = entry.getKey(); + Class clazz = entry.getValue(); + this.classIdMapping.put(clazz, id.getBytes(StandardCharsets.UTF_8)); + } + } + + public Map> getIdClassMapping() { + return Collections.unmodifiableMap(this.idClassMapping); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/ClassMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/ClassMapper.java new file mode 100644 index 00000000..882ce3ae --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/ClassMapper.java @@ -0,0 +1,37 @@ +/* + * Copyright 2017 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.kafka.support.converter; + +import org.apache.kafka.common.header.Headers; + +/** + * Strategy for setting metadata on messages such that one can create the class + * that needs to be instantiated when receiving a message. + * + * @author Mark Pollack + * @author James Carr + * @author Gary Russell + * + * @since 2.1 + */ +public interface ClassMapper { + + void fromClass(Class clazz, Headers headers); + + Class toClass(Headers headers); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/DefaultJackson2JavaTypeMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/DefaultJackson2JavaTypeMapper.java new file mode 100644 index 00000000..e5614298 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/DefaultJackson2JavaTypeMapper.java @@ -0,0 +1,199 @@ +/* + * Copyright 2017 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.kafka.support.converter; + +import java.util.Arrays; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +import org.apache.kafka.common.header.Headers; + +import org.springframework.messaging.converter.MessageConversionException; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; + +import com.fasterxml.jackson.databind.JavaType; +import com.fasterxml.jackson.databind.type.TypeFactory; + +/** + * Jackson 2 type mapper. + * + * @author Mark Pollack + * @author Sam Nelson + * @author Andreas Asplund + * @author Artem Bilan + * @author Gary Russell + * + * @since 2.1 + */ +public class DefaultJackson2JavaTypeMapper extends AbstractJavaTypeMapper + implements Jackson2JavaTypeMapper, ClassMapper { + + private static final List TRUSTED_PACKAGES = + Arrays.asList( + "java.util", + "java.lang" + ); + + private final Set trustedPackages = new LinkedHashSet(TRUSTED_PACKAGES); + + private volatile TypePrecedence typePrecedence = TypePrecedence.INFERRED; + + /** + * Return the precedence. + * @return the precedence. + * @see #setTypePrecedence(Jackson2JavaTypeMapper.TypePrecedence) + */ + @Override + public TypePrecedence getTypePrecedence() { + return this.typePrecedence; + } + + /** + * Set the precedence for evaluating type information in message properties. + * When using {@code @KafkaListener} at the method level, the framework attempts + * to determine the target type for payload conversion from the method signature. + * If so, this type is provided by the {@code MessagingMessageListenerAdapter}. + *

+ * By default, if the type is concrete (not abstract, not an interface), this will + * be used ahead of type information provided in the {@code __TypeId__} and + * associated headers provided by the sender. + *

+ * If you wish to force the use of the {@code __TypeId__} and associated headers + * (such as when the actual type is a subclass of the method argument type), + * set the precedence to {@link Jackson2JavaTypeMapper.TypePrecedence#TYPE_ID}. + * @param typePrecedence the precedence. + */ + public void setTypePrecedence(TypePrecedence typePrecedence) { + Assert.notNull(typePrecedence, "'typePrecedence' cannot be null"); + this.typePrecedence = typePrecedence; + } + + /** + * Specify a set of packages to trust during deserialization. + * The asterisk ({@code *}) means trust all. + * @param trustedPackages the trusted Java packages for deserialization + */ + @Override + public void addTrustedPackages(String... trustedPackages) { + if (trustedPackages != null) { + for (String whiteListClass : trustedPackages) { + if ("*".equals(whiteListClass)) { + this.trustedPackages.clear(); + break; + } + else { + this.trustedPackages.add(whiteListClass); + } + } + } + } + + @Override + public JavaType toJavaType(Headers headers) { + String typeIdHeader = retrieveHeaderAsString(headers, getClassIdFieldName()); + + if (typeIdHeader != null) { + + JavaType classType = getClassIdType(typeIdHeader); + if (!classType.isContainerType() || classType.isArrayType()) { + return classType; + } + + JavaType contentClassType = getClassIdType(retrieveHeader(headers, getContentClassIdFieldName())); + if (classType.getKeyType() == null) { + return TypeFactory.defaultInstance() + .constructCollectionLikeType(classType.getRawClass(), contentClassType); + } + + JavaType keyClassType = getClassIdType(retrieveHeader(headers, getKeyClassIdFieldName())); + return TypeFactory.defaultInstance() + .constructMapLikeType(classType.getRawClass(), keyClassType, contentClassType); + } + + return null; + } + + private JavaType getClassIdType(String classId) { + if (getIdClassMapping().containsKey(classId)) { + return TypeFactory.defaultInstance().constructType(getIdClassMapping().get(classId)); + } + else { + try { + if (!isTrustedPackage(classId)) { + throw new IllegalArgumentException("The class '" + classId + + "' is not in the trusted packages: " + + this.trustedPackages + ". " + + "If you believe this class is safe to deserialize, please provide its name. " + + "If the serialization is only done by a trusted source, you can also enable " + + "trust all (*)."); + } + else { + return TypeFactory.defaultInstance() + .constructType(ClassUtils.forName(classId, getClassLoader())); + } + } + catch (ClassNotFoundException e) { + throw new MessageConversionException("failed to resolve class name. Class not found [" + + classId + "]", e); + } + catch (LinkageError e) { + throw new MessageConversionException("failed to resolve class name. Linkage error [" + + classId + "]", e); + } + } + } + + private boolean isTrustedPackage(String requestedType) { + if (!this.trustedPackages.isEmpty()) { + String packageName = ClassUtils.getPackageName(requestedType).replaceFirst("\\[L", ""); + for (String trustedPackage : this.trustedPackages) { + if (packageName.equals(trustedPackage)) { + return true; + } + } + return false; + } + return true; + } + + @Override + public void fromJavaType(JavaType javaType, Headers headers) { + addHeader(headers, getClassIdFieldName(), javaType.getRawClass()); + + if (javaType.isContainerType() && !javaType.isArrayType()) { + addHeader(headers, getContentClassIdFieldName(), javaType.getContentType().getRawClass()); + } + + if (javaType.getKeyType() != null) { + addHeader(headers, getKeyClassIdFieldName(), javaType.getKeyType().getRawClass()); + } + } + + @Override + public void fromClass(Class clazz, Headers headers) { + fromJavaType(TypeFactory.defaultInstance().constructType(clazz), headers); + + } + + @Override + public Class toClass(Headers headers) { + return toJavaType(headers).getRawClass(); + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/Jackson2JavaTypeMapper.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/Jackson2JavaTypeMapper.java new file mode 100644 index 00000000..de01bc95 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/Jackson2JavaTypeMapper.java @@ -0,0 +1,53 @@ +/* + * Copyright 2017 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.kafka.support.converter; + +import org.apache.kafka.common.header.Headers; + +import com.fasterxml.jackson.databind.JavaType; + +/** + * Strategy for setting metadata on messages such that one can create the class that needs + * to be instantiated when receiving a message. + * + * @author Mark Pollack + * @author James Carr + * @author Sam Nelson + * @author Andreas Asplund + * @author Gary Russell + * + * @since 2.1 + */ +public interface Jackson2JavaTypeMapper extends ClassMapper { + + /** + * The precedence for type conversion - inferred from the method parameter or message + * headers. Only applies if both exist. + */ + enum TypePrecedence { + INFERRED, TYPE_ID + } + + void fromJavaType(JavaType javaType, Headers headers); + + JavaType toJavaType(Headers headers); + + TypePrecedence getTypePrecedence(); + + void addTrustedPackages(String... packages); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java index 9d84a680..24dc409f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java @@ -24,6 +24,7 @@ import org.apache.commons.logging.LogFactory; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.header.internals.RecordHeaders; import org.springframework.kafka.support.Acknowledgment; @@ -136,12 +137,22 @@ public class MessagingMessageConverter implements RecordMessageConverter { Object key = headers.get(KafkaHeaders.MESSAGE_KEY); Object payload = convertPayload(message); Long timestamp = headers.get(KafkaHeaders.TIMESTAMP, Long.class); - RecordHeaders recordHeaders = new RecordHeaders(); + Headers recordHeaders = initialRecordHeaders(message); this.headerMapper.fromHeaders(headers, recordHeaders); return new ProducerRecord(topic == null ? defaultTopic : topic, partition, timestamp, key, payload, recordHeaders); } + /** + * Subclasses can populate additional headers before they are mapped. + * @param message the message. + * @return the headers + * @since 2.1 + */ + protected Headers initialRecordHeaders(Message message) { + return new RecordHeaders(); + } + /** * Subclasses can convert the payload; by default, it's sent unchanged to Kafka. * @param message the message. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/StringJsonMessageConverter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/StringJsonMessageConverter.java index 833584b7..a56dfeb2 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/converter/StringJsonMessageConverter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/converter/StringJsonMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -20,8 +20,11 @@ import java.io.IOException; import java.lang.reflect.Type; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.header.internals.RecordHeaders; import org.springframework.kafka.support.KafkaNull; +import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -43,6 +46,8 @@ public class StringJsonMessageConverter extends MessagingMessageConverter { private final ObjectMapper objectMapper; + private Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); + public StringJsonMessageConverter() { this(new ObjectMapper()); this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false); @@ -54,6 +59,27 @@ public class StringJsonMessageConverter extends MessagingMessageConverter { this.objectMapper = objectMapper; } + public Jackson2JavaTypeMapper getTypeMapper() { + return this.typeMapper; + } + + /** + * Set a customized type mapper. + * @param typeMapper the type mapper. + * @since 2.1 + */ + public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) { + Assert.notNull(typeMapper, "'typeMapper' cannot be null"); + this.typeMapper = typeMapper; + } + + @Override + protected Headers initialRecordHeaders(Message message) { + RecordHeaders headers = new RecordHeaders(); + this.typeMapper.fromClass(message.getPayload().getClass(), headers); + return headers; + } + @Override protected Object convertPayload(Message message) { try { @@ -72,7 +98,12 @@ public class StringJsonMessageConverter extends MessagingMessageConverter { return KafkaNull.INSTANCE; } - JavaType javaType = TypeFactory.defaultInstance().constructType(type); + JavaType javaType = this.typeMapper.getTypePrecedence().equals(TypePrecedence.INFERRED) + ? TypeFactory.defaultInstance().constructType(type) + : this.typeMapper.toJavaType(record.headers()); + if (javaType == null) { // no headers + javaType = TypeFactory.defaultInstance().constructType(type); + } if (value instanceof String) { try { return this.objectMapper.readValue((String) value, javaType); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java index 3c0b58ad..97bfca3c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonDeserializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2017 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. @@ -21,12 +21,19 @@ import java.util.Arrays; import java.util.Map; import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.header.Headers; import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.ExtendedDeserializer; import org.springframework.core.ResolvableType; +import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper; +import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.StringUtils; import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JavaType; import com.fasterxml.jackson.databind.MapperFeature; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectReader; @@ -38,16 +45,34 @@ import com.fasterxml.jackson.databind.ObjectReader; * * @author Igor Stepanov * @author Artem Bilan + * @author Gary Russell */ -public class JsonDeserializer implements Deserializer { +public class JsonDeserializer implements ExtendedDeserializer { + + /** + * Kafka config property for the default key type if no header. + */ + public static final String DEFAULT_KEY_TYPE = "spring.json.key.default.type"; + + /** + * Kafka config property for the default value type if no header. + */ + public static final String DEFAULT_VALUE_TYPE = "spring.json.default.value.type"; + + /** + * Kafka config property for trusted deserialization packages. + */ + public static final String TRUSTED_PACKAGES = "spring.json.trusted.packages"; protected final ObjectMapper objectMapper; - protected final Class targetType; + protected Class targetType; private volatile ObjectReader reader; - protected JsonDeserializer() { + protected Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); + + public JsonDeserializer() { this((Class) null); } @@ -68,14 +93,89 @@ public class JsonDeserializer implements Deserializer { if (targetType == null) { targetType = (Class) ResolvableType.forClass(getClass()).getSuperType().resolveGeneric(0); } - Assert.notNull(targetType, "'targetType' cannot be resolved."); this.targetType = targetType; } - public void configure(Map configs, boolean isKey) { - // No-op + public Jackson2JavaTypeMapper getTypeMapper() { + return this.typeMapper; } + /** + * Set a customized type mapper. + * @param typeMapper the type mapper. + * @since 2.1 + */ + public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) { + Assert.notNull(typeMapper, "'typeMapper' cannot be null"); + this.typeMapper = typeMapper; + } + + @SuppressWarnings("unchecked") + @Override + public void configure(Map configs, boolean isKey) { + try { + if (isKey && configs.containsKey(DEFAULT_KEY_TYPE)) { + if (configs.get(DEFAULT_KEY_TYPE) instanceof Class) { + this.targetType = (Class) configs.get(DEFAULT_KEY_TYPE); + } + else if (configs.get(DEFAULT_KEY_TYPE) instanceof String) { + this.targetType = (Class) ClassUtils.forName((String) configs.get(DEFAULT_KEY_TYPE), null); + } + else { + throw new IllegalStateException(DEFAULT_KEY_TYPE + " must be Class or String"); + } + } + else if (!isKey && configs.containsKey(DEFAULT_VALUE_TYPE)) { + if (configs.get(DEFAULT_VALUE_TYPE) instanceof Class) { + this.targetType = (Class) configs.get(DEFAULT_VALUE_TYPE); + } + else if (configs.get(DEFAULT_VALUE_TYPE) instanceof String) { + this.targetType = (Class) ClassUtils.forName((String) configs.get(DEFAULT_VALUE_TYPE), null); + } + else { + throw new IllegalStateException(DEFAULT_VALUE_TYPE + " must be Class or String"); + } + } + } + catch (ClassNotFoundException | LinkageError e) { + throw new IllegalStateException(e); + } + if (configs.containsKey(TRUSTED_PACKAGES)) { + if (configs.get(TRUSTED_PACKAGES) instanceof String) { + this.typeMapper.addTrustedPackages( + StringUtils.commaDelimitedListToStringArray((String) configs.get(TRUSTED_PACKAGES))); + } + } + } + + /** + * Add trusted packages for deserialization. + * @param packages the packages. + * @since 2.1 + */ + public void addTrustedPackages(String... packages) { + this.typeMapper.addTrustedPackages(packages); + } + + @Override + public T deserialize(String topic, Headers headers, byte[] data) { + JavaType javaType = this.typeMapper.toJavaType(headers); + if (javaType == null) { + Assert.state(this.targetType != null, "No type information in headers and no default type provided"); + return deserialize(topic, data); + } + else { + try { + return this.objectMapper.readerFor(javaType).readValue(data); + } + catch (IOException e) { + throw new SerializationException("Can't deserialize data [" + Arrays.toString(data) + + "] from topic [" + topic + "]", e); + } + } + } + + @Override public T deserialize(String topic, byte[] data) { if (this.reader == null) { this.reader = this.objectMapper.readerFor(this.targetType); @@ -93,6 +193,7 @@ public class JsonDeserializer implements Deserializer { } } + @Override public void close() { // No-op } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java index 7974ea08..fd00417c 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/serializer/JsonSerializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -20,8 +20,12 @@ import java.io.IOException; import java.util.Map; import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.serialization.ExtendedSerializer; import org.apache.kafka.common.serialization.Serializer; +import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper; +import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper; import org.springframework.util.Assert; import com.fasterxml.jackson.databind.DeserializationFeature; @@ -35,11 +39,21 @@ import com.fasterxml.jackson.databind.ObjectMapper; * * @author Igor Stepanov * @author Artem Bilan + * @author Gary Russell */ -public class JsonSerializer implements Serializer { +public class JsonSerializer implements ExtendedSerializer { + + /** + * Kafka config property for to disable adding type headers. + */ + public static final String ADD_TYPE_INFO_HEADERS = "spring.json.add.type.headers"; protected final ObjectMapper objectMapper; + protected boolean addTypeInfo = true; + + protected Jackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); + public JsonSerializer() { this(new ObjectMapper()); this.objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false); @@ -51,10 +65,58 @@ public class JsonSerializer implements Serializer { this.objectMapper = objectMapper; } - public void configure(Map configs, boolean isKey) { - // No-op + public boolean isAddTypeInfo() { + return this.addTypeInfo; } + /** + * Set to false to disable adding type info headers. + * @param addTypeInfo true to add headers. + * @since 2.1 + */ + public void setAddTypeInfo(boolean addTypeInfo) { + this.addTypeInfo = addTypeInfo; + } + + public Jackson2JavaTypeMapper getTypeMapper() { + return this.typeMapper; + } + + /** + * Set a customized type mapper. + * @param typeMapper the type mapper. + * @since 2.1 + */ + public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) { + Assert.notNull(typeMapper, "'typeMapper' cannot be null"); + this.typeMapper = typeMapper; + } + + @Override + public void configure(Map configs, boolean isKey) { + if (configs.containsKey(ADD_TYPE_INFO_HEADERS)) { + Object config = configs.get(ADD_TYPE_INFO_HEADERS); + if (config instanceof Boolean) { + this.addTypeInfo = (Boolean) config; + } + else if (config instanceof String) { + this.addTypeInfo = Boolean.valueOf((String) config); + } + else { + throw new IllegalStateException(ADD_TYPE_INFO_HEADERS + " must be Boolean or String"); + } + } + } + + @Override + public byte[] serialize(String topic, Headers headers, T data) { + if (this.addTypeInfo) { + this.typeMapper.fromJavaType(this.objectMapper.constructType(data.getClass()), headers); + } + return serialize(topic, data); + } + + @Override public byte[] serialize(String topic, T data) { try { byte[] result = null; @@ -68,6 +130,7 @@ public class JsonSerializer implements Serializer { } } + @Override public void close() { // No-op } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 4580acf6..e1ce2279 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -79,6 +79,8 @@ import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.KafkaNull; import org.springframework.kafka.support.SendResult; import org.springframework.kafka.support.TopicPartitionInitialOffset; +import org.springframework.kafka.support.converter.DefaultJackson2JavaTypeMapper; +import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper; import org.springframework.kafka.support.converter.StringJsonMessageConverter; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -119,7 +121,7 @@ public class EnableKafkaIntegrationTests { "annotated18", "annotated19", "annotated20", "annotated21", "annotated21reply", "annotated22", "annotated22reply", "annotated23", "annotated23reply", "annotated24", "annotated24reply", "annotated25", "annotated25reply1", "annotated25reply2", "annotated26", "annotated27", "annotated28", - "annotated29", "annotated30", "annotated30reply"); + "annotated29", "annotated30", "annotated30reply", "annotated31"); // @Rule // public Log4jLevelAdjuster adjuster = new Log4jLevelAdjuster(Level.TRACE, @@ -310,6 +312,28 @@ public class EnableKafkaIntegrationTests { .isEqualTo("buz.explicitGroupId"); } + @Test + @DirtiesContext + public void testJsonHeaders() throws Exception { + ConcurrentMessageListenerContainer container = + (ConcurrentMessageListenerContainer) registry.getListenerContainer("jsonHeaders"); + Object messageListener = container.getContainerProperties().getMessageListener(); + DefaultJackson2JavaTypeMapper typeMapper = KafkaTestUtils.getPropertyValue(messageListener, + "messageConverter.typeMapper", DefaultJackson2JavaTypeMapper.class); + new DirectFieldAccessor(typeMapper).setPropertyValue("typePrecedence", + Jackson2JavaTypeMapper.TypePrecedence.TYPE_ID); + assertThat(container).isNotNull(); + Foo foo = new Foo(); + foo.setBar("bar"); + this.kafkaJsonTemplate.send(MessageBuilder.withPayload(foo) + .setHeader(KafkaHeaders.TOPIC, "annotated31") + .setHeader(KafkaHeaders.PARTITION_ID, 0) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .build()); + assertThat(this.listener.latch19.await(60, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.foo.getBar()).isEqualTo("bar"); + } + @Test public void testNulls() throws Exception { template.send("annotated12", null, null); @@ -649,7 +673,11 @@ public class EnableKafkaIntegrationTests { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); - factory.setMessageConverter(new StringJsonMessageConverter()); + StringJsonMessageConverter converter = new StringJsonMessageConverter(); + DefaultJackson2JavaTypeMapper typeMapper = new DefaultJackson2JavaTypeMapper(); + typeMapper.addTrustedPackages("*"); + converter.setTypeMapper(typeMapper); + factory.setMessageConverter(converter); return factory; } @@ -966,6 +994,8 @@ public class EnableKafkaIntegrationTests { private final CountDownLatch latch18 = new CountDownLatch(2); + private final CountDownLatch latch19 = new CountDownLatch(1); + private final CountDownLatch eventLatch = new CountDownLatch(1); private volatile Integer partition; @@ -1072,6 +1102,14 @@ public class EnableKafkaIntegrationTests { this.latch6.countDown(); } + @KafkaListener(id = "jsonHeaders", topics = "annotated31", + containerFactory = "kafkaJsonListenerContainerFactory", + groupId = "jsonHeaders") + public void jsonHeaders(Bar foo) { // should be mapped to Foo via Headers + this.foo = (Foo) foo; + this.latch19.countDown(); + } + @KafkaListener(id = "rebalanceListener", topics = "annotated11", idIsGroup = false, containerFactory = "kafkaRebalanceListenerContainerFactory") public void listen7(String foo) { @@ -1313,7 +1351,11 @@ public class EnableKafkaIntegrationTests { } - public static class Foo { + public interface Bar { + + } + + public static class Foo implements Bar { private String bar; diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 75818e42..506d53fc 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -52,6 +52,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.WakeupException; import org.junit.ClassRule; @@ -75,6 +76,8 @@ import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapte import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; import org.springframework.kafka.support.TopicPartitionInitialOffset; +import org.springframework.kafka.support.serializer.JsonDeserializer; +import org.springframework.kafka.support.serializer.JsonSerializer; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.ContainerTestUtils; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -91,6 +94,10 @@ public class KafkaMessageListenerContainerTests { private final Log logger = LogFactory.getLog(this.getClass()); + private static String topic1 = "testTopic1"; + + private static String topic2 = "testTopic2"; + private static String topic3 = "testTopic3"; private static String topic4 = "testTopic4"; @@ -122,8 +129,9 @@ public class KafkaMessageListenerContainerTests { private static String topic17 = "testTopic17"; @ClassRule - public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic3, topic4, topic5, - topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17); + public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5, + topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, + topic17); @Rule public TestName testName = new TestName(); @@ -1435,6 +1443,81 @@ public class KafkaMessageListenerContainerTests { logger.info("Stop manual ack rebalance"); } + @Test + public void testJsonSerDeConfiguredType() throws Exception { + this.logger.info("Start JSON1"); + Map props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class); + props.put(JsonDeserializer.DEFAULT_VALUE_TYPE, Foo.class); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + ContainerProperties containerProps = new ContainerProperties(topic1); + + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference> received = new AtomicReference<>(); + containerProps.setMessageListener((MessageListener) record -> { + KafkaMessageListenerContainerTests.this.logger.info("json: " + record); + received.set(record); + latch.countDown(); + }); + + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + container.setBeanName("testJson1"); + container.start(); + + ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); + senderProps.put(JsonSerializer.ADD_TYPE_INFO_HEADERS, false); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic1); + template.sendDefault(0, new Foo("bar")); + template.flush(); + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); + assertThat(received.get().value()).isInstanceOf(Foo.class); + container.stop(); + this.logger.info("Stop JSON1"); + } + + @Test + public void testJsonSerDeHeaderSimpleType() throws Exception { + this.logger.info("Start JSON2"); + Map props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka); + props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class); + props.put(JsonDeserializer.TRUSTED_PACKAGES, "*"); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + ContainerProperties containerProps = new ContainerProperties(topic2); + + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference> received = new AtomicReference<>(); + containerProps.setMessageListener((MessageListener) record -> { + KafkaMessageListenerContainerTests.this.logger.info("json: " + record); + received.set(record); + latch.countDown(); + }); + + KafkaMessageListenerContainer container = + new KafkaMessageListenerContainer<>(cf, containerProps); + container.setBeanName("testJson2"); + container.start(); + + ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class); + ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(topic2); + template.sendDefault(0, new Foo("bar")); + template.flush(); + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); + assertThat(received.get().value()).isInstanceOf(Foo.class); + container.stop(); + this.logger.info("Stop JSON2"); + } + private Consumer spyOnConsumer(KafkaMessageListenerContainer container) { Consumer consumer = spy( KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class)); @@ -1465,4 +1548,31 @@ public class KafkaMessageListenerContainerTests { } + public static class Foo { + + private String bar; + + public Foo() { + super(); + } + + public Foo(String bar) { + this.bar = bar; + } + + public String getBar() { + return this.bar; + } + + public void setBar(String bar) { + this.bar = bar; + } + + @Override + public String toString() { + return "Foo [bar=" + this.bar + "]"; + } + + } + } diff --git a/src/reference/asciidoc/appendix.adoc b/src/reference/asciidoc/appendix.adoc index efdb53a8..1bbf38db 100644 --- a/src/reference/asciidoc/appendix.adoc +++ b/src/reference/asciidoc/appendix.adoc @@ -1,4 +1,4 @@ [[history]] == Change History -include::./changes-1.0-1.1.adoc[] \ No newline at end of file +include::./changes-since-1.0.adoc[] diff --git a/src/reference/asciidoc/changes-since-1.0.adoc b/src/reference/asciidoc/changes-since-1.0.adoc index 4c87af94..08feec36 100644 --- a/src/reference/asciidoc/changes-since-1.0.adoc +++ b/src/reference/asciidoc/changes-since-1.0.adoc @@ -1,4 +1,30 @@ [[migration]] +=== Changes Between 1.3 an 2.0 + +==== Spring Framework and Java Versions + +The Spring for Apache Kafka project now requires Spring Framework 5.0 and Java 8. + +==== @KafkaListener Changes + +You can now annotate `@KafkaListener` methods (and classes, and `@KafkaHandler` methods) with `@SendTo`. +If the method returns a result, it is forwarded to the specified topic. +See <> for more information. + +==== Message Listeners + +Message listeners can now be aware of the `Consumer` object. +See <> for more information. + +==== ConsumerAwareRebalanceListener + +Rebalance listeners can now access the `Consumer` object during rebalance notifications. +See <> for more information. + +==== @EmbeddedKafka Annotation + +For convenience a test class level `@EmbeddedKafka` annotation is provided with the purpose to register `KafkaEmbedded` as a bean. +See <> for more information. === Changes Between 1.2 and 1.3 ==== Support for Transactions diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 34de965f..5df0c057 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1186,6 +1186,14 @@ Both `JsonSerializer` and `JsonDeserializer` can be customized with an `ObjectMa You can also extend them to implement some particular configuration logic in the `configure(Map configs, boolean isKey)` method. +Starting with _version 2.1_, type information can be conveyed in record `Headers`, allowing the handling of multiple types. +In addition, the serializer/deserializer can be configured using Kafka properties. + +- `JsonSerializer.ADD_TYPE_INFO_HEADERS` (default `true`); set to `false` to disable this feature. +- `JsonDeserializer.DEFAULT_KEY_TYPE`; fallback type for deserialization of keys if no header information is present. +- `JsonDeserializer.DEFAULT_VALUE_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. + 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` @@ -1223,6 +1231,11 @@ NOTE: When using the `StringJsonMessageConverter`, you should use a `StringDeser Starting with _version 1.3.2_ you can also use a `StringJsonMessageConverter` within a `BatchMessagingMessageConverter` for converting batch messages, when using a batch listener container factory. +By default, the type for the conversion is inferred from the listener argument. +If you configure the `StringJsonMessageConverter` with a `DefaultJackson2TypeMapper` that has its `TypePrecedence` set to `TYPE_ID` (instead of the default `INFERRED`), then the converter will use type information in headers (if present) instead. +This allows, for example, listener methods to be declared with interfaces instead of concrete classes. +Also, the type converter supports mapping so the deserialization can be to a different type than the source (as long as the data is compatible). + [source, java] ---- @Bean diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 8809ec49..9d8d3cf5 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -1,26 +1,10 @@ -=== What's new in 2.0 Since 1.3 +=== What's new in 2.1 Since 2.0 -==== Spring Framework and Java Versions +==== Kafka Client Version -The Spring for Apache Kafka project now requires Spring Framework 5.0 and Java 8. +This version requires the 1.0.0 `kafka-clients` or higher. -==== @KafkaListener Changes +==== JSON Improvements -You can now annotate `@KafkaListener` methods (and classes, and `@KafkaHandler` methods) with `@SendTo`. -If the method returns a result, it is forwarded to the specified topic. -See <> for more information. - -==== Message Listeners - -Message listeners can now be aware of the `Consumer` object. -See <> for more information. - -==== ConsumerAwareRebalanceListener - -Rebalance listeners can now access the `Consumer` object during rebalance notifications. -See <> for more information. - -==== @EmbeddedKafka Annotation - -For convenience a test class level `@EmbeddedKafka` annotation is provided with the purpose to register `KafkaEmbedded` as a bean. -See <> for more information. +The `StringJsonMessageConverter` and `JsonSerializer` now add type information in `Headers`, allowing the converter and `JsonDeserializer` to create specific types on reception, based on the message itself rather than a fixed configured type. +See <> for more information.