From bd1454bffff52d29f113f040076eb8fc656122cf Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 10 Mar 2017 08:55:51 -0500 Subject: [PATCH] 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 --- .../config/MessageChannelConfigurerTests.java | 20 +++++++++++------- .../cloud/stream/binder/BinderHeaders.java | 21 +++++++++++++++---- .../binding/MessageConverterConfigurer.java | 10 +++++++-- 3 files changed, 38 insertions(+), 13 deletions(-) diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java index 2a16910c0..ac3da7a0a 100644 --- a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java +++ b/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) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java index f4f06c6d5..f9fdb71e8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderHeaders.java +++ b/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 diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java index 920b45fe0..6d1f724a1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java +++ b/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(); } } }