From 908d51e9329bad8096dedbcd14ba262e33aaee98 Mon Sep 17 00:00:00 2001 From: Raja Kolli Date: Tue, 6 Jun 2023 06:42:32 +0000 Subject: [PATCH 1/6] adds integration test --- .../producer/docker/docker-compose.yml | 3 +- .../spring/ApplicationIntegrationTest.java | 59 ++++++++++++++++++- .../common/AbstractIntegrationTest.java | 24 +++++++- .../aws/spring/common/LocalStackConfig.java | 27 ++++++++- 4 files changed, 106 insertions(+), 7 deletions(-) diff --git a/aws-kinesis-project/producer/docker/docker-compose.yml b/aws-kinesis-project/producer/docker/docker-compose.yml index 53df9ce4..5e383add 100644 --- a/aws-kinesis-project/producer/docker/docker-compose.yml +++ b/aws-kinesis-project/producer/docker/docker-compose.yml @@ -6,8 +6,7 @@ services: ports: - "4566:4566" environment: - - EAGER_SERVICE_LOADING=1 - - SERVICES=kinesis, dynamodb + - SERVICES=kinesis - DEFAULT_REGION=us-east-1 - EDGE_PORT=4566 volumes: diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java index 5aea7f5d..5ac7966b 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java @@ -1,10 +1,67 @@ package com.learning.aws.spring; +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.core.JsonProcessingException; import com.learning.aws.spring.common.AbstractIntegrationTest; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; +import org.springframework.integration.aws.support.AwsHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; +import software.amazon.awssdk.core.SdkBytes; +import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest; +import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry; class ApplicationIntegrationTest extends AbstractIntegrationTest { + private static final String KINESIS_STREAM = "test_stream"; + @Test - void contextLoads() {} + void contextLoads() throws InterruptedException, JsonProcessingException { + PutRecordsRequest.Builder putRecordsRequest = + PutRecordsRequest.builder().streamName(KINESIS_STREAM); + + List putRecordsRequestEntryList = new ArrayList<>(); + + for (int i = 0; i < 10; i++) { + Message eventMessages = + MessageBuilder.withPayload("Message" + i) + .setHeader("event.eventType", "createEvent") + .build(); + PutRecordsRequestEntry putRecordsRequestEntry = + PutRecordsRequestEntry.builder() + .partitionKey("1") + .data( + SdkBytes.fromByteArray( + objectMapper.writeValueAsBytes(eventMessages))) + .build(); + putRecordsRequestEntryList.add(putRecordsRequestEntry); + } + putRecordsRequest.records(putRecordsRequestEntryList); + + amazonKinesis.putRecords(putRecordsRequest.build()); + + assertThat(this.messageBarrier.await(30, TimeUnit.SECONDS)).isTrue(); + + Message> message = this.messageHolder.get(); + assertThat(message.getHeaders()) + .containsKeys(AwsHeaders.CHECKPOINTER, AwsHeaders.SHARD, AwsHeaders.RECEIVED_STREAM) + .doesNotContainKeys(AwsHeaders.STREAM, AwsHeaders.PARTITION_KEY); + + List payload = message.getPayload(); + assertThat(payload).hasSize(10); + + Object item = payload.get(0); + + assertThat(item).isInstanceOf(GenericMessage.class); + + Message messageFromBatch = (Message) item; + + assertThat(messageFromBatch.getPayload()).isEqualTo("Message0"); + assertThat(messageFromBatch.getHeaders()).containsEntry("event.eventType", "createEvent"); + } } diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java index 8adcfb6c..25fd98af 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java @@ -4,15 +4,31 @@ import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT; import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.context.annotation.Import; +import org.springframework.messaging.Message; import org.springframework.test.context.ActiveProfiles; import org.springframework.test.web.servlet.MockMvc; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; @ActiveProfiles({PROFILE_TEST}) -@SpringBootTest(webEnvironment = RANDOM_PORT) +@SpringBootTest( + webEnvironment = RANDOM_PORT, + properties = { + "spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.destination=test_stream", + "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.idleBetweenPolls = 1", + "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.listenerMode = batch", + "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.checkpointMode = manual", + "spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.useNativeDecoding = true", + "spring.cloud.stream.kinesis.binder.headers = event.eventType", + "spring.cloud.stream.kinesis.binder.autoAddShards = true", + "spring.cloud.aws.region.static=eu-west-2" + }) @Import(LocalStackConfig.class) @AutoConfigureMockMvc public abstract class AbstractIntegrationTest { @@ -20,4 +36,10 @@ public abstract class AbstractIntegrationTest { @Autowired protected MockMvc mockMvc; @Autowired protected ObjectMapper objectMapper; + + @Autowired protected KinesisAsyncClient amazonKinesis; + + @Autowired protected CountDownLatch messageBarrier; + + @Autowired protected AtomicReference>> messageHolder; } diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java index e75dce36..0cd970b3 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java @@ -2,9 +2,14 @@ import static org.testcontainers.containers.localstack.LocalStackContainer.Service.KINESIS; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Primary; +import org.springframework.messaging.Message; import org.testcontainers.containers.localstack.LocalStackContainer; import org.testcontainers.utility.DockerImageName; import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; @@ -12,16 +17,14 @@ import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; -@TestConfiguration +@TestConfiguration(proxyBeanMethods = false) public class LocalStackConfig { static LocalStackContainer localStackContainer; static { - System.setProperty("com.amazonaws.sdk.disableCbor", "true"); localStackContainer = new LocalStackContainer( DockerImageName.parse("localstack/localstack").withTag("2.1.0")) - .withEnv("EAGER_SERVICE_LOADING", "1") .withServices(KINESIS) .withExposedPorts(4566); localStackContainer.start(); @@ -40,4 +43,22 @@ public KinesisAsyncClient amazonKinesis() { localStackContainer.getSecretKey()))) .build(); } + + @Bean + public AtomicReference>>> messageHolder() { + return new AtomicReference<>(); + } + + @Bean + public CountDownLatch messageBarrier() { + return new CountDownLatch(1); + } + + @Bean + public Consumer>>> eventConsumerBatchProcessingWithHeaders() { + return eventMessages -> { + messageHolder().set(eventMessages); + messageBarrier().countDown(); + }; + } } From e0c1dd9993fa6386185da02a018d1d847376355f Mon Sep 17 00:00:00 2001 From: Raja Kolli Date: Wed, 7 Jun 2023 16:49:25 +0530 Subject: [PATCH 2/6] fixes script --- .../producer/.localstack/init-aws.sh | 94 ++++++++++++++++++- .../producer/docker/docker-compose.yml | 2 +- 2 files changed, 93 insertions(+), 3 deletions(-) diff --git a/aws-kinesis-project/producer/.localstack/init-aws.sh b/aws-kinesis-project/producer/.localstack/init-aws.sh index 96dbbe8b..22b9a078 100755 --- a/aws-kinesis-project/producer/.localstack/init-aws.sh +++ b/aws-kinesis-project/producer/.localstack/init-aws.sh @@ -1,6 +1,91 @@ #!/bin/bash -awslocal kinesis create-stream --stream-name my-test-stream --shard-count 1 +# Variables +USER_NAME="localstack" +ROLE_NAME="localstack-role" +POLICY_NAME="kinesis-policy" +STREAM_NAME="my-test-stream" +SHARD_COUNT=1 + +# Create IAM user +awslocal iam create-user --user-name $USER_NAME + +# Create IAM role +awslocal iam create-role --role-name $ROLE_NAME --assume-role-policy-document '{ + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": { + "Service": "kinesis.amazonaws.com" + }, + "Action": "sts:AssumeRole" + } + ] +}' + +# Create IAM policy +cat > policy.json < Date: Thu, 8 Jun 2023 08:15:48 +0000 Subject: [PATCH 3/6] Another way to set data --- .../producer/.localstack/init-aws.sh | 110 ------------------ .../producer/docker/docker-compose.yml | 1 - aws-kinesis-project/producer/pom.xml | 2 + .../spring/ApplicationIntegrationTest.java | 29 ----- .../common/AbstractIntegrationTest.java | 4 +- .../aws/spring/common/LocalStackConfig.java | 38 ++---- 6 files changed, 14 insertions(+), 170 deletions(-) delete mode 100755 aws-kinesis-project/producer/.localstack/init-aws.sh diff --git a/aws-kinesis-project/producer/.localstack/init-aws.sh b/aws-kinesis-project/producer/.localstack/init-aws.sh deleted file mode 100755 index 22b9a078..00000000 --- a/aws-kinesis-project/producer/.localstack/init-aws.sh +++ /dev/null @@ -1,110 +0,0 @@ -#!/bin/bash - -# Variables -USER_NAME="localstack" -ROLE_NAME="localstack-role" -POLICY_NAME="kinesis-policy" -STREAM_NAME="my-test-stream" -SHARD_COUNT=1 - -# Create IAM user -awslocal iam create-user --user-name $USER_NAME - -# Create IAM role -awslocal iam create-role --role-name $ROLE_NAME --assume-role-policy-document '{ - "Version": "2012-10-17", - "Statement": [ - { - "Effect": "Allow", - "Principal": { - "Service": "kinesis.amazonaws.com" - }, - "Action": "sts:AssumeRole" - } - ] -}' - -# Create IAM policy -cat > policy.json <2.1.0 2.12.0 + 1.18.3 + ${project.build.directory}/test-results 2.37.0 8.2.1 diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java index 5ac7966b..7cd9a2f1 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/ApplicationIntegrationTest.java @@ -4,46 +4,17 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.learning.aws.spring.common.AbstractIntegrationTest; -import java.util.ArrayList; import java.util.List; import java.util.concurrent.TimeUnit; import org.junit.jupiter.api.Test; import org.springframework.integration.aws.support.AwsHeaders; -import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; -import software.amazon.awssdk.core.SdkBytes; -import software.amazon.awssdk.services.kinesis.model.PutRecordsRequest; -import software.amazon.awssdk.services.kinesis.model.PutRecordsRequestEntry; class ApplicationIntegrationTest extends AbstractIntegrationTest { - private static final String KINESIS_STREAM = "test_stream"; - @Test void contextLoads() throws InterruptedException, JsonProcessingException { - PutRecordsRequest.Builder putRecordsRequest = - PutRecordsRequest.builder().streamName(KINESIS_STREAM); - - List putRecordsRequestEntryList = new ArrayList<>(); - - for (int i = 0; i < 10; i++) { - Message eventMessages = - MessageBuilder.withPayload("Message" + i) - .setHeader("event.eventType", "createEvent") - .build(); - PutRecordsRequestEntry putRecordsRequestEntry = - PutRecordsRequestEntry.builder() - .partitionKey("1") - .data( - SdkBytes.fromByteArray( - objectMapper.writeValueAsBytes(eventMessages))) - .build(); - putRecordsRequestEntryList.add(putRecordsRequestEntry); - } - putRecordsRequest.records(putRecordsRequestEntryList); - - amazonKinesis.putRecords(putRecordsRequest.build()); assertThat(this.messageBarrier.await(30, TimeUnit.SECONDS)).isTrue(); diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java index 25fd98af..2811c5cb 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/AbstractIntegrationTest.java @@ -20,14 +20,12 @@ @SpringBootTest( webEnvironment = RANDOM_PORT, properties = { - "spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.destination=test_stream", "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.idleBetweenPolls = 1", "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.listenerMode = batch", "spring.cloud.stream.kinesis.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.checkpointMode = manual", "spring.cloud.stream.bindings.eventConsumerBatchProcessingWithHeaders-in-0.consumer.useNativeDecoding = true", "spring.cloud.stream.kinesis.binder.headers = event.eventType", - "spring.cloud.stream.kinesis.binder.autoAddShards = true", - "spring.cloud.aws.region.static=eu-west-2" + "spring.cloud.stream.kinesis.binder.autoAddShards = true" }) @Import(LocalStackConfig.class) @AutoConfigureMockMvc diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java index 0cd970b3..caf8cc2a 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java @@ -1,47 +1,31 @@ package com.learning.aws.spring.common; -import static org.testcontainers.containers.localstack.LocalStackContainer.Service.KINESIS; - import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Primary; import org.springframework.messaging.Message; import org.testcontainers.containers.localstack.LocalStackContainer; import org.testcontainers.utility.DockerImageName; -import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; -import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; -import software.amazon.awssdk.regions.Region; -import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; @TestConfiguration(proxyBeanMethods = false) public class LocalStackConfig { - static LocalStackContainer localStackContainer; + + static final LocalStackContainer localStackContainer = + new LocalStackContainer( + DockerImageName.parse("localstack/localstack").withTag("2.1.0")); static { - localStackContainer = - new LocalStackContainer( - DockerImageName.parse("localstack/localstack").withTag("2.1.0")) - .withServices(KINESIS) - .withExposedPorts(4566); localStackContainer.start(); - } - - @Bean - @Primary - public KinesisAsyncClient amazonKinesis() { - return KinesisAsyncClient.builder() - .endpointOverride(localStackContainer.getEndpointOverride(KINESIS)) - .region(Region.of(localStackContainer.getRegion())) - .credentialsProvider( - StaticCredentialsProvider.create( - AwsBasicCredentials.create( - localStackContainer.getAccessKey(), - localStackContainer.getSecretKey()))) - .build(); + System.setProperty( + "spring.cloud.aws.endpoint", localStackContainer.getEndpoint().toString()); + System.setProperty( + "spring.cloud.aws.credentials.access-key", localStackContainer.getAccessKey()); + System.setProperty( + "spring.cloud.aws.credentials.secret-key", localStackContainer.getSecretKey()); + System.setProperty("spring.cloud.aws.region.static", localStackContainer.getRegion()); } @Bean From 893b396dcf9902ee536c80a35d92d5972b7960d0 Mon Sep 17 00:00:00 2001 From: Raja Kolli Date: Tue, 25 Jul 2023 19:37:36 +0530 Subject: [PATCH 4/6] removes unnecessary declaration --- aws-kinesis-project/producer/pom.xml | 2 -- 1 file changed, 2 deletions(-) diff --git a/aws-kinesis-project/producer/pom.xml b/aws-kinesis-project/producer/pom.xml index ab439718..42c6dbbf 100644 --- a/aws-kinesis-project/producer/pom.xml +++ b/aws-kinesis-project/producer/pom.xml @@ -23,8 +23,6 @@ 2.1.0 2.13.0 - 1.18.3 - ${project.build.directory}/test-results 2.38.0 8.3.1 From e1656f4672e9d3a86932ec5f4dc19745bb46c605 Mon Sep 17 00:00:00 2001 From: Raja Kolli Date: Tue, 12 Mar 2024 08:03:26 +0000 Subject: [PATCH 5/6] fix : compile issue --- aws-kinesis-project/producer/pom.xml | 2 +- .../java/com/learning/aws/spring/common/LocalStackConfig.java | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/aws-kinesis-project/producer/pom.xml b/aws-kinesis-project/producer/pom.xml index 43fa0070..2cdf45dd 100644 --- a/aws-kinesis-project/producer/pom.xml +++ b/aws-kinesis-project/producer/pom.xml @@ -227,7 +227,7 @@ - 1.18.1 + 1.19.2 diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java index 2063d255..5e44c349 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java @@ -1,5 +1,7 @@ package com.learning.aws.spring.common; +import static org.testcontainers.containers.localstack.LocalStackContainer.Service.KINESIS; + import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicReference; @@ -20,7 +22,6 @@ public class LocalStackConfig { localStackContainer = new LocalStackContainer( DockerImageName.parse("localstack/localstack").withTag("3.2.0")) - .withEnv("EAGER_SERVICE_LOADING", "1") .withServices(KINESIS) .withExposedPorts(4566); localStackContainer.start(); From e1f2dfb6b9b6706c03bb0d38d1d426ba095dc2c6 Mon Sep 17 00:00:00 2001 From: Raja Kolli Date: Tue, 12 Mar 2024 08:11:54 +0000 Subject: [PATCH 6/6] adds kinesis asyncClient Bean --- .../aws/spring/common/LocalStackConfig.java | 26 ++++++++++++++----- 1 file changed, 19 insertions(+), 7 deletions(-) diff --git a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java index 5e44c349..c54a2fa3 100644 --- a/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java +++ b/aws-kinesis-project/producer/src/test/java/com/learning/aws/spring/common/LocalStackConfig.java @@ -8,9 +8,14 @@ import java.util.function.Consumer; import org.springframework.boot.test.context.TestConfiguration; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; import org.springframework.messaging.Message; import org.testcontainers.containers.localstack.LocalStackContainer; import org.testcontainers.utility.DockerImageName; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.kinesis.KinesisAsyncClient; @TestConfiguration(proxyBeanMethods = false) public class LocalStackConfig { @@ -25,13 +30,20 @@ public class LocalStackConfig { .withServices(KINESIS) .withExposedPorts(4566); localStackContainer.start(); - System.setProperty( - "spring.cloud.aws.endpoint", localStackContainer.getEndpoint().toString()); - System.setProperty( - "spring.cloud.aws.credentials.access-key", localStackContainer.getAccessKey()); - System.setProperty( - "spring.cloud.aws.credentials.secret-key", localStackContainer.getSecretKey()); - System.setProperty("spring.cloud.aws.region.static", localStackContainer.getRegion()); + } + + @Bean + @Primary + public KinesisAsyncClient amazonKinesis() { + return KinesisAsyncClient.builder() + .endpointOverride(localStackContainer.getEndpointOverride(KINESIS)) + .region(Region.of(localStackContainer.getRegion())) + .credentialsProvider( + StaticCredentialsProvider.create( + AwsBasicCredentials.create( + localStackContainer.getAccessKey(), + localStackContainer.getSecretKey()))) + .build(); } @Bean