GH-693: Add BytesJsonMessageConverter

Resolves #693

**cherry-pick to 2.1.x**

Author: garyrussell

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/converter/BytesJsonMessageConverter.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2018 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.utils.Bytes;
+
+import org.springframework.messaging.Message;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * JSON Message converter - {@code Bytes} on output, String, Bytes, or byte[] on input.
+ * Used in conjunction with Kafka {@code BytesSerializer/BytesDeserializer}. More
+ * efficient than {@link StringJsonMessageConverter} because the {@code String/byte[]} is
+ * avoided.
+ *
+ * @author Gary Russell
+ * @since 2.1.7
+ *
+ */
+public class BytesJsonMessageConverter extends StringJsonMessageConverter {
+
+	public BytesJsonMessageConverter() {
+		super();
+	}
+
+	public BytesJsonMessageConverter(ObjectMapper objectMapper) {
+		super(objectMapper);
+	}
+
+	@Override
+	protected Object convertPayload(Message<?> message) {
+		try {
+			return Bytes.wrap(getObjectMapper().writeValueAsBytes(message.getPayload()));
+		}
+		catch (JsonProcessingException e) {
+			throw new ConversionException("Failed to convert to JSON", e);
+		}
+	}
+
+}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/support/converter/StringJsonMessageConverter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2017 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -22,6 +22,7 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.utils.Bytes;
 
 import org.springframework.kafka.support.KafkaNull;
 import org.springframework.kafka.support.converter.Jackson2JavaTypeMapper.TypePrecedence;
@@ -36,7 +37,10 @@
 import com.fasterxml.jackson.databind.type.TypeFactory;
 
 /**
- * JSON Message converter - String on output, String or byte[] on input.
+ * JSON Message converter - String on output, String, Bytes, or byte[] on input. Used in
+ * conjunction with Kafka
+ * {@code StringSerializer/StringDeserializer or BytesDeserializer}. Consider using the
+ * BytesJsonMessageConverter instead.
  *
  * @author Gary Russell
  * @author Artem Bilan
@@ -73,6 +77,15 @@ public void setTypeMapper(Jackson2JavaTypeMapper typeMapper) {
 		this.typeMapper = typeMapper;
 	}
 
+	/**
+	 * Return the object mapper.
+	 * @return the mapper.
+	 * @since 2.1.7
+	 */
+	protected ObjectMapper getObjectMapper() {
+		return this.objectMapper;
+	}
+
 	@Override
 	protected Headers initialRecordHeaders(Message<?> message) {
 		RecordHeaders headers = new RecordHeaders();
@@ -104,6 +117,9 @@ protected Object extractAndConvertValue(ConsumerRecord<?, ?> record, Type type)
 		if (javaType == null) { // no headers
 			javaType = TypeFactory.defaultInstance().constructType(type);
 		}
+		if (value instanceof Bytes) {
+			value = ((Bytes) value).get();
+		}
 		if (value instanceof String) {
 			try {
 				return this.objectMapper.readValue((String) value, javaType);

Diff for: spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java

@@ -25,6 +25,9 @@
 import java.util.concurrent.TimeUnit;
 
 import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.BytesDeserializer;
+import org.apache.kafka.common.serialization.BytesSerializer;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -40,7 +43,7 @@
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.support.KafkaHeaders;
 import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
-import org.springframework.kafka.support.converter.StringJsonMessageConverter;
+import org.springframework.kafka.support.converter.BytesJsonMessageConverter;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.messaging.Message;
@@ -128,6 +131,7 @@ public DefaultKafkaConsumerFactory<Integer, String> consumerFactory() {
 		public Map<String, Object> consumerConfigs() {
 			Map<String, Object> consumerProps =
 					KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", embeddedKafka);
+			consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class);
 			consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 			return consumerProps;
 		}
@@ -140,8 +144,8 @@ public KafkaTemplate<Integer, Foo> template() {
 		}
 
 		@Bean
-		public StringJsonMessageConverter converter() {
-			return new StringJsonMessageConverter();
+		public BytesJsonMessageConverter converter() {
+			return new BytesJsonMessageConverter();
 		}
 
 		@Bean
@@ -151,7 +155,9 @@ public ProducerFactory<Integer, Foo> producerFactory() {
 
 		@Bean
 		public Map<String, Object> producerConfigs() {
-			return KafkaTestUtils.producerProps(embeddedKafka);
+			Map<String, Object> props = KafkaTestUtils.producerProps(embeddedKafka);
+			props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, BytesSerializer.class);
+			return props;
 		}
 
 		@Bean

Diff for: src/reference/asciidoc/kafka.adoc

@@ -994,7 +994,7 @@ static class MultiListenerBean {
 Starting with _version 2.1.3_, a `@KafkaHandler` method can be designated as the default method which is invoked if there is no match on other methods.
 At most one method can be so designated.
 When using `@KafkaHandler` methods, the payload must have already been converted to the domain object (so the match can be performed).
-Use a custom deserializer, the `JsonDeserializer` or the `StringJsonMessageConverter` with its `TypePrecedence` set to `TYPE_ID` - see <<serdes>> for more information.
+Use a custom deserializer, the `JsonDeserializer` or the `(String|Bytes)JsonMessageConverter` with its `TypePrecedence` set to `TYPE_ID` - see <<serdes>> for more information.
 
 [[kafkalistener-lifecycle]]
 ===== @KafkaListener Lifecycle Management
@@ -1471,7 +1471,7 @@ In addition, the serializer/deserializer can be configured using Kafka propertie
 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`
-abstraction with the `MessagingMessageConverter` implementation and its `StringJsonMessageConverter` customization.
+abstraction with the `MessagingMessageConverter` implementation and its `StringJsonMessageConverter` and `BytesJsonMessageConverter` customization.
 The `MessageConverter` can be injected into `KafkaTemplate` instance directly and via
 `AbstractKafkaListenerContainerFactory` bean definition for the `@KafkaListener.containerFactory()` property:
 
@@ -1502,14 +1502,17 @@ 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.
+Generally, the `BytesJsonMessageConverter` is more efficient because it avoids a `String` to/from `byte[]` conversion.
 
 [[payload-conversion-with-batch]]
 ===== Payload Conversion with Batch Listeners
 
-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.
+Starting with _version 1.3.2_, you can also use a `StringJsonMessageConverter` or `BytesJsonMessageConverter` within a `BatchMessagingMessageConverter` for converting batch messages, when using a batch listener container factory.
+See <<serdes>> for more information.
 
 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.
+If you configure the `(Bytes|String)JsonMessageConverter` 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).
 This is also useful when using <<class-level-kafkalistener,class-level `@KafkaListener` s>> where the payload must have already been converted, to determine which method to invoke.