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 6f31a67..f5ec5c5 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 @@ -165,21 +165,23 @@ listenerMode:: The mode in which records are processed. If `record`, each `Message` will contain `byte[]` from a single `Record.data`. If `batch`, each `Message` will contain a `List` extracted from the consumed records. - If `rawRecords`, each `Message` will non-converted `List`. + When `useNativeDecoding = true` is used on the consumer together with the `listenerMode = batch`, there is no any out-of-the-box conversion happened and a result message contains a payload like `List`. + It's up to target application to convert those records manually. + Default: `record` checkpointMode:: - The mode in which checkpoints are updated. If `record`, checkpoints occur after each record is processed (but this option - is only effective if `listenerMode` is set to `record`). If `batch`, checkpoints occur after each batch of records is - processed. If `manual`, checkpoints occur on demand via the `Checkpointer` callback. + The mode in which checkpoints are updated. + If `record`, checkpoints occur after each record is processed (but this option is only effective if `listenerMode` is set to `record`). If `batch`, checkpoints occur after each batch of records is processed. + If `manual`, checkpoints occur on demand via the `Checkpointer` callback. + Default: `batch` recordsLimit:: - The maximum number of records to poll per `GetRecords` request. Must not be greater than `10000`. + The maximum number of records to poll per `GetRecords` request. + Must not be greater than `10000`. + Default: `10000` idleBetweenPolls:: - The sleep interval used in the main loop between shards polling cycles, in milliseconds. Must not be less than `250` + The sleep interval used in the main loop between shards polling cycles, in milliseconds. Must not be less than `250`. + Default: `1000` consumerBackoff:: 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 a24ff8f..e104cc0 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 @@ -215,24 +215,14 @@ public class KinesisMessageChannelBinder extends ? kinesisShardOffset : KinesisShardOffset.trimHorizon()); - // Defer byte[] conversion to the InboundContentTypeConvertingInterceptor - adapter.setConverter(bytes -> bytes); + adapter.setListenerMode(kinesisConsumerProperties.getListenerMode()); - switch (kinesisConsumerProperties.getListenerMode()) { - - case record: - adapter.setListenerMode(ListenerMode.record); - break; - - case batch: - adapter.setListenerMode(ListenerMode.batch); - break; - - case rawRecords: - adapter.setListenerMode(ListenerMode.batch); + if (properties.isUseNativeDecoding()) { adapter.setConverter(null); - break; - + } + else { + // Defer byte[] conversion to the InboundContentTypeConvertingInterceptor + adapter.setConverter(bytes -> bytes); } adapter.setCheckpointMode(kinesisConsumerProperties.getCheckpointMode()); diff --git a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java index 731a2cb..8c92bc4 100644 --- a/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java +++ b/spring-cloud-stream-binder-kinesis/src/main/java/org/springframework/cloud/stream/binder/kinesis/properties/KinesisConsumerProperties.java @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.binder.kinesis.properties; import org.springframework.integration.aws.inbound.kinesis.CheckpointMode; +import org.springframework.integration.aws.inbound.kinesis.ListenerMode; /** * @@ -29,7 +30,7 @@ public class KinesisConsumerProperties { private int startTimeout = 60000; - private KinesisListenerMode listenerMode = KinesisListenerMode.record; + private ListenerMode listenerMode = ListenerMode.record; private CheckpointMode checkpointMode = CheckpointMode.batch; @@ -49,11 +50,11 @@ public class KinesisConsumerProperties { this.startTimeout = startTimeout; } - public KinesisListenerMode getListenerMode() { + public ListenerMode getListenerMode() { return this.listenerMode; } - public void setListenerMode(KinesisListenerMode listenerMode) { + public void setListenerMode(ListenerMode listenerMode) { this.listenerMode = listenerMode; } @@ -97,27 +98,4 @@ public class KinesisConsumerProperties { this.shardIteratorType = shardIteratorType; } - /** - * @see org.springframework.integration.aws.inbound.kinesis.ListenerMode - */ - public enum KinesisListenerMode { - - /** - * Each {@code Message} will be converted from a single {@code Record}. - */ - record, - - /** - * Each {@code Message} will contain {@code List} from {@code Record} list if not - * empty. - */ - batch, - - /** - * Each {@code Message} will contain {@code List} if not empty. - */ - rawRecords - - } - } diff --git a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderProcessorTests.java b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderProcessorTests.java index 75ff4c9..fd62673 100644 --- a/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderProcessorTests.java +++ b/spring-cloud-stream-binder-kinesis/src/test/java/org/springframework/cloud/stream/binder/kinesis/KinesisBinderProcessorTests.java @@ -71,7 +71,7 @@ import static org.assertj.core.api.Assertions.assertThat; "spring.cloud.stream.kinesis.bindings.input.consumer.idleBetweenPolls = 1", "spring.cloud.stream.kinesis.binder.headers = foo", "spring.cloud.stream.kinesis.binder.checkpoint.table = checkpointTable", - "spring.cloud.stream.kinesis.binder.locks.table = fakeTable" }) + "spring.cloud.stream.kinesis.binder.locks.table = lockTable" }) @DirtiesContext public class KinesisBinderProcessorTests { @@ -145,11 +145,6 @@ public class KinesisBinderProcessorTests { return localKinesisResource.getResource(); } - @Bean - public LockRegistry lockRegistry() { - return new DefaultLockRegistry(); - } - @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) public String transform(Message message) { String payload = new String((byte[]) message.getPayload()); 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 fd4e850..913482b 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 @@ -61,6 +61,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.aws.inbound.kinesis.KinesisShardOffset; +import org.springframework.integration.aws.inbound.kinesis.ListenerMode; import org.springframework.integration.aws.support.AwsRequestFailureException; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.NullChannel; @@ -443,7 +444,8 @@ public class KinesisBinderTests } ExtendedConsumerProperties consumerProperties = createConsumerProperties(); - consumerProperties.getExtension().setListenerMode(KinesisConsumerProperties.KinesisListenerMode.rawRecords); + consumerProperties.getExtension().setListenerMode(ListenerMode.batch); + consumerProperties.setUseNativeDecoding(true); QueueChannel input = new QueueChannel(); Binding inputBinding = binder.bindConsumer("testBatchListener", null, input,