@KafkaListener Add BytesToStringConverter

Currently a `byte[]` gets converted to a String using an
`ArrayToStringConverter`, which is not very useful.

Add a `BytesToStringConverter`.

* Remove obsolete getter.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -17,6 +17,8 @@
 package org.springframework.kafka.annotation;
 
 import java.lang.reflect.Method;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -152,6 +154,8 @@ public class KafkaListenerAnnotationBeanPostProcessor<K, V>
 
 	private BeanExpressionContext expressionContext;
 
+	private Charset charset = StandardCharsets.UTF_8;
+
 	@Override
 	public int getOrder() {
 		return LOWEST_PRECEDENCE;
@@ -204,6 +208,16 @@ public void setBeanFactory(BeanFactory beanFactory) {
 		}
 	}
 
+	/**
+	 * Set a charset to use when converting byte[] to String in method arguments.
+	 * Default UTF-8.
+	 * @param charset the charset.
+	 * @since 2.2
+	 */
+	public void setCharset(Charset charset) {
+		Assert.notNull(charset, "'charset' cannot be null");
+		this.charset = charset;
+	}
 
 	@Override
 	public void afterSingletonsInstantiated() {
@@ -729,7 +743,8 @@ private <T> Collection<T> getBeansOfType(Class<T> type) {
 	 */
 	private class KafkaHandlerMethodFactoryAdapter implements MessageHandlerMethodFactory {
 
-		private final DefaultFormattingConversionService defaultFormattingConversionService = new DefaultFormattingConversionService();
+		private final DefaultFormattingConversionService defaultFormattingConversionService =
+				new DefaultFormattingConversionService();
 
 		private MessageHandlerMethodFactory messageHandlerMethodFactory;
 
@@ -762,6 +777,9 @@ private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
 							(ConfigurableBeanFactory) KafkaListenerAnnotationBeanPostProcessor.this.beanFactory : null);
 
 
+			this.defaultFormattingConversionService.addConverter(
+					new BytesToStringConverter(KafkaListenerAnnotationBeanPostProcessor.this.charset));
+
 			defaultFactory.setConversionService(this.defaultFormattingConversionService);
 
 			List<HandlerMethodArgumentResolver> argumentResolvers = new ArrayList<>();
@@ -789,6 +807,22 @@ protected boolean isEmptyPayload(Object payload) {
 
 	}
 
+	private static class BytesToStringConverter implements Converter<byte[], String> {
+
+
+		private final Charset charset;
+
+		BytesToStringConverter(Charset charset) {
+			this.charset = charset;
+		}
+
+		@Override
+		public String convert(byte[] source) {
+			return new String(source, this.charset);
+		}
+
+	}
+
 	private static class ListenerScope implements Scope {
 
 		private final Map<String, Object> listeners = new HashMap<>();

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -46,6 +46,9 @@
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.OffsetCommitCallback;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -142,7 +145,7 @@ public class EnableKafkaIntegrationTests {
 			"annotated22reply", "annotated23", "annotated23reply", "annotated24", "annotated24reply",
 			"annotated25", "annotated25reply1", "annotated25reply2", "annotated26", "annotated27", "annotated28",
 			"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32", "annotated33",
-			"annotated34", "annotated35");
+			"annotated34", "annotated35", "annotated36");
 
 	private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka();
 
@@ -167,6 +170,9 @@ public class EnableKafkaIntegrationTests {
 	@Autowired
 	public KafkaTemplate<Integer, String> kafkaJsonTemplate;
 
+	@Autowired
+	public KafkaTemplate<byte[], String> bytesKeyTemplate;
+
 	@Autowired
 	public KafkaListenerEndpointRegistry registry;
 
@@ -697,6 +703,13 @@ public void testAutoConfigTm() {
 			.isInstanceOf(ChainedKafkaTransactionManager.class);
 	}
 
+	@Test
+	public void testKeyConversion() throws Exception {
+		this.bytesKeyTemplate.send("annotated36", "foo".getBytes(), "bar");
+		assertThat(this.listener.keyLatch.await(10, TimeUnit.SECONDS)).isTrue();
+		assertThat(this.listener.convertedKey).isEqualTo("foo");
+	}
+
 	@Configuration
 	@EnableKafka
 	@EnableTransactionManagement(proxyTargetClass = true)
@@ -807,6 +820,14 @@ public KafkaListenerContainerFactory<?> kafkaJsonListenerContainerFactory2() {
 			return factory;
 		}
 
+		@Bean
+		public KafkaListenerContainerFactory<?> bytesStringListenerContainerFactory() {
+			ConcurrentKafkaListenerContainerFactory<byte[], String> factory =
+					new ConcurrentKafkaListenerContainerFactory<>();
+			factory.setConsumerFactory(bytesStringConsumerFactory());
+			return factory;
+		}
+
 		@Bean
 		public KafkaListenerContainerFactory<?> batchFactory() {
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
@@ -935,6 +956,13 @@ public DefaultKafkaConsumerFactory<Integer, String> consumerFactory() {
 			return new DefaultKafkaConsumerFactory<>(consumerConfigs());
 		}
 
+		@Bean
+		public DefaultKafkaConsumerFactory<byte[], String> bytesStringConsumerFactory() {
+			Map<String, Object> configs = consumerConfigs();
+			configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
+			return new DefaultKafkaConsumerFactory<>(configs);
+		}
+
 		private ConsumerFactory<Integer, String> configuredConsumerFactory(String clientAndGroupId) {
 			Map<String, Object> configs = consumerConfigs();
 			configs.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
@@ -988,6 +1016,13 @@ public ProducerFactory<Integer, String> txProducerFactory() {
 			return pf;
 		}
 
+		@Bean
+		public ProducerFactory<byte[], String> bytesStringProducerFactory() {
+			Map<String, Object> configs = producerConfigs();
+			configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+			return new DefaultKafkaProducerFactory<>(configs);
+		}
+
 		@Bean
 		public Map<String, Object> producerConfigs() {
 			return KafkaTestUtils.producerProps(embeddedKafka);
@@ -998,6 +1033,11 @@ public KafkaTemplate<Integer, String> template() {
 			return new KafkaTemplate<>(producerFactory());
 		}
 
+		@Bean
+		public KafkaTemplate<byte[], String> bytesKeyTemplate() {
+			return new KafkaTemplate<>(bytesStringProducerFactory());
+		}
+
 		@Bean
 		public KafkaTemplate<Integer, String> partitionZeroReplyingTemplate() {
 			// reply always uses the no-partition, no-key method; subclasses can be used
@@ -1255,6 +1295,10 @@ static class Listener implements ConsumerSeekAware {
 
 		private final CountDownLatch eventLatch = new CountDownLatch(1);
 
+		private final CountDownLatch keyLatch = new CountDownLatch(1);
+
+		private String convertedKey;
+
 		private volatile Integer partition;
 
 		private volatile ConsumerRecord<?, ?> record;
@@ -1525,6 +1569,12 @@ public void ackWithAutoContainerListener(String payload, Acknowledgment ack) {
 			// empty
 		}
 
+		@KafkaListener(topics = "annotated36", containerFactory = "bytesStringListenerContainerFactory")
+		public void bytesKey(String in, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key) {
+			this.convertedKey = key;
+			this.keyLatch.countDown();
+		}
+
 		@KafkaListener(topics = "annotated29")
 		public void anonymousListener(String in) {
 		}