Upgrade Testcontainers tests to the latest images

* Un`@Disable` updated tests

common/cdc-debezium-boot-starter/src/test/java/org/springframework/cloud/fn/common/cdc/CdcBootStarterIntegrationTest.java

@@ -18,6 +18,7 @@ package org.springframework.cloud.fn.common.cdc;
 
 import java.time.Duration;
 
+import com.zaxxer.hikari.HikariDataSource;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.GenericContainer;
@@ -29,8 +30,6 @@ import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.test.jdbc.JdbcTestUtils;
 
-import com.zaxxer.hikari.HikariDataSource;
-
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
@@ -48,7 +47,7 @@ public class CdcBootStarterIntegrationTest {
 
 	@Container
 	static GenericContainer debeziumMySQL =
-			new GenericContainer<>(DockerImageName.parse("debezium/example-mysql:latest"))
+			new GenericContainer<>(DockerImageName.parse("debezium/example-mysql"))
 					.withEnv("MYSQL_ROOT_PASSWORD", "debezium")
 					.withEnv("MYSQL_USER", "mysqluser")
 					.withEnv("MYSQL_PASSWORD", "mysqlpw")

consumer/mqtt-consumer/src/test/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -19,7 +19,6 @@ package org.springframework.cloud.fn.consumer.mqtt;
 import java.util.function.Consumer;
 
 import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.GenericContainer;
 
@@ -43,7 +42,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class MqttConsumerTests {
 
 	static {
-		GenericContainer mosquitto = new GenericContainer("eclipse-mosquitto")
+		GenericContainer mosquitto = new GenericContainer("cyrilix/rabbitmq-mqtt")
 				.withExposedPorts(1883);
 		mosquitto.start();
 		final Integer mappedPort = mosquitto.getMappedPort(1883);
@@ -62,7 +61,6 @@ public class MqttConsumerTests {
 	}
 
 	@Test
-	@Disabled
 	public void testMqttConsumer() {
 		this.mqttConsumer.accept(MessageBuilder.withPayload("hello").build());
 		Message<?> in = this.queue.receive(10000);
@@ -97,5 +95,7 @@ public class MqttConsumerTests {
 		public QueueChannel queue() {
 			return new QueueChannel();
 		}
+
 	}
+
 }

supplier/mqtt-supplier/src/test/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 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.
@@ -19,7 +19,6 @@ package org.springframework.cloud.fn.supplier.mqtt;
 import java.util.function.Supplier;
 
 import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.testcontainers.containers.GenericContainer;
 import reactor.core.publisher.Flux;
@@ -45,6 +44,7 @@ import static org.assertj.core.api.Assertions.assertThat;
  * @author Janne Valkealahti
  * @author Gary Russell
  * @author Soby Chacko
+ * @author Artem Bilan
  *
  */
 @SpringBootTest(properties = {"mqtt.supplier.topics=test,fake", "mqtt.supplier.qos=0,0"})
@@ -52,7 +52,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 public class MqttSupplierTests {
 
 	static {
-		GenericContainer mosquitto = new GenericContainer("eclipse-mosquitto")
+		GenericContainer mosquitto = new GenericContainer("cyrilix/rabbitmq-mqtt")
 				.withExposedPorts(1883);
 		mosquitto.start();
 		final Integer mappedPort = mosquitto.getMappedPort(1883);
@@ -71,7 +71,6 @@ public class MqttSupplierTests {
 	private MessageHandler mqttOutbound;
 
 	@Test
-	@Disabled
 	public void testBasicFlow() {
 
 		mqttOutbound.handleMessage(MessageBuilder.withPayload("hello").build());

supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2020 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -117,6 +117,16 @@ public abstract class AbstractAwsS3SupplierMockTests {
 	@AfterAll
 	public static void tearDown() {
 		System.clearProperty("s3.supplier.localDir");
+		S3_OBJECTS.stream()
+				.map(S3Object::getObjectContent)
+				.forEach(stream -> {
+					try {
+						stream.close();
+					}
+					catch (IOException e) {
+						// Ignore
+					}
+				});
 	}
 
 	@SpringBootApplication
@@ -141,9 +151,7 @@ public abstract class AbstractAwsS3SupplierMockTests {
 				objectSummaries.add(s3ObjectSummary);
 			}
 
-			willAnswer(invocation -> {
-				return objectListing;
-			}).given(amazonS3).listObjects(any(ListObjectsRequest.class));
+			willAnswer(invocation -> objectListing).given(amazonS3).listObjects(any(ListObjectsRequest.class));
 
 			for (final S3Object s3Object : S3_OBJECTS) {
 				willAnswer(invocation -> s3Object).given(amazonS3).getObject(S3_BUCKET, s3Object.getKey());