GH-113: Add KPL+KCL tests against Local Stack

Fixes https://github.com/spring-projects/spring-integration-aws/issues/113

* Fix some properties in the `KclMessageDrivenChannelAdapter` for a
reasonable default value
* Introduce `LocalStackSslEnvironmentProvider` to populate a `USE_SSL`
ENV var into a Docker instance for Local Stack
* Add more factory methods into an `ExtendedDockerTestUtils`, especially
variants to obtain an HTTPS-based service clients

src/main/java/org/springframework/integration/aws/inbound/kinesis/KclMessageDrivenChannelAdapter.java

@@ -105,9 +105,9 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
 
 	private InitialPositionInStream streamInitialSequence = InitialPositionInStream.LATEST;
 
-	private int idleBetweenPolls;
+	private int idleBetweenPolls = 1000;
 
-	private int consumerBackoff;
+	private int consumerBackoff = 1000;
 
 	private Converter<byte[], Object> converter = new DeserializingConverter();
 
@@ -298,7 +298,9 @@ public class KclMessageDrivenChannelAdapter extends MessageProducerSupport {
 	@Override
 	public void destroy() {
 		super.destroy();
-		this.scheduler.shutdown();
+		if (isRunning()) {
+			this.scheduler.shutdown();
+		}
 	}
 
 	@Override

src/test/java/org/springframework/integration/aws/EnvironmentHostNameResolver.java

@@ -20,6 +20,11 @@ import cloud.localstack.docker.annotation.IHostNameResolver;
 import com.amazonaws.SDKGlobalConfiguration;
 
 /**
+ * An {@link IHostNameResolver} implementation for {@value EnvironmentHostNameResolver#DOCKER_HOST_NAME}
+ * environment variable to resolve for Local Stack Docker instance.
+ * Also this class places an {@value SDKGlobalConfiguration#AWS_CBOR_DISABLE_SYSTEM_PROPERTY}
+ * system property to disable CBOR for services requests.
+ *
  * @author Artem Bilan
  *
  * @since 2.3

src/test/java/org/springframework/integration/aws/ExtendedDockerTestUtils.java

@@ -25,12 +25,18 @@ import cloud.localstack.docker.LocalstackDocker;
 import com.amazonaws.ClientConfiguration;
 import com.amazonaws.client.builder.AwsAsyncClientBuilder;
 import com.amazonaws.client.builder.AwsClientBuilder;
+import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsync;
+import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsyncClientBuilder;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClientBuilder;
 import com.amazonaws.services.kinesis.AmazonKinesisAsync;
 import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder;
 
 /**
+ * An utility class for providing AWS {@code async} clients based on the Local Stack Docker instance.
+ * In addition it provides SSL-based async clients.
+ *
+ *
  * @author Artem Bilan
  *
  * @since 2.3
@@ -38,23 +44,61 @@ import com.amazonaws.services.kinesis.AmazonKinesisAsyncClientBuilder;
 public final class ExtendedDockerTestUtils {
 
 	public static AmazonKinesisAsync getClientKinesisAsync() {
+		return doGetClientKinesisAsync(false);
+	}
+
+	public static AmazonKinesisAsync getClientKinesisAsyncSsl() {
+		return doGetClientKinesisAsync(true);
+	}
+
+	private static AmazonKinesisAsync doGetClientKinesisAsync(boolean ssl) {
 		AmazonKinesisAsyncClientBuilder amazonKinesisAsyncClientBuilder =
 				AmazonKinesisAsyncClientBuilder.standard()
 						.withEndpointConfiguration(
-								createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointKinesis));
+								createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointKinesis, ssl));
 		return applyConfigurationAndBuild(amazonKinesisAsyncClientBuilder);
 	}
 
 	public static AmazonDynamoDBAsync getClientDynamoDbAsync() {
+		return doClientDynamoDbAsync(false);
+	}
+
+	public static AmazonDynamoDBAsync getClientDynamoDbAsyncSsl() {
+		return doClientDynamoDbAsync(true);
+	}
+
+	private static AmazonDynamoDBAsync doClientDynamoDbAsync(boolean ssl) {
 		AmazonDynamoDBAsyncClientBuilder dynamoDBAsyncClientBuilder =
 				AmazonDynamoDBAsyncClientBuilder.standard()
 						.withEndpointConfiguration(
-								createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointDynamoDB));
+								createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointDynamoDB, ssl));
 		return applyConfigurationAndBuild(dynamoDBAsyncClientBuilder);
 	}
 
-	private static AwsClientBuilder.EndpointConfiguration createEndpointConfiguration(Supplier<String> supplier) {
-		return new AwsClientBuilder.EndpointConfiguration(supplier.get(), DEFAULT_REGION);
+	public static AmazonCloudWatchAsync getClientCloudWatchAsync() {
+		return doClientCloudWatchAsync(false);
+	}
+
+	public static AmazonCloudWatchAsync getClientCloudWatchAsyncSsl() {
+		return doClientCloudWatchAsync(true);
+	}
+
+	private static AmazonCloudWatchAsync doClientCloudWatchAsync(boolean ssl) {
+		AmazonCloudWatchAsyncClientBuilder cloudWatchAsyncClientBuilder =
+				AmazonCloudWatchAsyncClientBuilder.standard()
+						.withEndpointConfiguration(
+								createEndpointConfiguration(LocalstackDocker.INSTANCE::getEndpointCloudWatch, ssl));
+		return applyConfigurationAndBuild(cloudWatchAsyncClientBuilder);
+	}
+
+	private static AwsClientBuilder.EndpointConfiguration createEndpointConfiguration(Supplier<String> supplier,
+			boolean ssl) {
+
+		String serviceEndpoint = supplier.get();
+		if (ssl) {
+			serviceEndpoint = serviceEndpoint.replaceFirst("http", "https");
+		}
+		return new AwsClientBuilder.EndpointConfiguration(serviceEndpoint, DEFAULT_REGION);
 	}
 
 	private static <T, C extends AwsAsyncClientBuilder<C, T>> T applyConfigurationAndBuild(C builder) {

src/test/java/org/springframework/integration/aws/LocalStackSslEnvironmentProvider.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2019 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
+ *
+ *      https://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 java.util.Collections;
+import java.util.Map;
+
+import cloud.localstack.docker.annotation.IEnvironmentVariableProvider;
+import com.amazonaws.SDKGlobalConfiguration;
+
+/**
+ * An {@link IEnvironmentVariableProvider} implementation to provide a {@code USE_SSL}
+ * environment variable for docker to start a Local Stack in TLS mode.
+ * Also this class populates a {@value SDKGlobalConfiguration#DISABLE_CERT_CHECKING_SYSTEM_PROPERTY}
+ * system property to disable SSL certificates validation.
+ *
+ * @author Artem Bilan
+ *
+ * @since 2.3
+ */
+public class LocalStackSslEnvironmentProvider implements IEnvironmentVariableProvider {
+
+	static {
+		System.setProperty(SDKGlobalConfiguration.DISABLE_CERT_CHECKING_SYSTEM_PROPERTY, "true");
+	}
+
+	@Override
+	public Map<String, String> getEnvironmentVariables() {
+		return Collections.singletonMap("USE_SSL", "true");
+	}
+
+}

src/test/java/org/springframework/integration/aws/kinesis/KinesisIntegrationTests.java

@@ -67,7 +67,6 @@ import com.amazonaws.services.kinesis.AmazonKinesisAsync;
  *
  * @since 1.1
  */
-//@Disabled("Looks like Kinesis is not supported well in Local Stack")
 @SpringJUnitConfig
 @EnabledIfEnvironmentVariable(named = EnvironmentHostNameResolver.DOCKER_HOST_NAME, matches = ".+")
 @ExtendWith(LocalstackDockerExtension.class)
@@ -123,21 +122,21 @@ public class KinesisIntegrationTests {
 				.contains("Channel 'kinesisReceiveChannel' expected one of the following data types "
 						+ "[class java.util.Date], but received [class java.lang.String]");
 
-		for (int i = 0; i < 10; i++) {
+		for (int i = 0; i < 2; i++) {
 			this.kinesisSendChannel
 					.send(MessageBuilder.withPayload(new Date()).setHeader(AwsHeaders.STREAM, TEST_STREAM).build());
 		}
 
 		Set<String> receivedSequences = new HashSet<>();
 
-		for (int i = 0; i < 10; i++) {
+		for (int i = 0; i < 2; i++) {
 			receive = this.kinesisReceiveChannel.receive(20_000);
 			assertThat(receive).isNotNull();
 			String sequenceNumber = receive.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, String.class);
 			assertThat(receivedSequences.add(sequenceNumber)).isTrue();
 		}
 
-		assertThat(receivedSequences.size()).isEqualTo(10);
+		assertThat(receivedSequences.size()).isEqualTo(2);
 
 		receive = this.kinesisReceiveChannel.receive(10);
 		assertThat(receive).isNull();

src/test/java/org/springframework/integration/aws/kinesis/KplKclIntegrationTests.java

@@ -0,0 +1,212 @@
+/*
+ * Copyright 2017-2019 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
+ *
+ *      https://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 static org.assertj.core.api.Assertions.entry;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Date;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.IntegrationMessageHeaderAccessor;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.aws.EnvironmentHostNameResolver;
+import org.springframework.integration.aws.ExtendedDockerTestUtils;
+import org.springframework.integration.aws.LocalStackSslEnvironmentProvider;
+import org.springframework.integration.aws.inbound.kinesis.KclMessageDrivenChannelAdapter;
+import org.springframework.integration.aws.inbound.kinesis.KinesisMessageHeaderErrorMessageStrategy;
+import org.springframework.integration.aws.outbound.KplMessageHandler;
+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.integration.support.json.EmbeddedJsonHeadersMessageMapper;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import cloud.localstack.TestUtils;
+import cloud.localstack.docker.LocalstackDocker;
+import cloud.localstack.docker.LocalstackDockerExtension;
+import cloud.localstack.docker.annotation.LocalstackDockerProperties;
+import com.amazonaws.services.cloudwatch.AmazonCloudWatch;
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDB;
+import com.amazonaws.services.kinesis.AmazonKinesis;
+import com.amazonaws.services.kinesis.clientlibrary.lib.worker.InitialPositionInStream;
+import com.amazonaws.services.kinesis.producer.KinesisProducer;
+import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration;
+
+/**
+ * @author Artem Bilan
+ *
+ * @since 1.1
+ */
+@SpringJUnitConfig
+@EnabledIfEnvironmentVariable(named = EnvironmentHostNameResolver.DOCKER_HOST_NAME, matches = ".+")
+@ExtendWith(LocalstackDockerExtension.class)
+@LocalstackDockerProperties(randomizePorts = true,
+		hostNameResolver = EnvironmentHostNameResolver.class,
+		environmentVariableProvider = LocalStackSslEnvironmentProvider.class,
+		services = { "kinesis", "dynamodb", "cloudwatch" })
+@DirtiesContext
+public class KplKclIntegrationTests {
+
+	private static final String TEST_STREAM = "TestStream";
+
+	private static AmazonKinesis AMAZON_KINESIS;
+
+	private static AmazonDynamoDB DYNAMO_DB;
+
+	private static AmazonCloudWatch CLOUD_WATCH;
+
+	@Autowired
+	private MessageChannel kinesisSendChannel;
+
+	@Autowired
+	private PollableChannel kinesisReceiveChannel;
+
+	@Autowired
+	private PollableChannel errorChannel;
+
+	@BeforeAll
+	static void setup() {
+		AMAZON_KINESIS = ExtendedDockerTestUtils.getClientKinesisAsyncSsl();
+		DYNAMO_DB = ExtendedDockerTestUtils.getClientDynamoDbAsyncSsl();
+		CLOUD_WATCH = ExtendedDockerTestUtils.getClientCloudWatchAsyncSsl();
+		AMAZON_KINESIS.createStream(TEST_STREAM, 1);
+	}
+
+	@AfterAll
+	static void tearDown() {
+		AMAZON_KINESIS.deleteStream(TEST_STREAM);
+	}
+
+	@Test
+	void testKinesisInboundOutbound() {
+		this.kinesisSendChannel
+				.send(MessageBuilder.withPayload("foo").setHeader(AwsHeaders.STREAM, TEST_STREAM).build());
+
+		Date now = new Date();
+		this.kinesisSendChannel.send(MessageBuilder.withPayload(now).setHeader(AwsHeaders.STREAM, TEST_STREAM)
+				.setHeader("foo", "BAR").build());
+
+		Message<?> receive = this.kinesisReceiveChannel.receive(20_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(10_000);
+		assertThat(errorMessage).isNotNull();
+		assertThat(errorMessage.getHeaders().get(AwsHeaders.RAW_RECORD)).isNotNull();
+		assertThat(((Exception) errorMessage.getPayload()).getMessage())
+				.contains("Channel 'kinesisReceiveChannel' expected one of the following data types "
+						+ "[class java.util.Date], but received [class java.lang.String]");
+
+		this.kinesisSendChannel
+				.send(MessageBuilder.withPayload(new Date()).setHeader(AwsHeaders.STREAM, TEST_STREAM).build());
+
+		receive = this.kinesisReceiveChannel.receive(20_000);
+		assertThat(receive).isNotNull();
+		assertThat(receive.getHeaders().get(AwsHeaders.RECEIVED_SEQUENCE_NUMBER, String.class)).isNotEmpty();
+
+		receive = this.kinesisReceiveChannel.receive(10);
+		assertThat(receive).isNull();
+	}
+
+	@Configuration
+	@EnableIntegration
+	public static class TestConfiguration {
+
+		@Bean
+		public KinesisProducerConfiguration kinesisProducerConfiguration() throws URISyntaxException {
+			URI kinesisUri = new URI(LocalstackDocker.INSTANCE.getEndpointKinesis());
+			URI cloudWatchUri = new URI(LocalstackDocker.INSTANCE.getEndpointCloudWatch());
+			return new KinesisProducerConfiguration()
+					.setCredentialsProvider(TestUtils.getCredentialsProvider())
+					.setRegion(TestUtils.DEFAULT_REGION)
+					.setKinesisEndpoint(kinesisUri.getHost())
+					.setKinesisPort(kinesisUri.getPort())
+					.setCloudwatchEndpoint(cloudWatchUri.getHost())
+					.setCloudwatchPort(cloudWatchUri.getPort())
+					.setVerifyCertificate(false);
+		}
+
+		@Bean
+		@ServiceActivator(inputChannel = "kinesisSendChannel")
+		public MessageHandler kplMessageHandler(KinesisProducerConfiguration kinesisProducerConfiguration) {
+			KplMessageHandler kinesisMessageHandler =
+					new KplMessageHandler(new KinesisProducer(kinesisProducerConfiguration));
+			kinesisMessageHandler.setPartitionKey("1");
+			kinesisMessageHandler.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo"));
+			return kinesisMessageHandler;
+		}
+
+		@Bean
+		public KclMessageDrivenChannelAdapter kclMessageDrivenChannelAdapter() {
+			KclMessageDrivenChannelAdapter adapter = new KclMessageDrivenChannelAdapter(
+					TEST_STREAM, AMAZON_KINESIS, CLOUD_WATCH, DYNAMO_DB, TestUtils.getCredentialsProvider());
+			adapter.setOutputChannel(kinesisReceiveChannel());
+			adapter.setErrorChannel(errorChannel());
+			adapter.setErrorMessageStrategy(new KinesisMessageHeaderErrorMessageStrategy());
+			adapter.setEmbeddedHeadersMapper(new EmbeddedJsonHeadersMessageMapper("foo"));
+			adapter.setStreamInitialSequence(InitialPositionInStream.TRIM_HORIZON);
+			adapter.setBindSourceRecord(true);
+			return adapter;
+		}
+
+		@Bean
+		public PollableChannel kinesisReceiveChannel() {
+			QueueChannel queueChannel = new QueueChannel();
+			queueChannel.setDatatypes(Date.class);
+			return queueChannel;
+		}
+
+		@Bean
+		public PollableChannel errorChannel() {
+			QueueChannel queueChannel = new QueueChannel();
+			queueChannel.addInterceptor(new ChannelInterceptor() {
+
+				@Override
+				public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+					if (message instanceof ErrorMessage) {
+						throw (RuntimeException) ((ErrorMessage) message).getPayload();
+					}
+				}
+
+			});
+			return queueChannel;
+		}
+
+	}
+
+}

src/test/java/org/springframework/integration/aws/leader/DynamoDbLockRegistryLeaderInitiatorTests.java

@@ -86,7 +86,6 @@ class DynamoDbLockRegistryLeaderInitiatorTests {
 		DYNAMO_DB.deleteTable(DynamoDbLockRegistry.DEFAULT_TABLE_NAME);
 	}
 
-	//	@Disabled("Doesn't work properly against Local Stack when two instances try to lock in table")
 	@Test
 	void testDistributedLeaderElection() throws Exception {
 		CountDownLatch granted = new CountDownLatch(1);