Spring/spring-integration-aws
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Increase poll timeouts in tests

Browse Source 
* Add `describeStream()` requests to Kinesis tests
to wait for `ACTIVE` before starting to interaction with the stream
This commit is contained in:
Artem Bilan
2022-03-31 10:26:34 -04:00
parent 62b54d1cdf
commit 15ea0b181e
5 changed files with 29 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/org/springframework/integration/aws/inbound/kinesis/KinesisMessageDrivenChannelAdapter.java
View File

@@ -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

3
src/test/java/org/springframework/integration/aws/LocalstackContainerTest.java
View File

@@ -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() {

15
src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java
View File

@@ -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<String> 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();

13
src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java
View File

@@ -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();

10
src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java
View File

@@ -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();