kGH-21: Add BinderHeaders.PARTITION_OVERRIDE

See spring-cloud/spring-cloud-stream-binder-kafka#109
See #851

Add Prefix to Partition Headers

originalContentType remains as-is for backwards-compatibility.

Enhance Javadoc

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.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.
@@ -21,6 +21,7 @@ import java.util.concurrent.TimeUnit;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -46,9 +47,11 @@ import org.springframework.tuple.Tuple;
 import org.springframework.util.MimeTypeUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertNull;
 
 /**
  * @author Ilayaperumal Gopinathan
+ * @author Gary Russell
  */
 @RunWith(SpringJUnit4ClassRunner.class)
 @SpringBootTest(classes = {MessageChannelConfigurerTests.TestSink.class, MessageChannelConfigurerTests.TestSource.class})
@@ -104,16 +107,19 @@ public class MessageChannelConfigurerTests {
 
 	@Test
 	public void testPartitionHeader() throws Exception {
-		testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").build());
-		Message<?> message = messageCollector.forChannel(testSource.output()).poll(1, TimeUnit.SECONDS);
+		this.testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").build());
+		Message<?> message = this.messageCollector.forChannel(testSource.output()).poll(1, TimeUnit.SECONDS);
 		assertThat(message.getHeaders().get(BinderHeaders.PARTITION_HEADER).equals(0));
+		assertNull(message.getHeaders().get(BinderHeaders.PARTITION_OVERRIDE));
 	}
 
 	@Test
-	public void testPartitionHeaderWithExplicitHeader() throws Exception {
-		testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}").setHeader(BinderHeaders.PARTITION_HEADER, "customerId-123").build());
-		Message<?> message = messageCollector.forChannel(testSource.output()).poll(1, TimeUnit.SECONDS);
-		assertThat(message.getHeaders().get(BinderHeaders.PARTITION_HEADER).equals("customerId-123"));
+	public void testPartitionHeaderWithPartitionOverride() throws Exception {
+		this.testSource.output().send(MessageBuilder.withPayload("{\"message\":\"Hi\"}")
+				.setHeader(BinderHeaders.PARTITION_OVERRIDE, 123).build());
+		Message<?> message = this.messageCollector.forChannel(testSource.output()).poll(1, TimeUnit.SECONDS);
+		assertThat(message.getHeaders().get(BinderHeaders.PARTITION_HEADER).equals(123));
+		assertNull(message.getHeaders().get(BinderHeaders.PARTITION_OVERRIDE));
 	}
 
 	@EnableBinding(Sink.class)

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 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.
@@ -27,6 +27,8 @@ import org.springframework.messaging.MessageHeaders;
  */
 public final class BinderHeaders {
 
+	private static final String PREFIX = "scst_";
+
 	/**
 	 * Indicates the original content type of a message that has been
 	 * transformed in a native transport format.
@@ -34,10 +36,21 @@ public final class BinderHeaders {
 	public static final String BINDER_ORIGINAL_CONTENT_TYPE = "originalContentType";
 
 	/**
-	 * Indicates the target partition of an outbound message. Binders must
-	 * observe this value when sending data on the transport.
+	 * Indicates the target partition of an outbound message.
+	 * Binders must observe this value when sending data on the transport.
+	 * This header is internally set by the framework when partitioning is
+	 * configured.
+	 * It may be overridden by {@link BinderHeaders#PARTITION_OVERRIDE} if set
+	 * by the user.
 	 */
-	public static final String PARTITION_HEADER = "partition";
+	public static final String PARTITION_HEADER = PREFIX + "partition";
+
+	/**
+	 * Indicates the target partition of an outbound message.
+	 * Overrides any partition selected by the binder.
+	 * This header takes precedence over {@link BinderHeaders#PARTITION_HEADER}.
+	 */
+	public static final String PARTITION_OVERRIDE = PREFIX + "partitionOverride";
 
 	/**
 	 * The headers that will be propagated, by default, by binder implementations

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java

@@ -60,6 +60,7 @@ import org.springframework.util.StringUtils;
  * @author Ilayaperumal Gopinathan
  * @author Marius Bogoevici
  * @author Maxim Kirilov
+ * @author Gary Russell
  */
 public class MessageConverterConfigurer implements MessageChannelConfigurer, BeanFactoryAware, InitializingBean {
 
@@ -275,7 +276,7 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
 
 		@Override
 		public Message<?> preSend(Message<?> message, MessageChannel channel) {
-			if (!message.getHeaders().containsKey(BinderHeaders.PARTITION_HEADER)) {
+			if (!message.getHeaders().containsKey(BinderHeaders.PARTITION_OVERRIDE)) {
 				int partition = this.partitionHandler.determinePartition(message);
 				return MessageConverterConfigurer.this.messageBuilderFactory
 						.fromMessage(message)
@@ -283,7 +284,12 @@ public class MessageConverterConfigurer implements MessageChannelConfigurer, Bea
 						.build();
 			}
 			else {
-				return message;
+				return MessageConverterConfigurer.this.messageBuilderFactory
+						.fromMessage(message)
+						.setHeader(BinderHeaders.PARTITION_HEADER,
+								message.getHeaders().get(BinderHeaders.PARTITION_OVERRIDE))
+						.removeHeader(BinderHeaders.PARTITION_OVERRIDE)
+						.build();
 			}
 		}
 	}