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 b29f353..da24e08 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 @@ -1,5 +1,5 @@ /* - * Copyright 2017-2021 the original author or authors. + * Copyright 2017-2022 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. @@ -336,7 +336,7 @@ public class KinesisMessageDrivenChannelAdapter extends MessageProducerSupport } /** - * Specify a {@link LockRegistry} for an exclusive access to provided streams. This is not used + * Specify a {@link LockRegistry} for exclusive access to provided streams. This is not used * when shards-based configuration is provided. * @param lockRegistry the {@link LockRegistry} to use. * @since 2.0 diff --git a/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java b/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java index e499e14..65eb35e 100644 --- a/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java +++ b/src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java @@ -46,8 +46,7 @@ public interface LocalstackContainerTest { .withServices( LocalStackContainer.Service.DYNAMODB, LocalStackContainer.Service.KINESIS, - LocalStackContainer.Service.CLOUDWATCH) - .withReuse(true); + LocalStackContainer.Service.CLOUDWATCH); static AmazonDynamoDBAsync dynamoDbClient() { 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 75f6c8d..15f44b8 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java @@ -80,9 +80,16 @@ public class KinesisIntegrationTests implements LocalstackContainerTest { private PollableChannel errorChannel; @BeforeAll - static void setup() { + static void setup() throws Exception { AMAZON_KINESIS_ASYNC = LocalstackContainerTest.kinesisClient(); AMAZON_KINESIS_ASYNC.createStream(TEST_STREAM, 1); + + int n = 0; + while (n++ < 100 && !"ACTIVE".equals( + AMAZON_KINESIS_ASYNC.describeStream(TEST_STREAM).getStreamDescription().getStreamStatus())) { + + Thread.sleep(200); + } } @AfterAll @@ -99,13 +106,13 @@ public class KinesisIntegrationTests implements LocalstackContainerTest { this.kinesisSendChannel.send(MessageBuilder.withPayload(now).setHeader(AwsHeaders.STREAM, TEST_STREAM) .setHeader("foo", "BAR").build()); - Message receive = this.kinesisReceiveChannel.receive(20_000); + Message receive = this.kinesisReceiveChannel.receive(30_000); assertThat(receive).isNotNull(); assertThat(receive.getPayload()).isEqualTo(now); assertThat(receive.getHeaders()).contains(entry("foo", "BAR")); assertThat(receive.getHeaders()).containsKey(IntegrationMessageHeaderAccessor.SOURCE_DATA); - Message errorMessage = this.errorChannel.receive(20_000); + Message errorMessage = this.errorChannel.receive(30_000); assertThat(errorMessage).isNotNull(); assertThat(errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD)).isNotNull(); assertThat(((Exception) errorMessage.getPayload()).getMessage()) @@ -120,7 +127,7 @@ public class KinesisIntegrationTests implements LocalstackContainerTest { Set receivedSequences = new HashSet<>(); for (int i = 0; i < 2; i++) { - receive = this.kinesisReceiveChannel.receive(20_000); + receive = this.kinesisReceiveChannel.receive(30_000); assertThat(receive).isNotNull(); String sequenceNumber = receive.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, String.class); assertThat(receivedSequences.add(sequenceNumber)).isTrue(); diff --git a/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java b/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java index 6b119da..619426d 100644 --- a/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java +++ b/src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java @@ -88,11 +88,18 @@ public class KplKclIntegrationTests implements LocalstackContainerTest { private PollableChannel errorChannel; @BeforeAll - static void setup() { + static void setup() throws InterruptedException { AMAZON_KINESIS = LocalstackContainerTest.kinesisClient(); DYNAMO_DB = LocalstackContainerTest.dynamoDbClient(); CLOUD_WATCH = LocalstackContainerTest.cloudWatchClient(); AMAZON_KINESIS.createStream(TEST_STREAM, 1); + + int n = 0; + while (n++ < 100 && !"ACTIVE".equals( + AMAZON_KINESIS.describeStream(TEST_STREAM).getStreamDescription().getStreamStatus())) { + + Thread.sleep(200); + } } @AfterAll @@ -115,7 +122,7 @@ public class KplKclIntegrationTests implements LocalstackContainerTest { assertThat(receive.getHeaders()).contains(entry("foo", "BAR")); assertThat(receive.getHeaders()).containsKey(IntegrationMessageHeaderAccessor.SOURCE_DATA); - Message errorMessage = this.errorChannel.receive(10_000); + Message errorMessage = this.errorChannel.receive(30_000); assertThat(errorMessage).isNotNull(); assertThat(errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD)).isNotNull(); assertThat(((Exception) errorMessage.getPayload()).getMessage()) @@ -125,7 +132,7 @@ public class KplKclIntegrationTests implements LocalstackContainerTest { this.kinesisSendChannel .send(MessageBuilder.withPayload(new Date()).setHeader(AwsHeaders.STREAM, TEST_STREAM).build()); - receive = this.kinesisReceiveChannel.receive(20_000); + receive = this.kinesisReceiveChannel.receive(30_000); assertThat(receive).isNotNull(); assertThat(receive.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, String.class)).isNotEmpty(); diff --git a/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java b/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java index 69a05b6..51d9503 100644 --- a/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java +++ b/src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java @@ -98,7 +98,7 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes initiator.start(); } - assertThat(granted.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(granted.await(20, TimeUnit.SECONDS)).isTrue(); LockRegistryLeaderInitiator initiator1 = countingPublisher.initiator; @@ -158,7 +158,7 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes initiator1.getContext().yield(); - assertThat(revoked11.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(revoked11.await(20, TimeUnit.SECONDS)).isTrue(); assertThat(initiator1.getContext().isLeader()).isFalse(); initiator1.stop(); @@ -181,11 +181,11 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes initiator.start(); - assertThat(granted.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(granted.await(20, TimeUnit.SECONDS)).isTrue(); destroy(); - assertThat(countingPublisher.revoked.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(countingPublisher.revoked.await(20, TimeUnit.SECONDS)).isTrue(); granted = new CountDownLatch(1); countingPublisher = new CountingPublisher(granted); @@ -195,7 +195,7 @@ class DynamoDbLockRegistryLeaderInitiatorTests implements LocalstackContainerTes lockRepository.afterPropertiesSet(); - assertThat(granted.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(granted.await(20, TimeUnit.SECONDS)).isTrue(); initiator.stop();