From c4a3fcdee2afde88f4127a339711168470778281 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 2 Aug 2017 20:01:13 -0400 Subject: [PATCH] GH-72: Add `KinesisLocalRunning` and tests Fixes spring-projects/spring-integration-aws#72 * Document Kinesis Channel Adapters * Fix some inconsistency in the `KinesisMessageHandler` * Add integration test against `KinesisLocalRunning` `@Rule` * Document testing against Kinesalite --- README.md | 111 +++++++++++++++++ .../aws/outbound/KinesisMessageHandler.java | 16 ++- ...Running.java => DynamoDbLocalRunning.java} | 13 +- .../integration/aws/KinesisLocalRunning.java | 113 +++++++++++++++++ .../aws/kinesis/KinesisIntegrationTests.java | 116 ++++++++++++++++++ .../metadata/DynamoDbMetadataStoreTests.java | 4 +- 6 files changed, 361 insertions(+), 12 deletions(-) rename src/test/java/org/springframework/integration/aws/{DynamoDbRunning.java => DynamoDbLocalRunning.java} (86%) create mode 100644 src/test/java/org/springframework/integration/aws/KinesisLocalRunning.java create mode 100644 src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java diff --git a/README.md b/README.md index abedab1..d3a3937 100644 --- a/README.md +++ b/README.md @@ -504,6 +504,114 @@ The implementation is based on a simple table with `KEY` and `VALUE` attributes, By default the `SpringIntegrationMetadataStore` table is used and it is created during `DynamoDbMetaDataStore` initialization if that doesn't exist yet. The `DynamoDbMetaDataStore` can be used for the `KinesisMessageDrivenChannelAdapter` as a cloud-based `cehckpointStore`. +For testing application with the `DynamoDbMetaDataStore` you can use [Dynalite][] NPM module. +What you need in your application is to configure DynamoDB client properly: + +````java +String url = "http://localhost:" + this.port; + +this.amazonDynamoDB = AmazonDynamoDBAsyncClientBuilder.standard() + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", ""))) + .withClientConfiguration( + new ClientConfiguration() + .withMaxErrorRetry(0) + .withConnectionTimeout(1000)) + .withEndpointConfiguration( + new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName())) + .build(); +```` + +Where you should specify the port on which you have ran the Dynalite service. +Also you can use for you testing purpose a copy of `org.springframework.integration.aws.DynamoDbLocalRunning` in the `/test` directory of this project. + +## Amazon Kinesis + +Amazon Kinesis is a platform for streaming data on AWS, making it easy to load and analyze streaming data, and also providing the ability for you to build custom streaming data applications for specialized needs. +The Spring Integration solution is fully based on the Standard `aws-java-sdk-kinesis` and doesn't use [Kinesis Client Library][] and isn't compatible with it. + +### Inbound Channel Adapter + +The `KinesisMessageDrivenChannelAdapter` is an extension of the `MessageProducerSupport` - event-driver channel adapter. + +See `KinesisMessageDrivenChannelAdapter` JavaDocs and its setters for more information how to use and how to configure it in the application for Kinesis streams ingestion. + +The Java Configuration is pretty simple: + +````java +@SpringBootApplication +public static class MyConfiguration { + + @Bean + public KinesisMessageDrivenChannelAdapter kinesisInboundChannelChannel(AmazonKinesis amazonKinesis) { + KinesisMessageDrivenChannelAdapter adapter = + new KinesisMessageDrivenChannelAdapter(amazonKinesis, "MY_STREAM"); + adapter.setOutputChannel(kinesisReceiveChannel()); + return adapter; + } +} +```` + +This channel adapter can be configured with the `DynamoDbMetaDataStore` mentioned above to track sequence checkpoints for shards in the cloud environment when we have several instances of our Kinesis application. +By default this adapter uses `DeserializingConverter` to convert `byte[]` from the `Record` data. +Can be specified as `null` with meaning no conversion and the target `Message` is sent with the `byte[]` payload. + +The consumer group is included to the metadata store `key`. +When records are consumed, they are filtered by the last stored `lastCheckpoint` under the key as `[CONSUMER_GROUP]:[STREAM]:[SHARD_ID]`. + +### Outbound Channel Adapter + +The `KinesisMessageHandler` is a `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 `payload` of request message can be: + +- `PutRecordsRequest` to perform `AmazonKinesisAsync.putRecordsAsync` +- `PutRecordRequest` to perform `AmazonKinesisAsync.putRecordAsync` +- `ByteBuffer` to represent a data of the `PutRecordRequest` +- `byte[]` which is wrapped to the `ByteBuffer` +- any other type which is converted to the `byte[]` by the provided `Converter`; the `SerializingConverter` is used by default. + +The Java Configuration for the message handler: + +````java +@SpringBootApplication +public static class MyConfiguration { + + @Bean + @ServiceActivator(inputChannel = "kinesisSendChannel") + public MessageHandler kinesisMessageHandler(AmazonKinesis amazonKinesis) { + KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(amazonKinesis); + kinesisMessageHandler.setPartitionKey("1"); + return kinesisMessageHandler; + } + +} +```` + +For testing application with the Kinesis Channel Adapters you can use [Kinesalite][] NPM module. +What you need in your application is to configure Kinesis client properly: + +````java +String url = "http://localhost:" + this.port; + +// See https://github.com/mhart/kinesalite#cbor-protocol-issues-with-the-java-sdk +System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true"); + +this.amazonKinesis = AmazonKinesisAsyncClientBuilder.standard() + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", ""))) + .withClientConfiguration( + new ClientConfiguration() + .withMaxErrorRetry(0) + .withConnectionTimeout(1000)) + .withEndpointConfiguration( + new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName())) + .build(); +```` + +Where you should specify the port on which you have ran the Kinesalite service. +Also you can use for you testing purpose a copy of `org.springframework.integration.aws.KinesisLocalRunning` in the `/test` directory of this project. + [Spring Cloud AWS]: https://github.com/spring-cloud/spring-cloud-aws [AWS SDK for Java]: http://aws.amazon.com/sdkforjava/ [Amazon Web Services]: http://aws.amazon.com/ @@ -513,3 +621,6 @@ The `DynamoDbMetaDataStore` can be used for the `KinesisMessageDrivenChannelAdap [Reference Manual]: http://docs.spring.io/spring-integration/reference/html/ftp.html [Pull requests]: http://help.github.com/send-pull-requests [contributor guidelines]: https://github.com/spring-projects/spring-integration/blob/master/CONTRIBUTING.adoc +[Dynalite]: [https://github.com/mhart/dynalite] +[Kinesis Client Library]: [https://github.com/awslabs/amazon-kinesis-client] +[Kinesalite]: [https://github.com/mhart/kinesalite] \ No newline at end of file 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 8bbc5f5..764a91a 100644 --- a/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java +++ b/src/main/java/org/springframework/integration/aws/outbound/KinesisMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * 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. @@ -131,7 +131,7 @@ public class KinesisMessageHandler extends AbstractMessageHandler { this.explicitHashKeyExpression = explicitHashKeyExpression; } - public void setSequenceNumberString(String sequenceNumberExpression) { + public void setSequenceNumberExpressionString(String sequenceNumberExpression) { setSequenceNumberExpression(EXPRESSION_PARSER.parseExpression(sequenceNumberExpression)); } @@ -143,6 +143,14 @@ public class KinesisMessageHandler extends AbstractMessageHandler { this.sync = sync; } + public void setSendTimeout(long sendTimeout) { + setSendTimeoutExpression(new ValueExpression<>(sendTimeout)); + } + + public void setSendTimeoutExpressionString(String sendTimeoutExpression) { + setSendTimeoutExpression(EXPRESSION_PARSER.parseExpression(sendTimeoutExpression)); + } + public void setSendTimeoutExpression(Expression sendTimeoutExpression) { this.sendTimeoutExpression = sendTimeoutExpression; } @@ -162,7 +170,6 @@ public class KinesisMessageHandler extends AbstractMessageHandler { (AsyncHandler) this.asyncHandler); } else { - PutRecordRequest putRecordRequest = (message.getPayload() instanceof PutRecordRequest) ? (PutRecordRequest) message.getPayload() : buildPutRecordRequest(message); @@ -227,8 +234,7 @@ public class KinesisMessageHandler extends AbstractMessageHandler { ? (byte[]) payload : this.converter.convert(payload); - data = ByteBuffer.wrap( - bytes); + data = ByteBuffer.wrap(bytes); } return new PutRecordRequest() diff --git a/src/test/java/org/springframework/integration/aws/DynamoDbRunning.java b/src/test/java/org/springframework/integration/aws/DynamoDbLocalRunning.java similarity index 86% rename from src/test/java/org/springframework/integration/aws/DynamoDbRunning.java rename to src/test/java/org/springframework/integration/aws/DynamoDbLocalRunning.java index f5c9e8b..3e24ef0 100644 --- a/src/test/java/org/springframework/integration/aws/DynamoDbRunning.java +++ b/src/test/java/org/springframework/integration/aws/DynamoDbLocalRunning.java @@ -38,13 +38,16 @@ import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder; /** + * The {@link TestWatcher} implementation for local Amazon DynamoDB service. + * See https://github.com/mhart/dynalite. + * * @author Artem Bilan * * @since 1.1 */ -public final class DynamoDbRunning extends TestWatcher { +public final class DynamoDbLocalRunning extends TestWatcher { - private static Log logger = LogFactory.getLog(DynamoDbRunning.class); + private static Log logger = LogFactory.getLog(DynamoDbLocalRunning.class); // Static so that we only test once on failure: speeds up test suite private static Map dynamoDbOnline = new HashMap<>(); @@ -53,7 +56,7 @@ public final class DynamoDbRunning extends TestWatcher { private AmazonDynamoDBAsync amazonDynamoDB; - private DynamoDbRunning(int port) { + private DynamoDbLocalRunning(int port) { this.port = port; dynamoDbOnline.put(port, true); } @@ -89,8 +92,8 @@ public final class DynamoDbRunning extends TestWatcher { } - public static DynamoDbRunning isRunning(int port) { - return new DynamoDbRunning(port); + public static DynamoDbLocalRunning isRunning(int port) { + return new DynamoDbLocalRunning(port); } } diff --git a/src/test/java/org/springframework/integration/aws/KinesisLocalRunning.java b/src/test/java/org/springframework/integration/aws/KinesisLocalRunning.java new file mode 100644 index 0000000..2c5839c --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/KinesisLocalRunning.java @@ -0,0 +1,113 @@ +/* + * 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; + +import static org.junit.Assume.assumeNoException; +import static org.junit.Assume.assumeTrue; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.junit.rules.TestWatcher; +import org.junit.runner.Description; +import org.junit.runners.model.Statement; + +import com.amazonaws.ClientConfiguration; +import com.amazonaws.SDKGlobalConfiguration; +import com.amazonaws.SdkClientException; +import com.amazonaws.auth.AWSStaticCredentialsProvider; +import com.amazonaws.auth.BasicAWSCredentials; +import com.amazonaws.client.builder.AwsClientBuilder; +import com.amazonaws.regions.Regions; +import com.amazonaws.services.kinesis.AmazonKinesisAsync; +import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder; + +/** + * The {@link TestWatcher} implementation for local Amazon Kinesis service. + * See https://github.com/mhart/kinesalite. + * + * @author Artem Bilan + * + * @since 1.1 + */ +public final class KinesisLocalRunning extends TestWatcher { + + private static Log logger = LogFactory.getLog(KinesisLocalRunning.class); + + // Static so that we only test once on failure: speeds up test suite + private static Map kinesisOnline = new HashMap<>(); + + private final int port; + + private AmazonKinesisAsync amazonKinesis; + + private KinesisLocalRunning(int port) { + this.port = port; + kinesisOnline.put(port, true); + } + + public AmazonKinesisAsync getKinesis() { + return this.amazonKinesis; + } + + @Override + public Statement apply(Statement base, Description description) { + assumeTrue(kinesisOnline.get(this.port)); + + String url = "http://localhost:" + this.port; + + // See https://github.com/mhart/kinesalite#cbor-protocol-issues-with-the-java-sdk + System.setProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY, "true"); + + this.amazonKinesis = AmazonKinesisAsyncClientBuilder.standard() + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials("", ""))) + .withClientConfiguration( + new ClientConfiguration() + .withMaxErrorRetry(0) + .withConnectionTimeout(1000)) + .withEndpointConfiguration( + new AwsClientBuilder.EndpointConfiguration(url, Regions.DEFAULT_REGION.getName())) + .build(); + + try { + this.amazonKinesis.listStreams(); + } + catch (SdkClientException e) { + logger.warn("Tests not running because no Kinesis on " + url, e); + assumeNoException(e); + } + + + return new Statement() { + public void evaluate() throws Throwable { + try { + base.evaluate(); + } finally { + System.clearProperty(SDKGlobalConfiguration.AWS_CBOR_DISABLE_SYSTEM_PROPERTY); + } + + } + }; + } + + public static KinesisLocalRunning isRunning(int port) { + return new KinesisLocalRunning(port); + } + +} diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java new file mode 100644 index 0000000..1c9f760 --- /dev/null +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -0,0 +1,116 @@ +/* + * 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.kinesis; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.Date; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.ClassRule; +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.integration.annotation.ServiceActivator; +import org.springframework.integration.aws.KinesisLocalRunning; +import org.springframework.integration.aws.inbound.kinesis.KinesisMessageDrivenChannelAdapter; +import org.springframework.integration.aws.outbound.KinesisMessageHandler; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Artem Bilan + * + * @since 1.1 + */ +@RunWith(SpringRunner.class) +@DirtiesContext +public class KinesisIntegrationTests { + + @ClassRule + public static final KinesisLocalRunning KINESIS_LOCAL_RUNNING = KinesisLocalRunning.isRunning(4567); + + private static final String TEST_STREAM = "TestStream"; + + @Autowired + private MessageChannel kinesisSendChannel; + + @Autowired + private PollableChannel kinesisReceiveChannel; + + @BeforeClass + public static void setup() { + KINESIS_LOCAL_RUNNING.getKinesis().createStream(TEST_STREAM, 1); + } + + @AfterClass + public static void tearDown() { + KINESIS_LOCAL_RUNNING.getKinesis().deleteStream(TEST_STREAM); + } + + @Test + public void testKinesisInboundOutbound() { + 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); + } + + @Configuration + @EnableIntegration + public static class TestConfiguration { + + @Bean + @ServiceActivator(inputChannel = "kinesisSendChannel") + public MessageHandler kinesisMessageHandler() { + KinesisMessageHandler kinesisMessageHandler = new KinesisMessageHandler(KINESIS_LOCAL_RUNNING.getKinesis()); + kinesisMessageHandler.setPartitionKey("1"); + return kinesisMessageHandler; + } + + @Bean + public KinesisMessageDrivenChannelAdapter kinesisInboundChannelChannel() { + KinesisMessageDrivenChannelAdapter adapter = + new KinesisMessageDrivenChannelAdapter(KINESIS_LOCAL_RUNNING.getKinesis(), TEST_STREAM); + adapter.setOutputChannel(kinesisReceiveChannel()); + return adapter; + } + + @Bean + public PollableChannel kinesisReceiveChannel() { + return new QueueChannel(); + } + + } + +} diff --git a/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java b/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java index 827ffeb..1c819a1 100644 --- a/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java +++ b/src/test/java/org/springframework/integration/aws/metadata/DynamoDbMetadataStoreTests.java @@ -26,7 +26,7 @@ import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; -import org.springframework.integration.aws.DynamoDbRunning; +import org.springframework.integration.aws.DynamoDbLocalRunning; import org.springframework.integration.test.util.TestUtils; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; @@ -47,7 +47,7 @@ import com.amazonaws.waiters.WaiterParameters; public class DynamoDbMetadataStoreTests { @ClassRule - public static final DynamoDbRunning DYNAMO_DB_RUNNING = DynamoDbRunning.isRunning(4567); + public static final DynamoDbLocalRunning DYNAMO_DB_RUNNING = DynamoDbLocalRunning.isRunning(4567); private static final String TEST_TABLE = "testMetadataStore";