Prep work for Kafka Issue #236

See https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/236

- add `none` as a synonym for `HeaderMode.raw`
- add `headers` meaning "use native headers"
- change the default header mode - `null`
- remove `supportsHeadersNatively` - allow binders to chose header mode for each binding
  binders decide their own default when the property is null
- Rabbit will choose `headers` as default (but can be overridden to `none` for a binding)
  `embeddedHeaders` will not be allowed
- Kafka will choose `embeddedHeaders` as default for compatibility with 1.x peer apps
- remove deprecations on header embedding code

Polishing; support fallback

- if the consumer side is configured to support embedded headers,
  binders should look for native headers first and signal their presence
  using the `BinderHeaders.NATIVE_HEADERS_PRESENT` header;
  fall back to embedded headers if it appears the payload might contain
  them
- remove support for XD embedded headers (first byte < 0xff)
- add a quick check for embedded headers by examining the first byte for 0xff
- change the log from error to debug when decoding embedded headers fails

- see the kafka binder PR for a test case with a consumer that can handle all
  message flavors (native, embedded, and no headers).

spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc

@@ -1165,7 +1165,7 @@ Default: null (the default binder will be used, if one exists).
 
 The following binding properties are available for input bindings only and must be prefixed with `spring.cloud.stream.bindings.<channelName>.consumer.`, e.g. `spring.cloud.stream.bindings.input.consumer.concurrency=3`.
 
-Default values can be set by using the prefix `spring.cloud.stream.default.consumer`, e.g. `spring.cloud.stream.default.consumer.headerMode=raw`.
+Default values can be set by using the prefix `spring.cloud.stream.default.consumer`, e.g. `spring.cloud.stream.default.consumer.headerMode=none`.
 
 concurrency::
   The concurrency of the inbound consumer.
@@ -1176,11 +1176,13 @@ partitioned::
 +
 Default: `false`.
 headerMode::
-    When set to `raw`, disables header parsing on input.
+    When set to `none`, disables header parsing on input.
 Effective only for messaging middleware that does not support message headers natively and requires header embedding.
-Useful when inbound data is coming from outside Spring Cloud Stream applications.
+This option is useful when consuming data from non-Spring Cloud Stream applications when native headers are not supported.
+When set to `headers`, uses the middleware's native header mechanism.
+When set to `embeddedHeaders`, embeds headers into the message payload.
 +
-Default: `embeddedHeaders`.
+Default: depends on binder implementation.
 maxAttempts::
   If processing fails, the number of attempts to process the message (including the first).
   Set to 1 to disable retry.
@@ -1252,16 +1254,18 @@ Default: `1`.
 requiredGroups::
   A comma-separated list of groups to which the producer must ensure message delivery even if they start after it has been created (e.g., by pre-creating durable queues in RabbitMQ).
 headerMode::
-  When set to `raw`, disables header embedding on output.
+  When set to `none`, disables header embedding on output.
 Effective only for messaging middleware that does not support message headers natively and requires header embedding.
-Useful when producing data for non-Spring Cloud Stream applications.
+This option is useful when producing data for non-Spring Cloud Stream applications when native headers are not supported.
+When set to `headers`, uses the middleware's native header mechanism.
+When set to `embeddedHeaders`, embeds headers into the message payload.
 +
-Default: `embeddedHeaders`.
+Default: Depends on binder implementation.
 useNativeEncoding::
   When set to `true`, the outbound message is serialized directly by client library, which must be configured correspondingly (e.g. setting an appropriate Kafka producer value serializer).
 When this configuration is being used, the outbound message marshalling is not based on the `contentType` of the binding.
 When native encoding is used, it is the responsibility of the consumer to use appropriate decoder (ex: Kafka consumer value de-serializer) to deserialize the inbound message.
-Also, when native encoding/decoding is used the `headerMode` property is ignored and headers will not be embedded into the message.
+Also, when native encoding/decoding is used the `headerMode=embeddedHeaders` property is ignored and headers will not be embedded into the message.
 +
 Default: `false`.
 errorChannelEnabled::

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -74,24 +74,23 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	protected final PP provisioningProvider;
 
 	/**
-	 * Indicates whether the implementation and the message broker have native support for
-	 * message headers. If false, headers will be embedded in the message payloads.
-	 */
-	private final boolean supportsHeadersNatively;
-
-	/**
-	 * Indicates what headers are to be embedded in the payload if
-	 * {@link #supportsHeadersNatively} is true.
+	 * Indicates which headers are to be embedded in the payload if
+	 * a binding requires embedding headers.
 	 */
 	private final String[] headersToEmbed;
 
