More Spring Integration Kafka Support

- don't map message channel headers, unless they are String
- tolerate `byte[]` in topic header.

* Fix NPE - a null topic header is ok, as long as there's a default provided.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/support/AbstractKafkaHeaderMapper.java

@@ -24,6 +24,7 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.PatternMatchUtils;
 
@@ -64,6 +65,21 @@ public AbstractKafkaHeaderMapper(String... patterns) {
 		}
 	}
 
+	protected boolean matches(String header, Object value) {
+		if (matches(header)) {
+			if ((header.equals(MessageHeaders.REPLY_CHANNEL) || header.equals(MessageHeaders.ERROR_CHANNEL))
+					&& !(value instanceof String)) {
+				if (this.logger.isDebugEnabled()) {
+					this.logger.debug("Cannot map " + header + " when type is [" + value.getClass()
+							+ "]; it must be a String");
+				}
+				return false;
+			}
+			return true;
+		}
+		return false;
+	}
+
 	protected boolean matches(String header) {
 		for (SimplePatternBasedHeaderMatcher matcher : this.matchers) {
 			if (matcher.matchHeader(header)) {

spring-kafka/src/main/java/org/springframework/kafka/support/DefaultKafkaHeaderMapper.java

@@ -168,7 +168,7 @@ public void addTrustedPackages(String... trustedPackages) {
 	public void fromHeaders(MessageHeaders headers, Headers target) {
 		final Map<String, String> jsonHeaders = new HashMap<>();
 		headers.forEach((k, v) -> {
-			if (matches(k)) {
+			if (matches(k, v)) {
 				if (v instanceof byte[]) {
 					target.add(new RecordHeader(k, (byte[]) v));
 				}

spring-kafka/src/main/java/org/springframework/kafka/support/SimpleKafkaHeaderMapper.java

@@ -65,7 +65,7 @@ public SimpleKafkaHeaderMapper(String... patterns) {
 	@Override
 	public void fromHeaders(MessageHeaders headers, Headers target) {
 		headers.forEach((k, v) -> {
-			if (v instanceof byte[] && matches(k)) {
+			if (v instanceof byte[] && matches(k, v)) {
 				target.add(new RecordHeader(k, (byte[]) v));
 			}
 		});

spring-kafka/src/main/java/org/springframework/kafka/support/converter/MessagingMessageConverter.java

@@ -17,6 +17,7 @@
 package org.springframework.kafka.support.converter;
 
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 import java.util.Map;
 
 import org.apache.commons.logging.Log;
@@ -36,6 +37,7 @@
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.Assert;
 
 /**
  * A Messaging {@link MessageConverter} implementation for a message listener that
@@ -133,7 +135,21 @@ public Message<?> toMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowle
 	@Override
 	public ProducerRecord<?, ?> fromMessage(Message<?> message, String defaultTopic) {
 		MessageHeaders headers = message.getHeaders();
-		String topic = headers.get(KafkaHeaders.TOPIC, String.class);
+		Object topicHeader = headers.get(KafkaHeaders.TOPIC);
+		String topic = null;
+		if (topicHeader instanceof byte[]) {
+			topic = new String(((byte[]) topicHeader), StandardCharsets.UTF_8);
+		}
+		else if (topicHeader instanceof String) {
+			topic = (String) topicHeader;
+		}
+		else if (topicHeader == null) {
+			Assert.state(defaultTopic != null, "With no topic header, a defaultTopic is required");
+		}
+		else {
+			throw new IllegalStateException(KafkaHeaders.TOPIC + " must be a String or byte[], not "
+					+ topicHeader.getClass());
+		}
 		Integer partition = headers.get(KafkaHeaders.PARTITION_ID, Integer.class);
 		Object key = headers.get(KafkaHeaders.MESSAGE_KEY);
 		Object payload = convertPayload(message);

spring-kafka/src/test/java/org/springframework/kafka/support/DefaultKafkaHeaderMapperTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-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.
@@ -28,6 +28,7 @@
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper.NonTrustedHeaderType;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.ExecutorSubscribableChannel;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.util.MimeType;
 import org.springframework.util.MimeTypeUtils;
@@ -47,12 +48,14 @@ public void test() {
 				.setHeader("foo", "bar".getBytes())
 				.setHeader("baz", "qux")
 				.setHeader("fix", new Foo())
+				.setHeader(MessageHeaders.REPLY_CHANNEL, new ExecutorSubscribableChannel())
+				.setHeader(MessageHeaders.ERROR_CHANNEL, "errors")
 				.setHeader(MessageHeaders.CONTENT_TYPE, utf8Text)
 				.setHeader("simpleContentType", MimeTypeUtils.TEXT_PLAIN)
 				.build();
 		RecordHeaders recordHeaders = new RecordHeaders();
 		mapper.fromHeaders(message.getHeaders(), recordHeaders);
-		assertThat(recordHeaders.toArray().length).isEqualTo(6); // 5 + json_types
+		assertThat(recordHeaders.toArray().length).isEqualTo(7); // 6 + json_types
 		Map<String, Object> headers = new HashMap<>();
 		mapper.toHeaders(recordHeaders, headers);
 		assertThat(headers.get("foo")).isInstanceOf(byte[].class);
@@ -61,18 +64,20 @@ public void test() {
 		assertThat(headers.get("fix")).isInstanceOf(NonTrustedHeaderType.class);
 		assertThat(headers.get(MessageHeaders.CONTENT_TYPE)).isEqualTo(utf8Text);
 		assertThat(headers.get("simpleContentType")).isEqualTo(MimeTypeUtils.TEXT_PLAIN);
+		assertThat(headers.get(MessageHeaders.REPLY_CHANNEL)).isNull();
+		assertThat(headers.get(MessageHeaders.ERROR_CHANNEL)).isEqualTo("errors");
 		NonTrustedHeaderType ntht = (NonTrustedHeaderType) headers.get("fix");
 		assertThat(ntht.getHeaderValue()).isNotNull();
 		assertThat(ntht.getUntrustedType()).isEqualTo(Foo.class.getName());
-		assertThat(headers.size()).isEqualTo(5);
+		assertThat(headers.size()).isEqualTo(6);
 		mapper.addTrustedPackages(getClass().getPackage().getName());
 		headers = new HashMap<>();
 		mapper.toHeaders(recordHeaders, headers);
 		assertThat(headers.get("foo")).isInstanceOf(byte[].class);
 		assertThat(new String((byte[]) headers.get("foo"))).isEqualTo("bar");
 		assertThat(headers.get("baz")).isEqualTo("qux");
 		assertThat(headers.get("fix")).isEqualTo(new Foo());
-		assertThat(headers.size()).isEqualTo(5);
+		assertThat(headers.size()).isEqualTo(6);
 	}
 
 	public static final class Foo {