From 7d283fea87e688d2f7c7ac871287e84a78ade957 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 25 Jan 2018 17:27:44 -0500 Subject: [PATCH] Handle errors in KinesisMessageDrivenChAdapter Relates spring-cloud-stream-binder-aws-kinesis/#34 * Introduce `KinesisMessageHeaderErrorMessageStrategy` to populate `AwsHeaders.RAW_RECORD` to the `ErrorMessage` headers * Improve `KinesisMessageDrivenChannelAdapter` for the `AttributeAccessor`. Also `try...finally` the `processRecords()` to perform important tasks independently of the `processRecords()` result --- .../KinesisMessageDrivenChannelAdapter.java | 307 +++++++++--------- ...esisMessageHeaderErrorMessageStrategy.java | 54 +++ .../integration/aws/support/AwsHeaders.java | 5 + .../aws/kinesis/KinesisIntegrationTests.java | 28 +- 4 files changed, 247 insertions(+), 147 deletions(-) create mode 100644 src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageHeaderErrorMessageStrategy.java diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java index edbd1fa..cb35ca2 100644 --- a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java @@ -39,6 +39,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import org.springframework.beans.factory.DisposableBean; +import org.springframework.core.AttributeAccessor; import org.springframework.core.convert.converter.Converter; import org.springframework.core.serializer.support.DeserializingConverter; import org.springframework.integration.aws.support.AwsHeaders; @@ -46,9 +47,12 @@ import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.metadata.MetadataStore; import org.springframework.integration.metadata.SimpleMetadataStore; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.messaging.Message; import org.springframework.scheduling.SchedulingAwareRunnable; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; @@ -81,6 +85,8 @@ import com.amazonaws.services.kinesis.model.StreamStatus; @IntegrationManagedResource public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport implements DisposableBean { + private static final ThreadLocal attributesHolder = new ThreadLocal<>(); + private final AmazonKinesis amazonKinesis; private final String[] streams; @@ -144,6 +150,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i public KinesisMessageDrivenChannelAdapter(AmazonKinesis amazonKinesis, KinesisShardOffset... shardOffsets) { + Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null."); Assert.notEmpty(shardOffsets, "'shardOffsets' must not be null."); Assert.noNullElements(shardOffsets, "'shardOffsets' must not contain null elements."); @@ -274,7 +281,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } @Override - public void destroy() throws Exception { + public void destroy() { if (!this.consumerExecutorExplicitlySet) { ((ExecutorService) this.consumerExecutor).shutdown(); } @@ -444,113 +451,102 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } private void populateShardsForStream(final String stream, final CountDownLatch shardsGatherLatch) { - this.dispatcherExecutor.execute(new Runnable() { + this.dispatcherExecutor.execute(() -> { + try { + int describeStreamRetries = 0; + List shardsToConsume = new ArrayList<>(); - @Override - public void run() { - try { - int describeStreamRetries = 0; - List shardsToConsume = new ArrayList<>(); + String exclusiveStartShardId = null; + while (true) { + DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest() + .withStreamName(stream) + .withExclusiveStartShardId(exclusiveStartShardId); - String exclusiveStartShardId = null; - while (true) { - DescribeStreamRequest describeStreamRequest = new DescribeStreamRequest() - .withStreamName(stream) - .withExclusiveStartShardId(exclusiveStartShardId); + DescribeStreamResult describeStreamResult = null; + // Call DescribeStream, with backoff and retries (if we get LimitExceededException). + try { + describeStreamResult = this.amazonKinesis.describeStream(describeStreamRequest); + } + catch (LimitExceededException e) { + logger.info("Got LimitExceededException when describing stream [" + stream + "]. " + + "Backing off for [" + this.describeStreamBackoff + "] millis."); + } - DescribeStreamResult describeStreamResult = null; - // Call DescribeStream, with backoff and retries (if we get LimitExceededException). + if (describeStreamResult == null || + !StreamStatus.ACTIVE.toString() + .equals(describeStreamResult.getStreamDescription().getStreamStatus())) { + if (describeStreamRetries++ > this.describeStreamRetries) { + ResourceNotFoundException resourceNotFoundException = + new ResourceNotFoundException("The stream [" + stream + + "] isn't ACTIVE or doesn't exist."); + resourceNotFoundException.setServiceName("Kinesis"); + throw resourceNotFoundException; + } try { - describeStreamResult = KinesisMessageDrivenChannelAdapter.this.amazonKinesis - .describeStream(describeStreamRequest); + Thread.sleep(this.describeStreamBackoff); + continue; } - catch (LimitExceededException e) { - logger.info("Got LimitExceededException when describing stream [" + stream + "]. " + - "Backing off for [" + - KinesisMessageDrivenChannelAdapter.this.describeStreamBackoff + "] millis."); + catch (InterruptedException e) { + Thread.interrupted(); + throw new IllegalStateException("The [describeStream] thread for the stream [" + + stream + "] has been interrupted.", e); } + } - if (describeStreamResult == null || - !StreamStatus.ACTIVE.toString() - .equals(describeStreamResult.getStreamDescription().getStreamStatus())) { - if (describeStreamRetries++ > - KinesisMessageDrivenChannelAdapter.this.describeStreamRetries) { - ResourceNotFoundException resourceNotFoundException = - new ResourceNotFoundException("The stream [" + stream + - "] isn't ACTIVE or doesn't exist."); - resourceNotFoundException.setServiceName("Kinesis"); - throw resourceNotFoundException; + List shards = describeStreamResult.getStreamDescription().getShards(); + for (Shard shard : shards) { + String endingSequenceNumber = shard.getSequenceNumberRange().getEndingSequenceNumber(); + if (endingSequenceNumber != null) { + String key = buildCheckpointKeyForShard(stream, shard.getShardId()); + String checkpoint = this.checkpointStore.get(key); + + boolean skipClosedShard = checkpoint != null && + new BigInteger(endingSequenceNumber) + .compareTo(new BigInteger(checkpoint)) <= 0; + + if (logger.isTraceEnabled()) { + logger.trace("The shard [" + shard + "] in stream [" + stream + + "] is closed CLOSED with endingSequenceNumber [" + endingSequenceNumber + + "].\nThe last processed checkpoint is [" + checkpoint + "]." + + (skipClosedShard ? "\nThe shard will be skipped." : "")); } - try { - Thread.sleep(KinesisMessageDrivenChannelAdapter.this.describeStreamBackoff); + + if (skipClosedShard) { + // Skip CLOSED shard which has been read before according a checkpoint continue; } - catch (InterruptedException e) { - Thread.interrupted(); - throw new IllegalStateException("The [describeStream] thread for the stream [" - + stream + "] has been interrupted.", e); - } } - List shards = describeStreamResult.getStreamDescription().getShards(); - for (Shard shard : shards) { - String endingSequenceNumber = shard.getSequenceNumberRange().getEndingSequenceNumber(); - if (endingSequenceNumber != null) { - String key = buildCheckpointKeyForShard(stream, shard.getShardId()); - String checkpoint = KinesisMessageDrivenChannelAdapter.this.checkpointStore.get(key); - - boolean skipClosedShard = checkpoint != null && - new BigInteger(endingSequenceNumber) - .compareTo(new BigInteger(checkpoint)) <= 0; - - if (logger.isTraceEnabled()) { - logger.trace("The shard [" + shard + "] in stream [" + stream + - "] is closed CLOSED with endingSequenceNumber [" + endingSequenceNumber + - "].\nThe last processed checkpoint is [" + checkpoint + "]." + - (skipClosedShard ? "\nThe shard will be skipped." : "")); - } - - if (skipClosedShard) { - // Skip CLOSED shard which has been read before according a checkpoint - continue; - } - } - - shardsToConsume.add(shard); - } - - if (describeStreamResult.getStreamDescription().getHasMoreShards()) { - exclusiveStartShardId = shards.get(shards.size() - 1).getShardId(); - } - else { - break; - } + shardsToConsume.add(shard); } - for (Shard shard : shardsToConsume) { - KinesisShardOffset shardOffset = - new KinesisShardOffset(KinesisMessageDrivenChannelAdapter.this.streamInitialSequence); - shardOffset.setShard(shard.getShardId()); - shardOffset.setStream(stream); - boolean addedOffset; - synchronized (KinesisMessageDrivenChannelAdapter.this.shardOffsets) { - addedOffset = KinesisMessageDrivenChannelAdapter.this.shardOffsets.add(shardOffset); - } - if (addedOffset - && shardsGatherLatch == null - && KinesisMessageDrivenChannelAdapter.this.active) { - populateConsumer(shardOffset); - } + if (describeStreamResult.getStreamDescription().getHasMoreShards()) { + exclusiveStartShardId = shards.get(shards.size() - 1).getShardId(); + } + else { + break; } } - finally { - if (shardsGatherLatch != null) { - shardsGatherLatch.countDown(); + + for (Shard shard : shardsToConsume) { + KinesisShardOffset shardOffset = new KinesisShardOffset(this.streamInitialSequence); + shardOffset.setShard(shard.getShardId()); + shardOffset.setStream(stream); + boolean addedOffset; + synchronized (this.shardOffsets) { + addedOffset = this.shardOffsets.add(shardOffset); + } + if (addedOffset && shardsGatherLatch == null && this.active) { + populateConsumer(shardOffset); } - KinesisMessageDrivenChannelAdapter.this.inResharding.remove(stream); } } - + finally { + if (shardsGatherLatch != null) { + shardsGatherLatch.countDown(); + } + this.inResharding.remove(stream); + } }); } @@ -596,7 +592,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } private String buildCheckpointKeyForShard(String stream, String shardId) { - return KinesisMessageDrivenChannelAdapter.this.consumerGroup + ":" + stream + ":" + shardId; + return this.consumerGroup + ":" + stream + ":" + shardId; } @Override @@ -617,6 +613,31 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i this.consumerInvokers.clear(); } + /** + * If there's an error channel, we create a new attributes holder here. + * Then set the attributes for use by the {@link ErrorMessageStrategy}. + * @param record the Kinesis record to use. + * @param message the Spring Messaging message to use. + */ + private void setAttributesIfNecessary(Object record, Message message) { + if (getErrorChannel() != null) { + AttributeAccessor attributes = ErrorMessageUtils.getAttributeAccessor(message, null); + attributesHolder.set(attributes); + attributes.setAttribute(AwsHeaders.RAW_RECORD, record); + } + } + + @Override + protected AttributeAccessor getErrorMessageAttributes(org.springframework.messaging.Message message) { + AttributeAccessor attributes = attributesHolder.get(); + if (attributes == null) { + return super.getErrorMessageAttributes(message); + } + else { + return attributes; + } + } + @Override public String toString() { return "KinesisMessageDrivenChannelAdapter{" + @@ -731,37 +752,31 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i case NEW: case EXPIRED: - this.task = new Runnable() { - - @Override - public void run() { - if (logger.isInfoEnabled() && ShardConsumer.this.state == ConsumerState.NEW) { - logger.info("The [" + ShardConsumer.this + "] has been started."); + this.task = () -> { + if (logger.isInfoEnabled() && this.state == ConsumerState.NEW) { + logger.info("The [" + this + "] has been started."); + } + if (this.shardOffset.isReset()) { + this.checkpointer.remove(); + } + else { + String checkpoint = this.checkpointer.getCheckpoint(); + if (checkpoint != null) { + this.shardOffset.setSequenceNumber(checkpoint); + this.shardOffset.setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER); } - if (ShardConsumer.this.shardOffset.isReset()) { - ShardConsumer.this.checkpointer.remove(); - } - else { - String checkpoint = ShardConsumer.this.checkpointer.getCheckpoint(); - if (checkpoint != null) { - ShardConsumer.this.shardOffset.setSequenceNumber(checkpoint); - ShardConsumer.this.shardOffset - .setIteratorType(ShardIteratorType.AFTER_SEQUENCE_NUMBER); - } - } - - GetShardIteratorRequest shardIteratorRequest = - ShardConsumer.this.shardOffset.toShardIteratorRequest(); - ShardConsumer.this.shardIterator = - KinesisMessageDrivenChannelAdapter.this.amazonKinesis - .getShardIterator(shardIteratorRequest) - .getShardIterator(); - if (ConsumerState.STOP != ShardConsumer.this.state) { - ShardConsumer.this.state = ConsumerState.CONSUME; - } - ShardConsumer.this.task = null; } + GetShardIteratorRequest shardIteratorRequest = this.shardOffset.toShardIteratorRequest(); + this.shardIterator = + KinesisMessageDrivenChannelAdapter.this.amazonKinesis + .getShardIterator(shardIteratorRequest) + .getShardIterator(); + if (ConsumerState.STOP != this.state) { + this.state = ConsumerState.CONSUME; + } + + this.task = null; }; break; @@ -779,14 +794,14 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i case STOP: if (this.shardIterator == null) { if (logger.isInfoEnabled()) { - logger.info("Stopping the [" + ShardConsumer.this + + logger.info("Stopping the [" + this + "] on the checkpoint [" + this.checkpointer.getCheckpoint() + "] because the shard has been CLOSED and exhausted."); } } else { if (logger.isInfoEnabled()) { - logger.info("Stopping the [" + ShardConsumer.this + "]."); + logger.info("Stopping the [" + this + "]."); } } this.task = null; @@ -805,36 +820,38 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } private Runnable processTask() { - return new Runnable() { + return () -> { + GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); + getRecordsRequest.setShardIterator(this.shardIterator); + getRecordsRequest.setLimit(KinesisMessageDrivenChannelAdapter.this.recordsLimit); - @Override - public void run() { - GetRecordsRequest getRecordsRequest = new GetRecordsRequest(); - getRecordsRequest.setShardIterator(ShardConsumer.this.shardIterator); - getRecordsRequest.setLimit(KinesisMessageDrivenChannelAdapter.this.recordsLimit); - - GetRecordsResult result = getRecords(getRecordsRequest); + GetRecordsResult result = getRecords(getRecordsRequest); + try { if (result != null) { List records = result.getRecords(); if (!records.isEmpty()) { processRecords(records); } + } + } + finally { + attributesHolder.remove(); + if (result != null) { + this.shardIterator = result.getNextShardIterator(); - ShardConsumer.this.shardIterator = result.getNextShardIterator(); - - if (ShardConsumer.this.shardIterator == null) { + if (this.shardIterator == null) { // Shard is closed: nothing to consume any more. // Resharding is possible. stop(); } - if (ConsumerState.STOP != ShardConsumer.this.state && records.isEmpty()) { + if (ConsumerState.STOP != this.state && result.getRecords().isEmpty()) { if (logger.isDebugEnabled()) { - logger.debug("No records for [" + ShardConsumer.this + + logger.debug("No records for [" + this + "] on sequenceNumber [" + - ShardConsumer.this.checkpointer.getLastCheckpointValue() + + this.checkpointer.getLastCheckpointValue() + "]. Suspend consuming for [" + KinesisMessageDrivenChannelAdapter.this.consumerBackoff + "] milliseconds."); } @@ -842,9 +859,8 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } } - ShardConsumer.this.task = null; + this.task = null; } - }; } @@ -909,7 +925,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i messageBuilder.setHeader(AwsHeaders.CHECKPOINTER, this.checkpointer); } - sendMessage(messageBuilder.build()); + performSend(messageBuilder, records); if (CheckpointMode.record.equals(KinesisMessageDrivenChannelAdapter.this.checkpointMode)) { this.checkpointer.checkpoint(record.getSequenceNumber()); @@ -927,7 +943,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i messageBuilder.setHeader(AwsHeaders.CHECKPOINTER, this.checkpointer); } - sendMessage(messageBuilder.build()); + performSend(messageBuilder, records); break; @@ -935,6 +951,12 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i } } + private void performSend(AbstractIntegrationMessageBuilder messageBuilder, Object rawRecord) { + Message messageToSend = messageBuilder.build(); + setAttributesIfNecessary(rawRecord, messageToSend); + sendMessage(messageToSend); + } + @Override public String toString() { return "ShardConsumer{" + @@ -957,14 +979,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport i private final Semaphore processBarrier = new Semaphore(0); - private final Runnable notifier = new Runnable() { - - @Override - public void run() { - notifyBarrier(); - } - - }; + private final Runnable notifier = this::notifyBarrier; ConsumerInvoker(Collection shardConsumers) { for (ShardConsumer shardConsumer : shardConsumers) { diff --git a/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageHeaderErrorMessageStrategy.java b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageHeaderErrorMessageStrategy.java new file mode 100644 index 0000000..ad10b69 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageHeaderErrorMessageStrategy.java @@ -0,0 +1,54 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aws.inbound.kinesis; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.core.AttributeAccessor; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.support.ErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageUtils; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.ErrorMessage; + +/** + * The {@link ErrorMessageStrategy} implementation to build an {@link ErrorMessage} + * with the {@link AwsHeaders#RAW_RECORD} header by the value from + * the the provided {@link AttributeAccessor}. + * + * @author Artem Bilan + * + * @since 2.0 + */ +public class KinesisMessageHeaderErrorMessageStrategy implements ErrorMessageStrategy { + + @Override + public ErrorMessage buildErrorMessage(Throwable throwable, AttributeAccessor context) { + Object inputMessage = + context == null ? null : context.getAttribute(ErrorMessageUtils.INPUT_MESSAGE_CONTEXT_KEY); + + Map headers = + context == null + ? new HashMap<>() + : Collections.singletonMap(AwsHeaders.RAW_RECORD, context.getAttribute(AwsHeaders.RAW_RECORD)); + + return new ErrorMessage(throwable, headers, inputMessage instanceof Message ? (Message) inputMessage : null); + } + +} diff --git a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java index 8f1849b..79bb865 100644 --- a/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java +++ b/src/main/java/org/springframework/integration/aws/support/AwsHeaders.java @@ -110,4 +110,9 @@ public abstract class AwsHeaders { */ public static final String SERVICE_RESULT = PREFIX + "serviceResult"; + /** + * The {@value RAW_RECORD} header represents received Kinesis record(s). + */ + public static final String RAW_RECORD = PREFIX + "rawRecord"; + } diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java index 1c9f760..bef79b9 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017 the original author or authors. + * Copyright 2017-2018 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. @@ -32,6 +32,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.aws.KinesisLocalRunning; import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter; +import org.springframework.integration.aws.inbound.kinesis.KinesisMessageHeaderErrorMessageStrategy; import org.springframework.integration.aws.outbound.KinesisMessageHandler; import org.springframework.integration.aws.support.AwsHeaders; import org.springframework.integration.channel.QueueChannel; @@ -64,6 +65,9 @@ public class KinesisIntegrationTests { @Autowired private PollableChannel kinesisReceiveChannel; + @Autowired + private PollableChannel errorChannel; + @BeforeClass public static void setup() { KINESIS_LOCAL_RUNNING.getKinesis().createStream(TEST_STREAM, 1); @@ -76,14 +80,27 @@ public class KinesisIntegrationTests { @Test public void testKinesisInboundOutbound() { + this.kinesisSendChannel.send( + MessageBuilder.withPayload("foo") + .setHeader(AwsHeaders.STREAM, TEST_STREAM) + .build()); + Date now = new Date(); this.kinesisSendChannel.send( MessageBuilder.withPayload(now) .setHeader(AwsHeaders.STREAM, TEST_STREAM) .build()); + Message receive = this.kinesisReceiveChannel.receive(10_000); assertThat(receive).isNotNull(); assertThat(receive.getPayload()).isEqualTo(now); + + Message errorMessage = this.errorChannel.receive(10_000); + assertThat(errorMessage).isNotNull(); + assertThat(errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD)).isNotNull(); + assertThat(((Exception) errorMessage.getPayload()).getMessage()) + .contains("Channel 'kinesisReceiveChannel' expected one of the following datataypes " + + "[class java.util.Date], but received [class java.lang.String]"); } @Configuration @@ -103,11 +120,20 @@ public class KinesisIntegrationTests { KinesisMessageDrivenChannelAdapter adapter = new KinesisMessageDrivenChannelAdapter(KINESIS_LOCAL_RUNNING.getKinesis(), TEST_STREAM); adapter.setOutputChannel(kinesisReceiveChannel()); + adapter.setErrorChannel(errorChannel()); + adapter.setErrorMessageStrategy(new KinesisMessageHeaderErrorMessageStrategy()); return adapter; } @Bean public PollableChannel kinesisReceiveChannel() { + QueueChannel queueChannel = new QueueChannel(); + queueChannel.setDatatypes(Date.class); + return queueChannel; + } + + @Bean + public PollableChannel errorChannel() { return new QueueChannel(); }