-	public AbstractMessageChannelBinder(boolean supportsHeadersNatively, String[] headersToEmbed,
+	public AbstractMessageChannelBinder(String[] headersToEmbed,
 			PP provisioningProvider) {
-		this.supportsHeadersNatively = supportsHeadersNatively;
 		this.headersToEmbed = headersToEmbed == null ? new String[0] : headersToEmbed;
 		this.provisioningProvider = provisioningProvider;
 	}
 
+	@Deprecated
+	protected AbstractMessageChannelBinder(boolean supportsHeadersNatively, String[] headersToEmbed,
+			PP provisioningProvider) {
+		this(headersToEmbed, provisioningProvider);
+	}
+
 	/**
 	 * Binds an outbound channel to a given destination. The implementation delegates to
 	 * {@link ProvisioningProvider#provisionProducerDestination(String, ProducerProperties)}
@@ -141,7 +140,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			((Lifecycle) producerMessageHandler).start();
 		}
 		((SubscribableChannel) outputChannel).subscribe(
-				new SendingHandler(producerMessageHandler, !this.supportsHeadersNatively && HeaderMode.embeddedHeaders
+				new SendingHandler(producerMessageHandler, HeaderMode.embeddedHeaders
 						.equals(producerProperties.getHeaderMode()), this.headersToEmbed,
 						producerProperties.isUseNativeEncoding()));
 
@@ -225,7 +224,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			final ConsumerDestination destination = this.provisioningProvider.provisionConsumerDestination(name, group,
 					properties);
 			final boolean extractEmbeddedHeaders = HeaderMode.embeddedHeaders.equals(
-					properties.getHeaderMode()) && !this.supportsHeadersNatively;
+					properties.getHeaderMode());
 			ReceivingHandler rh = new ReceivingHandler(extractEmbeddedHeaders);
 			rh.setOutputChannel(inputChannel);
 			final FixedSubscriberChannel bridge = new FixedSubscriberChannel(rh);
@@ -525,7 +524,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return destination.getName() + ".errors";
 	}
 
-	@Deprecated
 	private final class ReceivingHandler extends AbstractReplyProducingMessageHandler {
 
 		private final boolean extractEmbeddedHeaders;
@@ -542,13 +540,15 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				return requestMessage;
 			}
 			MessageValues messageValues;
-			if (this.extractEmbeddedHeaders) {
+			if (this.extractEmbeddedHeaders
+					&& !requestMessage.getHeaders().containsKey(BinderHeaders.NATIVE_HEADERS_PRESENT)
+					&& EmbeddedHeaderUtils.mayHaveEmbeddedHeaders((byte[]) requestMessage.getPayload())) {
 				try {
 					messageValues = EmbeddedHeaderUtils.extractHeaders((Message<byte[]>) requestMessage,
 							true);
 				}
 				catch (Exception e) {
-					AbstractMessageChannelBinder.this.logger.error(
+					AbstractMessageChannelBinder.this.logger.debug(
 							EmbeddedHeaderUtils.decodeExceptionMessage(
 									requestMessage),
 							e);
@@ -575,7 +575,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 	}
 
-	@Deprecated
 	private final class SendingHandler extends AbstractMessageHandler implements Lifecycle {
 
 		private final boolean embedHeaders;

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java

@@ -61,7 +61,14 @@ public final class BinderHeaders {
 	 */
 	public static final String PARTITION_OVERRIDE = PREFIX + "partitionOverride";
 
+	/**
+	 * Indicates that an incoming message has native headers.
+	 * @since 2.0
+	 */
+	public static final String NATIVE_HEADERS_PRESENT = PREFIX + "nativeHeadersPresent";
+
 	private BinderHeaders() {
+		super();
 	}
 
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.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.
@@ -25,6 +25,7 @@ import com.fasterxml.jackson.annotation.JsonInclude;
  *
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
  */
 @JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class ConsumerProperties {
@@ -45,7 +46,7 @@ public class ConsumerProperties {
 
 	private double backOffMultiplier = 2.0;
 
-	private HeaderMode headerMode = HeaderMode.embeddedHeaders;
+	private HeaderMode headerMode;
 
 	@Min(value = 1, message = "Concurrency should be greater than zero.")
 	public int getConcurrency() {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/EmbeddedHeaderUtils.java

@@ -16,7 +16,6 @@
 
 package org.springframework.cloud.stream.binder;
 
-import java.io.UnsupportedEncodingException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -24,7 +23,6 @@ import java.util.Map;
 
 import javax.xml.bind.DatatypeConverter;
 
-import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
@@ -111,10 +109,7 @@ public abstract class EmbeddedHeaderUtils {
 			MessageHeaders requestHeaders) throws Exception {
 		ByteBuffer byteBuffer = ByteBuffer.wrap(payload);
 		int headerCount = byteBuffer.get() & 0xff;
-		if (headerCount < 255) {
-			return oldExtractHeaders(byteBuffer, payload, headerCount, copyRequestHeaders, requestHeaders);
-		}
-		else {
+		if (headerCount == 0xff) {
 			headerCount = byteBuffer.get() & 0xff;
 			Map<String, Object> headers = new HashMap<String, Object>();
 			for (int i = 0; i < headerCount; i++) {
@@ -131,6 +126,9 @@ public abstract class EmbeddedHeaderUtils {
 			byteBuffer.get(newPayload);
 			return buildMessageValues(newPayload, headers, copyRequestHeaders, requestHeaders);
 		}
+		else {
+			return buildMessageValues(payload, new HashMap<>(), copyRequestHeaders, requestHeaders);
+		}
 	}
 
 	/**
@@ -146,29 +144,6 @@ public abstract class EmbeddedHeaderUtils {
 		return extractHeaders(payload, false, null);
 	}
 
-	private static MessageValues oldExtractHeaders(ByteBuffer byteBuffer, byte[] bytes, int headerCount,
-			boolean copyRequestHeaders, MessageHeaders requestHeaders) throws UnsupportedEncodingException {
-		Map<String, Object> headers = new HashMap<String, Object>();
-		for (int i = 0; i < headerCount; i++) {
-			int len = byteBuffer.get();
-			String headerName = new String(bytes, byteBuffer.position(), len, "UTF-8");
-			byteBuffer.position(byteBuffer.position() + len);
-			len = byteBuffer.get() & 0xff;
-			String headerValue = new String(bytes, byteBuffer.position(), len, "UTF-8");
-			byteBuffer.position(byteBuffer.position() + len);
-			if (IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(headerName)
-					|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(headerName)) {
-				headers.put(headerName, Integer.parseInt(headerValue));
-			}
-			else {
-				headers.put(headerName, headerValue);
-			}
-		}
-		byte[] newPayload = new byte[byteBuffer.remaining()];
-		byteBuffer.get(newPayload);
-		return buildMessageValues(newPayload, headers, copyRequestHeaders, requestHeaders);
-	}
-
 	private static MessageValues buildMessageValues(byte[] payload, Map<String, Object> headers,
 			boolean copyRequestHeaders, MessageHeaders requestHeaders) {
 		MessageValues messageValues = new MessageValues(payload, headers);
@@ -193,4 +168,14 @@ public abstract class EmbeddedHeaderUtils {
 		return headersToMap;
 	}
 
+	/**
+	 * Return true if the bytes might have embedded headers.
+	 * (First byte is 0xff and long enough for at least one header).
+	 * @param bytes the array.
+	 * @return true if it may have embedded headers.
+	 */
+	public static boolean mayHaveEmbeddedHeaders(byte[] bytes) {
+		return bytes.length > 8 && (bytes[0] & 0xff) == 0xff;
+	}
+
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/HeaderMode.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.
@@ -18,8 +18,28 @@ package org.springframework.cloud.stream.binder;
 
 /**
  * @author Marius Bogoevici
+ * @author Gary Russell
  */
 public enum HeaderMode {
+
+	/**
+	 * @deprecated - use {@link #none}.
+	 */
 	raw,
+
+	/**
+	 * No headers.
+	 */
+	none,
+
+	/**
+	 * Native headers.
+	 */
+	headers,
+
+	/**
+	 * Headers embedded in payload - e.g. kafka < 0.11
+	 */
 	embeddedHeaders
+
 }

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ProducerProperties.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.
@@ -30,6 +30,7 @@ import org.springframework.expression.Expression;
  *
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
  */
 @JsonInclude(Include.NON_DEFAULT)
 public class ProducerProperties {
@@ -48,7 +49,7 @@ public class ProducerProperties {
 
 	private String[] requiredGroups = new String[] {};
 
-	private HeaderMode headerMode = HeaderMode.embeddedHeaders;
+	private HeaderMode headerMode;
 
 	private boolean useNativeEncoding = false;
 

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 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.
@@ -34,13 +34,14 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 /**
  * @author Marius Bogoevici
  * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringBootTest(classes = SourceBindingWithGlobalPropertiesTest.TestSource.class, properties = {
 		"spring.cloud.stream.default.contentType=application/json",
 		"spring.cloud.stream.bindings.output.destination=ticktock",
 		"spring.cloud.stream.default.producer.requiredGroups=someGroup",
-		"spring.cloud.stream.bindings.output.producer.headerMode=raw" })
+		"spring.cloud.stream.bindings.output.producer.headerMode=none" })
 public class SourceBindingWithGlobalPropertiesTest {
 
 	@Autowired
@@ -53,7 +54,7 @@ public class SourceBindingWithGlobalPropertiesTest {
 		Assertions.assertThat(bindingProperties.getContentType()).isEqualTo("application/json");
 		Assertions.assertThat(bindingProperties.getDestination()).isEqualTo("ticktock");
 		Assertions.assertThat(bindingProperties.getProducer().getRequiredGroups()).containsExactly("someGroup");
-		Assertions.assertThat(bindingProperties.getProducer().getHeaderMode()).isEqualTo(HeaderMode.raw);
+		Assertions.assertThat(bindingProperties.getProducer().getHeaderMode()).isEqualTo(HeaderMode.none);
 	}
 
 	@EnableBinding(Source.class)