diff --git a/pom.xml b/pom.xml index 481e4e6..1b667b1 100644 --- a/pom.xml +++ b/pom.xml @@ -15,9 +15,9 @@ pom - 2.0.0.RELEASE - 2.0.0.M4 - 2.0.0.M2 + 2.0.1.BUILD-SNAPSHOT + 2.0.0.RC2 + 2.0.0.BUILD-SNAPSHOT 1.8 diff --git a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc index 38f06c0..d103eae 100644 --- a/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kinesis-docs/src/main/asciidoc/overview.adoc @@ -36,10 +36,12 @@ AWS Kinesis uses the partition key as input to a hash function that maps the par Specifically, an MD5 hash function is used to map partition keys to 128-bit integer values and to map associated data records to shards. As a result of this hashing mechanism, all data records with the same partition key map to the same shard within the stream. But at the same time we can't select target shard to send explicitly. -Although calculation the hash manually (and use `explicitHashKeyExpression` for producer, respectively), we may track the target shard by inclusion into its `HashKeyRange`. +Although calculating the hash manually (and use `explicitHashKeyExpression` for producer, respectively), we may track the target shard by inclusion into its `HashKeyRange`. By default partition key is a result of the `Object.hash()` from the message `payload`. +The Spring Cloud Stream partition handling logic is excluded in case of AWS Kinesis Binder since it is out of use and the provided `producer.partitionKeyExpression` is propagated to the `KinesisMessageHandler` directly. + On the consumer side the `instanceCount` and `instanceIndex` are used to distribute shards between consumers in group evenly. == Configuration Options diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java index 8f14117..f9f27b7 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/KinesisMessageChannelBinder.java @@ -38,13 +38,16 @@ import org.springframework.cloud.stream.binder.kinesis.properties.KinesisExtende import org.springframework.cloud.stream.binder.kinesis.properties.KinesisProducerProperties; import org.springframework.cloud.stream.binder.kinesis.provisioning.KinesisConsumerDestination; import org.springframework.cloud.stream.binder.kinesis.provisioning.KinesisStreamProvisioner; +import org.springframework.cloud.stream.binding.MessageConverterConfigurer; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.expression.Expression; import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter; import org.springframework.integration.aws.inbound.kinesis.KinesisMessageHeaderErrorMessageStrategy; import org.springframework.integration.aws.inbound.kinesis.KinesisShardOffset; import org.springframework.integration.aws.inbound.kinesis.ListenerMode; import org.springframework.integration.aws.outbound.KinesisMessageHandler; +import org.springframework.integration.channel.ChannelInterceptorAware; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.metadata.ConcurrentMetadataStore; @@ -52,6 +55,7 @@ import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; @@ -105,9 +109,6 @@ public class KinesisMessageChannelBinder extends return this.extendedBindingProperties.getExtendedProducerProperties(channelName); } - // below are the main methods to implement - these will create the message - // handlers used by the application - // to put and consume messages @Override protected MessageHandler createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties producerProperties, MessageChannel errorChannel) { @@ -116,14 +117,14 @@ public class KinesisMessageChannelBinder extends kinesisMessageHandler.setSync(producerProperties.getExtension().isSync()); kinesisMessageHandler.setSendTimeout(producerProperties.getExtension().getSendTimeout()); kinesisMessageHandler.setStream(destination.getName()); - if (producerProperties.isPartitioned()) { + Expression partitionKeyExpression = producerProperties.getPartitionKeyExpression(); + if (partitionKeyExpression != null) { kinesisMessageHandler - .setPartitionKeyExpressionString( - "'partitionKey-' + headers['" + BinderHeaders.PARTITION_HEADER + "']"); + .setPartitionKeyExpression(partitionKeyExpression); } else { - kinesisMessageHandler - .setPartitionKeyExpression(new FunctionExpression>(m -> m.getPayload().hashCode())); + kinesisMessageHandler.setPartitionKeyExpression( + new FunctionExpression>(m -> m.getPayload().hashCode())); } kinesisMessageHandler.setFailureChannel(errorChannel); kinesisMessageHandler.setBeanFactory(getBeanFactory()); @@ -131,6 +132,22 @@ public class KinesisMessageChannelBinder extends return kinesisMessageHandler; } + @Override + protected void postProcessOutputChannel(MessageChannel outputChannel, + ExtendedProducerProperties producerProperties) { + + if (outputChannel instanceof ChannelInterceptorAware) { + ChannelInterceptorAware channelInterceptorAware = (ChannelInterceptorAware) outputChannel; + List channelInterceptors = channelInterceptorAware.getChannelInterceptors(); + for (ChannelInterceptor channelInterceptor : channelInterceptors) { + if (channelInterceptor instanceof MessageConverterConfigurer.PartitioningInterceptor) { + channelInterceptorAware.removeInterceptor(channelInterceptor); + break; + } + } + } + } + @Override protected MessageProducer createConsumerEndpoint(ConsumerDestination destination, String group, ExtendedConsumerProperties properties) { @@ -179,7 +196,7 @@ public class KinesisMessageChannelBinder extends } else { adapter = new KinesisMessageDrivenChannelAdapter(this.amazonKinesis, - shardOffsets.toArray(new KinesisShardOffset[shardOffsets.size()])); + shardOffsets.toArray(new KinesisShardOffset[0])); } boolean anonymous = !StringUtils.hasText(group); @@ -196,18 +213,18 @@ public class KinesisMessageChannelBinder extends switch (kinesisConsumerProperties.getListenerMode()) { - case record: - adapter.setListenerMode(ListenerMode.record); - break; + case record: + adapter.setListenerMode(ListenerMode.record); + break; - case batch: - adapter.setListenerMode(ListenerMode.batch); - break; + case batch: + adapter.setListenerMode(ListenerMode.batch); + break; - case rawRecords: - adapter.setListenerMode(ListenerMode.batch); - adapter.setConverter(null); - break; + case rawRecords: + adapter.setListenerMode(ListenerMode.batch); + adapter.setConverter(null); + break; } diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java index 41a8fbe..70e6c74 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderTests.java @@ -317,8 +317,7 @@ public class KinesisBinderTests consumerProperties); ExtendedProducerProperties producerProperties = createProducerProperties(); - producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("payload")); - producerProperties.setPartitionSelectorExpression(spelExpressionParser.parseExpression("hashCode()")); + producerProperties.setPartitionKeyExpression(spelExpressionParser.parseExpression("headers.partitionKey")); producerProperties.setPartitionCount(3); DirectChannel output = createBindableChannel("test.output", @@ -333,12 +332,14 @@ public class KinesisBinderTests } Message message2 = MessageBuilder.withPayload(2) + .setHeader("partitionKey", 2) .setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, "foo") .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, 42) - .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43).build(); + .setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, 43) + .build(); output.send(message2); - output.send(new GenericMessage<>(1)); - output.send(new GenericMessage<>(0)); + output.send(MessageBuilder.withPayload(1).setHeader("partitionKey", 1).build()); + output.send(MessageBuilder.withPayload(0).setHeader("partitionKey", 0).build()); assertThat(receiveLatch.await(20, TimeUnit.SECONDS)).isTrue();