GH-1951: Add BytesToNumberConverter

Resolves #1951

Converts `byte[]` (particularly headers) to `long`, `int`, `short` or `byte` (or wrappers).

Author: garyrussell

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

@@ -20,6 +20,7 @@
 import java.io.StringReader;
 import java.lang.reflect.AnnotatedElement;
 import java.lang.reflect.Method;
+import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
@@ -68,6 +69,8 @@
 import org.springframework.core.Ordered;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.core.convert.TypeDescriptor;
+import org.springframework.core.convert.converter.ConditionalGenericConverter;
 import org.springframework.core.convert.converter.Converter;
 import org.springframework.core.convert.converter.GenericConverter;
 import org.springframework.core.log.LogAccessor;
@@ -1081,6 +1084,7 @@ private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
 			defaultFactory.setBeanFactory(KafkaListenerAnnotationBeanPostProcessor.this.beanFactory);
 			this.defaultFormattingConversionService.addConverter(
 					new BytesToStringConverter(KafkaListenerAnnotationBeanPostProcessor.this.charset));
+			this.defaultFormattingConversionService.addConverter(new BytesToNumberConverter());
 			defaultFactory.setConversionService(this.defaultFormattingConversionService);
 			GenericMessageConverter messageConverter = new GenericMessageConverter(this.defaultFormattingConversionService);
 			defaultFactory.setMessageConverter(messageConverter);
@@ -1165,4 +1169,62 @@ public interface AnnotationEnhancer extends BiFunction<Map<String, Object>, Anno
 
 	}
 
+	private final class BytesToNumberConverter implements ConditionalGenericConverter {
+
+		BytesToNumberConverter() {
+		}
+
+		@Override
+		@Nullable
+		public Set<ConvertiblePair> getConvertibleTypes() {
+			HashSet<ConvertiblePair> pairs = new HashSet<>();
+			pairs.add(new ConvertiblePair(byte[].class, long.class));
+			pairs.add(new ConvertiblePair(byte[].class, int.class));
+			pairs.add(new ConvertiblePair(byte[].class, short.class));
+			pairs.add(new ConvertiblePair(byte[].class, byte.class));
+			pairs.add(new ConvertiblePair(byte[].class, Long.class));
+			pairs.add(new ConvertiblePair(byte[].class, Integer.class));
+			pairs.add(new ConvertiblePair(byte[].class, Short.class));
+			pairs.add(new ConvertiblePair(byte[].class, Byte.class));
+			return pairs;
+		}
+
+		@Override
+		@Nullable
+		public Object convert(@Nullable Object source, TypeDescriptor sourceType, TypeDescriptor targetType) {
+			byte[] bytes = (byte[]) source;
+			if (targetType.getType().equals(long.class) || targetType.getType().equals(Long.class)) {
+				Assert.state(bytes.length >= 8, "At least 8 bytes needed to convert a byte[] to a long");
+				return ByteBuffer.wrap(bytes).getLong();
+			}
+			else if (targetType.getType().equals(int.class) || targetType.getType().equals(Integer.class)) {
+				Assert.state(bytes.length >= 4, "At least 4 bytes needed to convert a byte[] to an integer");
+				return ByteBuffer.wrap(bytes).getInt();
+			}
+			else if (targetType.getType().equals(short.class) || targetType.getType().equals(Short.class)) {
+				Assert.state(bytes.length >= 2, "At least 2 bytes needed to convert a byte[] to a short");
+				return ByteBuffer.wrap(bytes).getShort();
+			}
+			else if (targetType.getType().equals(byte.class) || targetType.getType().equals(Byte.class)) {
+				Assert.state(bytes.length >= 1, "At least 1 byte needed to convert a byte[] to a byte");
+				return ByteBuffer.wrap(bytes).get();
+			}
+			return null;
+		}
+
+		@Override
+		public boolean matches(TypeDescriptor sourceType, TypeDescriptor targetType) {
+			if (sourceType.getType().equals(byte[].class)) {
+				Class<?> target = targetType.getType();
+				return target.equals(long.class) || target.equals(int.class) || target.equals(short.class)
+						|| target.equals(byte.class) || target.equals(Long.class) || target.equals(Integer.class)
+						|| target.equals(Short.class) || target.equals(Byte.class);
+			}
+			else {
+				return false;
+			}
+		}
+
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/HeaderMethodArgumentResolverTests.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2021 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
+ *
+ *      https://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.listener;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
+import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+import org.springframework.messaging.support.GenericMessage;
+
+/**
+ * @author Gary Russell
+ * @since 2.8
+ *
+ */
+public class HeaderMethodArgumentResolverTests {
+
+	@SuppressWarnings("rawtypes")
+	@Test
+	void bytesToNumbers() throws Exception {
+		KafkaListenerAnnotationBeanPostProcessor bpp = new KafkaListenerAnnotationBeanPostProcessor<>();
+		MessageHandlerMethodFactory factory = bpp.getMessageHandlerMethodFactory();
+		InvocableHandlerMethod method = factory.createInvocableHandlerMethod(this, getClass().getDeclaredMethod(
+				"numbers", String.class,
+				long.class, int.class, short.class, byte.class, Long.class, Integer.class, Short.class, Byte.class));
+		method.invoke(new GenericMessage<>("foo", Map.of(
+				"l1", ByteBuffer.allocate(8).putLong(1L).array(),
+				"i1", ByteBuffer.allocate(4).putInt(2).array(),
+				"s1", ByteBuffer.allocate(2).putShort((short) 3).array(),
+				"b1", ByteBuffer.allocate(1).put((byte) 4).array(),
+				"l2", ByteBuffer.allocate(8).putLong(5L).array(),
+				"i2", ByteBuffer.allocate(4).putInt(6).array(),
+				"s2", ByteBuffer.allocate(2).putShort((short) 7).array(),
+				"b2", ByteBuffer.allocate(1).put((byte) 8).array())));
+	}
+
+	public void numbers(String payload,
+			@Header("l1") long l1, @Header("i1") int i1, @Header("s1") short s1, @Header("b1") byte b1,
+			@Header("l2") Long l2, @Header("i2") Integer i2, @Header("s2") Short s2, @Header("b2") Byte b2) {
+
+		assertThat(l1).isEqualTo(1L);
+		assertThat(i1).isEqualTo(2);
+		assertThat(s1).isEqualTo((short) 3);
+		assertThat(b1).isEqualTo((byte) 4);
+		assertThat(l2).isEqualTo(5L);
+		assertThat(i2).isEqualTo(6);
+		assertThat(s2).isEqualTo(Short.valueOf((short) 7));
+		assertThat(b2).isEqualTo(Byte.valueOf((byte) 8));
+	}
+
+}