diff --git a/README.md b/README.md index ac73851..e589a7b 100644 --- a/README.md +++ b/README.md @@ -560,10 +560,16 @@ When records are consumed, they are filtered by the last stored `lastCheckpoint` ### Outbound Channel Adapter -The `KinesisMessageHandler` is a `AbstractMessageHandler` to perform put record to the Kinesis stream. +The `KinesisMessageHandler` is an `AbstractMessageHandler` to perform put record to the Kinesis stream. The stream, partition key (or explicit hash key) and sequence number can be determined against request message via evaluation provided expressions or can be specified statically. They also can specified as `AwsHeaders.STREAM`, `AwsHeaders.PARTITION_KEY` and `AwsHeaders.SEQUENCE_NUMBER` respectively. +The `KinesisMessageHandler` can be configured with channels for sending a `Message` on send success (in which the payload is either +the `data` from the `PutRecordRequest` or the full `PutRecordsRequest`), or an `ErrorMessage` on send failure +(in which the payload is `AwsRequestFailureException`). A `com.amazonaws.handlers.AsyncHandler` can also be +provided to the `KinesisMessageHandler` for custom handling after sending record(s) to the stream, but doing so +precludes the usage of such channels. + The `payload` of request message can be: - `PutRecordsRequest` to perform `AmazonKinesisAsync.putRecordsAsync` @@ -580,9 +586,13 @@ public static class MyConfiguration { @Bean @ServiceActivator(inputChannel = "kinesisSendChannel") - public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis) { + public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis, + MessageChannel channel, + MessageChannel errorChannel) { KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis); kinesisMessageHandler.setPartitionKey("1"); + kinesisMessageHandler.setOutputChannel(channel); + kinesisMessageHandler.setSendFailureChannel(errorChannel); return kinesisMessageHandler; } diff --git a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java index 764a91a..247e13e 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java @@ -28,14 +28,21 @@ import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.aws.support.AwsRequestFailureException; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.handler.AbstractMessageProducingHandler; +import org.springframework.integration.support.DefaultErrorMessageStrategy; +import org.springframework.integration.support.ErrorMessageStrategy; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import com.amazonaws.AmazonWebServiceRequest; +import com.amazonaws.AmazonWebServiceResult; import com.amazonaws.handlers.AsyncHandler; import com.amazonaws.services.kinesis.AmazonKinesisAsync; import com.amazonaws.services.kinesis.model.PutRecordRequest; @@ -47,12 +54,13 @@ import com.amazonaws.services.kinesis.model.PutRecordsResult; * The {@link AbstractMessageHandler} implementation for the Amazon Kinesis {@code putRecord(s)}. * * @author Artem Bilan + * @author Jacob Severson * @since 1.1 * * @see AmazonKinesisAsync#putRecord(PutRecordRequest) * @see com.amazonaws.handlers.AsyncHandler */ -public class KinesisMessageHandler extends AbstractMessageHandler { +public class KinesisMessageHandler extends AbstractMessageProducingHandler { private static final long DEFAULT_SEND_TIMEOUT = 10000; @@ -76,6 +84,12 @@ public class KinesisMessageHandler extends AbstractMessageHandler { private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT); + private MessageChannel sendFailureChannel; + + private String sendFailureChannelName; + + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); + public KinesisMessageHandler(AmazonKinesisAsync amazonKinesis) { Assert.notNull(amazonKinesis, "'amazonKinesis' must not be null."); this.amazonKinesis = amazonKinesis; @@ -155,6 +169,41 @@ public class KinesisMessageHandler extends AbstractMessageHandler { this.sendTimeoutExpression = sendTimeoutExpression; } + /** + * Set the failure channel. After a send failure, an {@link ErrorMessage} will be sent + * to this channel with a payload of a {@link AwsRequestFailureException} with the + * failed message and cause. + * @param sendFailureChannel the failure channel. + * @since 1.1.0 + */ + public void setSendFailureChannel(MessageChannel sendFailureChannel) { + this.sendFailureChannel = sendFailureChannel; + } + + protected MessageChannel getSendFailureChannel() { + if (this.sendFailureChannel != null) { + return this.sendFailureChannel; + + } + else if (this.sendFailureChannelName != null) { + this.sendFailureChannel = getChannelResolver().resolveDestination(this.sendFailureChannelName); + return this.sendFailureChannel; + } + + return null; + } + + /** + * Set the failure channel name. After a send failure, an {@link ErrorMessage} will be + * sent to this channel name with a payload of a {@link AwsRequestFailureException} + * with the failed message and cause. + * @param sendFailureChannelName the failure channel name. + * @since 1.1.0 + */ + public void setSendFailureChannelName(String sendFailureChannelName) { + this.sendFailureChannelName = sendFailureChannelName; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -163,19 +212,22 @@ public class KinesisMessageHandler extends AbstractMessageHandler { @Override @SuppressWarnings("unchecked") - protected void handleMessageInternal(Message message) throws Exception { - Future resultFuture = null; + protected void handleMessageInternal(final Message message) throws Exception { + Future resultFuture; + if (message.getPayload() instanceof PutRecordsRequest) { + resultFuture = this.amazonKinesis.putRecordsAsync((PutRecordsRequest) message.getPayload(), - (AsyncHandler) this.asyncHandler); + (AsyncHandler) getAsyncHandler(message, + (PutRecordsRequest) message.getPayload())); } else { - PutRecordRequest putRecordRequest = (message.getPayload() instanceof PutRecordRequest) + final PutRecordRequest putRecordRequest = (message.getPayload() instanceof PutRecordRequest) ? (PutRecordRequest) message.getPayload() : buildPutRecordRequest(message); resultFuture = this.amazonKinesis.putRecordAsync(putRecordRequest, - (AsyncHandler) this.asyncHandler); + (AsyncHandler) getAsyncHandler(message, putRecordRequest)); } if (this.sync) { @@ -245,4 +297,44 @@ public class KinesisMessageHandler extends AbstractMessageHandler { .withData(data); } + @SuppressWarnings("rawtypes") + private AsyncHandler getAsyncHandler(final Message message, + final AmazonWebServiceRequest request) { + if (this.asyncHandler != null) { + return this.asyncHandler; + } + else { + return new AsyncHandler() { + + @Override + public void onError(Exception ex) { + if (getSendFailureChannel() != null) { + KinesisMessageHandler.this.messagingTemplate.send(getSendFailureChannel(), + KinesisMessageHandler.this.errorMessageStrategy.buildErrorMessage( + new AwsRequestFailureException(message, request, ex), null)); + } + } + + @Override + public void onSuccess(AmazonWebServiceRequest request, AmazonWebServiceResult result) { + Message resultMessage; + + if (result instanceof PutRecordResult) { + resultMessage = getMessageBuilderFactory().fromMessage(message) + .setHeader(AwsHeaders.SHARD, ((PutRecordResult) result).getShardId()) + .setHeader(AwsHeaders.SEQUENCE_NUMBER, ((PutRecordResult) result).getSequenceNumber()) + .build(); + } + else { + resultMessage = getMessageBuilderFactory().fromMessage(message).build(); + } + + if (getOutputChannel() != null) { + KinesisMessageHandler.this.messagingTemplate.send(getOutputChannel(), resultMessage); + } + } + }; + } + } + } diff --git a/src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java b/src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java new file mode 100644 index 0000000..a9f9cc8 --- /dev/null +++ b/src/main/java/org/springframework/integration/aws/support/AwsRequestFailureException.java @@ -0,0 +1,50 @@ +/* + * Copyright 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. + * 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.support; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; + +import com.amazonaws.AmazonWebServiceRequest; + +/** + * An exception that is the payload of an {@code ErrorMessage} when a send fails. + * + * @author Jacob Severson + * @since 1.1.0 + */ +public class AwsRequestFailureException extends MessagingException { + + private static final long serialVersionUID = 1L; + + private final AmazonWebServiceRequest request; + + public AwsRequestFailureException(Message message, AmazonWebServiceRequest request, Throwable cause) { + super(message, cause); + this.request = request; + } + + public AmazonWebServiceRequest getRequest() { + return this.request; + } + + @Override + public String toString() { + return super.toString() + " [record=" + this.request + "]"; + } + +} diff --git a/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java b/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java new file mode 100644 index 0000000..ca4afee --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/outbound/KinesisProducingMessageHandlerTests.java @@ -0,0 +1,242 @@ +/* + * Copyright 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. + * 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.outbound; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.mock; + +import java.nio.ByteBuffer; +import java.util.concurrent.Future; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.convert.converter.Converter; +import org.springframework.core.serializer.support.SerializingConverter; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.aws.support.AwsRequestFailureException; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +import com.amazonaws.handlers.AsyncHandler; +import com.amazonaws.services.kinesis.AmazonKinesisAsync; +import com.amazonaws.services.kinesis.model.PutRecordRequest; +import com.amazonaws.services.kinesis.model.PutRecordResult; +import com.amazonaws.services.kinesis.model.PutRecordsRequest; +import com.amazonaws.services.kinesis.model.PutRecordsRequestEntry; +import com.amazonaws.services.kinesis.model.PutRecordsResult; + +/** + * @author Jacob Severson + * @since 1.1.0 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class KinesisProducingMessageHandlerTests { + + @Autowired + protected MessageChannel kinesisSendChannel; + + @Autowired + protected KinesisMessageHandler kinesisMessageHandler; + + @Autowired + protected PollableChannel errorChannel; + + @Autowired + protected PollableChannel successChannel; + + @Test + @SuppressWarnings("unchecked") + public void testKinesisMessageHandler() { + Message message = MessageBuilder.withPayload("message").build(); + try { + this.kinesisSendChannel.send(message); + } + catch (Exception e) { + assertThat(e).isInstanceOf(MessageHandlingException.class); + assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(e.getMessage()).contains("'stream' must not be null for sending a Kinesis record"); + } + + this.kinesisMessageHandler.setStream("foo"); + try { + this.kinesisSendChannel.send(message); + } + catch (Exception e) { + assertThat(e).isInstanceOf(MessageHandlingException.class); + assertThat(e.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(e.getMessage()).contains("'partitionKey' must not be null for sending a Kinesis record"); + } + + message = MessageBuilder.fromMessage(message) + .setHeader(AwsHeaders.PARTITION_KEY, "fooKey") + .setHeader(AwsHeaders.SEQUENCE_NUMBER, "10") + .build(); + + this.kinesisSendChannel.send(message); + + Message success = this.successChannel.receive(10000); + assertThat(success.getHeaders().get(AwsHeaders.PARTITION_KEY)).isEqualTo("fooKey"); + assertThat(success.getHeaders().get(AwsHeaders.SEQUENCE_NUMBER)).isEqualTo("10"); + assertThat(success.getPayload()).isEqualTo("message"); + + message = MessageBuilder.fromMessage(message) + .setHeader(AwsHeaders.PARTITION_KEY, "fooKey") + .setHeader(AwsHeaders.SEQUENCE_NUMBER, "10") + .build(); + + this.kinesisSendChannel.send(message); + + Message failed = this.errorChannel.receive(10000); + AwsRequestFailureException putRecordFailure = (AwsRequestFailureException) failed.getPayload(); + assertThat(putRecordFailure.getCause().getMessage()).isEqualTo("putRecordRequestEx"); + assertThat(((PutRecordRequest) putRecordFailure.getRequest()).getStreamName()).isEqualTo("foo"); + assertThat(((PutRecordRequest) putRecordFailure.getRequest()).getPartitionKey()).isEqualTo("fooKey"); + assertThat(((PutRecordRequest) putRecordFailure.getRequest()).getSequenceNumberForOrdering()).isEqualTo("10"); + assertThat(((PutRecordRequest) putRecordFailure.getRequest()).getExplicitHashKey()).isNull(); + assertThat(((PutRecordRequest) putRecordFailure.getRequest()) + .getData()).isEqualTo(ByteBuffer.wrap("message".getBytes())); + + message = new GenericMessage<>(new PutRecordsRequest() + .withStreamName("myStream") + .withRecords(new PutRecordsRequestEntry() + .withData(ByteBuffer.wrap("test".getBytes())) + .withPartitionKey("testKey"))); + + this.kinesisSendChannel.send(message); + + success = this.successChannel.receive(10000); + assertThat(((PutRecordsRequest) success.getPayload()).getRecords()) + .containsExactlyInAnyOrder(new PutRecordsRequestEntry() + .withData(ByteBuffer.wrap("test".getBytes())) + .withPartitionKey("testKey")); + + message = new GenericMessage<>(new PutRecordsRequest() + .withStreamName("myStream") + .withRecords(new PutRecordsRequestEntry() + .withData(ByteBuffer.wrap("test".getBytes())) + .withPartitionKey("testKey"))); + + this.kinesisSendChannel.send(message); + + failed = this.errorChannel.receive(10000); + AwsRequestFailureException putRecordsFailure = (AwsRequestFailureException) failed.getPayload(); + assertThat(putRecordsFailure.getCause().getMessage()).isEqualTo("putRecordsRequestEx"); + assertThat(((PutRecordsRequest) putRecordsFailure.getRequest()).getStreamName()).isEqualTo("myStream"); + assertThat(((PutRecordsRequest) putRecordsFailure.getRequest()).getRecords()) + .containsExactlyInAnyOrder(new PutRecordsRequestEntry() + .withData(ByteBuffer.wrap("test".getBytes())) + .withPartitionKey("testKey")); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + @SuppressWarnings("unchecked") + public AmazonKinesisAsync amazonKinesis() { + AmazonKinesisAsync mock = mock(AmazonKinesisAsync.class); + + given(mock.putRecordAsync(any(PutRecordRequest.class), any(AsyncHandler.class))) + .willAnswer(invocation -> { + PutRecordRequest request = invocation.getArgumentAt(0, PutRecordRequest.class); + AsyncHandler handler = + invocation.getArgumentAt(1, AsyncHandler.class); + PutRecordResult result = new PutRecordResult() + .withSequenceNumber(request.getSequenceNumberForOrdering()) + .withShardId("shardId-1"); + handler.onSuccess(new PutRecordRequest(), result); + return mock(Future.class); + }) + .willAnswer(invocation -> { + AsyncHandler handler = invocation.getArgumentAt(1, AsyncHandler.class); + handler.onError(new RuntimeException("putRecordRequestEx")); + return mock(Future.class); + }); + + + given(mock.putRecordsAsync(any(PutRecordsRequest.class), any(AsyncHandler.class))) + .willAnswer(invocation -> { + AsyncHandler handler = + invocation.getArgumentAt(1, AsyncHandler.class); + handler.onSuccess(new PutRecordsRequest(), new PutRecordsResult()); + return mock(Future.class); + }) + .willAnswer(invocation -> { + AsyncHandler handler = invocation.getArgumentAt(1, AsyncHandler.class); + handler.onError(new RuntimeException("putRecordsRequestEx")); + return mock(Future.class); + }); + + return mock; + } + + @Bean + public PollableChannel errorChannel() { + return new QueueChannel(); + } + + @Bean + public PollableChannel successChannel() { + return new QueueChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "kinesisSendChannel") + public MessageHandler kinesisMessageHandler() { + KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis()); + kinesisMessageHandler.setSync(true); + kinesisMessageHandler.setOutputChannel(successChannel()); + kinesisMessageHandler.setSendFailureChannel(errorChannel()); + kinesisMessageHandler.setConverter(new Converter() { + + private SerializingConverter serializingConverter = new SerializingConverter(); + + @Override + public byte[] convert(Object source) { + if (source instanceof String) { + return ((String) source).getBytes(); + } + else { + return this.serializingConverter.convert(source); + } + } + + }); + return kinesisMessageHandler; + } + + } + +}