Use testcontainers for Rabbit binder integration tests

Fixes #2341

Resolves #2358

binders/rabbit-binder/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-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.
@@ -44,22 +44,31 @@ public class RabbitTestSupport
 		extends AbstractExternalResourceTestSupport<CachingConnectionFactory> {
 
 	private final boolean management;
+	private final int ampqPort;
+	private final int managmentPort;
 
 	public RabbitTestSupport() {
 		this(false);
 	}
 
 	public RabbitTestSupport(boolean management) {
+		this(management, 5672, 15672);
+	}
+
+	public RabbitTestSupport(boolean management, int amqpPort, int managementPort) {
 		super("RABBIT");
 		this.management = management;
+		this.ampqPort = amqpPort;
+		this.managmentPort = managementPort;
+
 	}
 
 	@Override
 	protected void obtainResource() throws Exception {
-		resource = new CachingConnectionFactory("localhost");
+		resource = new CachingConnectionFactory("localhost", this.ampqPort);
 		resource.createConnection().close();
 		if (management) {
-			Socket socket = SocketFactory.getDefault().createSocket("localhost", 15672);
+			Socket socket = SocketFactory.getDefault().createSocket("localhost", this.managmentPort);
 			socket.close();
 		}
 	}
@@ -96,6 +105,10 @@ public class RabbitTestSupport
 		}
 
 		public void start() throws IOException {
+			start(5672);
+		}
+
+		public void start(int amqpPort) throws IOException {
 			this.serverSocket = ServerSocketFactory.getDefault()
 					.createServerSocket(this.port, 10);
 			LOGGER.info("Proxy started");
@@ -114,7 +127,7 @@ public class RabbitTestSupport
 									try {
 										final Socket rabbitSocket = SocketFactory
 												.getDefault()
-												.createSocket("localhost", 5672);
+												.createSocket("localhost", amqpPort);
 										socketExec.execute(new Runnable() {
 
 											@Override

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/LocalizedQueueConnectionFactoryIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-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.
@@ -21,6 +21,7 @@ import java.util.UUID;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.testcontainers.containers.RabbitMQContainer;
 
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
@@ -34,20 +35,22 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Gary Russell
+ * @author Chris Bono
  */
 public class LocalizedQueueConnectionFactoryIntegrationTests {
 
+	private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
+
 	@RegisterExtension
-	public static RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true);
+	private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
 
 	private LocalizedQueueConnectionFactory lqcf;
 
 	@BeforeEach
 	public void setup() {
-		ConnectionFactory defaultConnectionFactory = rabbitAvailableRule.getResource();
-		String[] addresses = new String[] { "localhost:9999", "localhost:5672" };
-		String[] adminAddresses = new String[] { "http://localhost:15672",
-				"http://localhost:15672" };
+		ConnectionFactory defaultConnectionFactory = rabbitTestSupport.getResource();
+		String[] addresses = new String[] { "localhost:9999", "localhost:" + RABBITMQ.getAmqpPort() };
+		String[] adminAddresses = new String[] { RABBITMQ.getHttpUrl(), RABBITMQ.getHttpUrl() };
 		String[] nodes = new String[] { "foo@bar", "rabbit@localhost" };
 		String vhost = "/";
 		String username = "guest";

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-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.
@@ -22,13 +22,13 @@ import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 
-import com.fasterxml.jackson.annotation.JsonProperty;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.http.client.Client;
 import com.rabbitmq.http.client.domain.QueueInfo;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
+import org.testcontainers.containers.RabbitMQContainer;
 
 import org.springframework.amqp.core.Base64UrlNamingStrategy;
 import org.springframework.amqp.core.BindingBuilder;
@@ -48,17 +48,19 @@ import static org.junit.Assert.fail;
 
 /**
  * @author Gary Russell
+ * @author Chris Bono
  * @since 1.2
  */
 public class RabbitBinderCleanerTests {
 
+	private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
+
 	private static final String BINDER_PREFIX = "binder.";
 
 	private static final Client client;
-
 	static {
 		try {
-			client = new Client("http://localhost:15672/api", "guest", "guest");
+			client = new Client(RABBITMQ.getHttpUrl() + "/api", "guest", "guest");
 		}
 		catch (MalformedURLException | URISyntaxException e) {
 			throw new RabbitAdminException("Couldn't create a Client", e);
@@ -66,7 +68,7 @@ public class RabbitBinderCleanerTests {
 	}
 
 	@RegisterExtension
-	public RabbitTestSupport rabbitWithMgmtEnabled = new RabbitTestSupport(true);
+	private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
 
 	@Test
 	public void testCleanStream() {
@@ -74,7 +76,7 @@ public class RabbitBinderCleanerTests {
 		final String stream1 = new Base64UrlNamingStrategy("foo").generateName();
 		String stream2 = stream1 + "-1";
 		String firstQueue = null;
-		CachingConnectionFactory connectionFactory = rabbitWithMgmtEnabled.getResource();
+		CachingConnectionFactory connectionFactory = rabbitTestSupport.getResource();
 		RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
 		for (int i = 0; i < 5; i++) {
 			String queue1Name = AbstractBinder.applyPrefix(BINDER_PREFIX,
@@ -120,7 +122,7 @@ public class RabbitBinderCleanerTests {
 						new DefaultConsumer(channel));
 				try {
 					waitForConsumerStateNot(queueName, 0);
-					cleaner.clean(stream1, false);
+					doClean(cleaner, stream1, false);
 					fail("Expected exception");
 				}
 				catch (RabbitAdminException e) {
@@ -130,7 +132,7 @@ public class RabbitBinderCleanerTests {
 				channel.basicCancel(consumerTag);
 				waitForConsumerStateNot(queueName, 1);
 				try {
-					cleaner.clean(stream1, false);
+					doClean(cleaner, stream1, false);
 					fail("Expected exception");
 				}
 				catch (RabbitAdminException e) {
@@ -157,7 +159,7 @@ public class RabbitBinderCleanerTests {
 		rabbitAdmin.declareExchange(topic1);
 		rabbitAdmin.deleteQueue(foreignQueue);
 		connectionFactory.destroy();
-		Map<String, List<String>> cleanedMap = cleaner.clean(stream1, false);
+		Map<String, List<String>> cleanedMap = doClean(cleaner, stream1, false);
 		assertThat(cleanedMap).hasSize(2);
 		List<String> cleanedQueues = cleanedMap.get("queues");
 		// should *not* clean stream2
@@ -172,7 +174,7 @@ public class RabbitBinderCleanerTests {
 		assertThat(cleanedExchanges).hasSize(6);
 
 		// wild card *should* clean stream2
-		cleanedMap = cleaner.clean(stream1 + "*", false);
+		cleanedMap = doClean(cleaner, stream1 + "*", false);
 		assertThat(cleanedMap).hasSize(2);
 		cleanedQueues = cleanedMap.get("queues");
 		assertThat(cleanedQueues).hasSize(5);
@@ -184,34 +186,8 @@ public class RabbitBinderCleanerTests {
 		assertThat(cleanedExchanges).hasSize(6);
 	}
 
-	public static class AmqpQueue {
-
-		private boolean autoDelete;
-
-		private boolean durable;
-
-		public AmqpQueue(boolean autoDelete, boolean durable) {
-			this.autoDelete = autoDelete;
-			this.durable = durable;
-		}
-
-		@JsonProperty("auto_delete")
-		protected boolean isAutoDelete() {
-			return autoDelete;
-		}
-
-		protected void setAutoDelete(boolean autoDelete) {
-			this.autoDelete = autoDelete;
-		}
-
-		protected boolean isDurable() {
-			return durable;
-		}
-
-		protected void setDurable(boolean durable) {
-			this.durable = durable;
-		}
-
+	private static Map<String, List<String>> doClean(RabbitBindingCleaner cleaner, String entity, boolean isJob) {
+		return cleaner.clean(RABBITMQ.getHttpUrl() + "/api", "guest", "guest", "/", BINDER_PREFIX, entity, isJob);
 	}
 
 }

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java

@@ -49,6 +49,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.mockito.ArgumentCaptor;
+import org.testcontainers.containers.RabbitMQContainer;
 
 import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
@@ -147,30 +148,30 @@ import static org.mockito.Mockito.when;
  * @author Gary Russell
  * @author David Turanski
  * @author Artem Bilan
+ * @author Chris Bono
  */
 // @checkstyle:off
 public class RabbitBinderTests extends
 		PartitionCapableBinderTests<RabbitTestBinder, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
 
-	private final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class
-			.getSimpleName();
+	private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
 
-	public static final String TEST_PREFIX = "bindertest.";
+	private static final String CLASS_UNDER_TEST_NAME = RabbitMessageChannelBinder.class.getSimpleName();
 
-	private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000])
-			.replaceAll("\u0000", "x");
-
-	private int maxStackTraceSize;
+	private static final String TEST_PREFIX = "bindertest.";
 
+	private static final String BIG_EXCEPTION_MESSAGE = new String(new byte[10_000]).replaceAll("\u0000", "x");
 
 	@RegisterExtension
-	RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true);
+	private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
+
+	private int maxStackTraceSize;
 
 	@Override
 	protected RabbitTestBinder getBinder() {
 		if (this.testBinder == null) {
 			RabbitProperties rabbitProperties = new RabbitProperties();
-			this.testBinder = new RabbitTestBinder(this.rabbitAvailableRule.getResource(), rabbitProperties);
+			this.testBinder = new RabbitTestBinder(this.rabbitTestSupport.getResource(), rabbitProperties);
 		}
 		return this.testBinder;
 	}
@@ -249,7 +250,7 @@ public class RabbitBinderTests extends
 	@Test
 	public void testProducerErrorChannel(TestInfo testInfo) throws Exception {
 		RabbitTestBinder binder = getBinder();
-		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		CachingConnectionFactory ccf = this.rabbitTestSupport.getResource();
 		ccf.setPublisherReturns(true);
 		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
 		ccf.resetConnection();
@@ -332,7 +333,7 @@ public class RabbitBinderTests extends
 	@Test
 	public void testProducerAckChannel(TestInfo testInfo) throws Exception {
 		RabbitTestBinder binder = getBinder();
-		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		CachingConnectionFactory ccf = this.rabbitTestSupport.getResource();
 		ccf.setPublisherReturns(true);
 		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
 		ccf.resetConnection();
@@ -363,7 +364,7 @@ public class RabbitBinderTests extends
 	@Test
 	public void testProducerConfirmHeader(TestInfo testInfo) throws Exception {
 		RabbitTestBinder binder = getBinder();
-		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		CachingConnectionFactory ccf = this.rabbitTestSupport.getResource();
 		ccf.setPublisherReturns(true);
 		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
 		ccf.resetConnection();
@@ -509,7 +510,7 @@ public class RabbitBinderTests extends
 
 	@Test
 	public void testConsumerPropertiesWithUserInfrastructureNoBind() throws Exception {
-		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource());
 		Queue queue = new Queue("propsUser1.infra");
 		admin.declareQueue(queue);
 		DirectExchange exchange = new DirectExchange("propsUser1");
@@ -532,7 +533,7 @@ public class RabbitBinderTests extends
 		assertThat(container.isRunning()).isTrue();
 		consumerBinding.unbind();
 		assertThat(container.isRunning()).isFalse();
-		Client client = new Client("http://guest:guest@localhost:15672/api/");
+		Client client = new Client(adminUri());
 		List<?> bindings = client.getBindingsBySource("/", exchange.getName());
 		assertThat(bindings.size()).isEqualTo(1);
 	}
@@ -604,7 +605,7 @@ public class RabbitBinderTests extends
 		consumerBinding.unbind();
 		assertThat(container.isRunning()).isFalse();
 		assertThat(container.getQueueNames()[0]).isEqualTo(group);
-		Client client = new Client("http://guest:guest@localhost:15672/api/");
+		Client client = new Client(adminUri());
 		List<BindingInfo> bindings = client.getBindingsBySource("/", "propsUser2");
 		int n = 0;
 		while (n++ < 100 && bindings == null || bindings.size() < 1) {
@@ -632,6 +633,9 @@ public class RabbitBinderTests extends
 		verifyAutoDeclareContextClear(binder);
 	}
 
+	private String adminUri() {
+		return String.format("http://guest:guest@localhost:%d/api", RABBITMQ.getHttpPort());
+	}
 
 	@Test
 	public void testConsumerPropertiesWithUserInfrastructureCustomQueueArgs()
@@ -675,7 +679,7 @@ public class RabbitBinderTests extends
 		SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
 				"messageListenerContainer", SimpleMessageListenerContainer.class);
 		assertThat(container.isRunning()).isTrue();
-		Client client = new Client("http://guest:guest@localhost:15672/api");
+		Client client = new Client(adminUri());
 		List<BindingInfo> bindings = client.getBindingsBySource("/", "propsUser3");
 		int n = 0;
 		while (n++ < 100 && bindings == null || bindings.size() < 1) {
@@ -787,7 +791,7 @@ public class RabbitBinderTests extends
 		consumerBinding.unbind();
 		assertThat(container.isRunning()).isFalse();
 		assertThat(container.getQueueNames()[0]).isEqualTo("propsHeader." + group);
-		Client client = new Client("http://guest:guest@localhost:15672/api/");
+		Client client = new Client(adminUri());
 		List<BindingInfo> bindings = client.getBindingsBySource("/", "propsHeader");
 		int n = 0;
 		while (n++ < 100 && bindings == null || bindings.size() < 1) {
@@ -865,7 +869,7 @@ public class RabbitBinderTests extends
 				producerBinding, "lifecycle.amqpTemplate.connectionFactory",
 				ConnectionFactory.class);
 
-		assertThat(this.rabbitAvailableRule.getResource())
+		assertThat(this.rabbitTestSupport.getResource())
 				.isSameAs(producerConnectionFactory);
 
 		endpoint = extractEndpoint(producerBinding);
@@ -882,7 +886,7 @@ public class RabbitBinderTests extends
 		verifyFooRequestProducer(endpoint);
 		channel.send(new GenericMessage<>("foo"));
 		org.springframework.amqp.core.Message received = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource())
+				this.rabbitTestSupport.getResource())
 						.receive("foo.props.0.prodPropsRequired-0", 10_000);
 		assertThat(received).isNotNull();
 		assertThat(received.getMessageProperties().getReceivedDelay()).isEqualTo(42);
@@ -896,7 +900,7 @@ public class RabbitBinderTests extends
 
 	@Test
 	public void testDurablePubSubWithAutoBindDLQ() throws Exception {
-		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource());
 
 		RabbitTestBinder binder = getBinder();
 
@@ -920,7 +924,7 @@ public class RabbitBinderTests extends
 				"tgroup", moduleInputChannel, consumerProperties);
 
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend(TEST_PREFIX + "durabletest.0", "", "foo");
 
 		int n = 0;
@@ -945,7 +949,7 @@ public class RabbitBinderTests extends
 
 	@Test
 	public void testNonDurablePubSubWithAutoBindDLQ() throws Exception {
-		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource());
 
 		RabbitTestBinder binder = getBinder();
 		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
@@ -1007,7 +1011,7 @@ public class RabbitBinderTests extends
 		assertThat(container.getQueueNames().length).isEqualTo(2);
 
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("", TEST_PREFIX + "dlqtest.default", "foo");
 
 		int n = 0;
@@ -1066,7 +1070,7 @@ public class RabbitBinderTests extends
 		DirectChannel moduleInputChannel = createBindableChannel("input",
 				bindingProperties);
 		moduleInputChannel.setBeanName("dlqTestManual");
-		Client client = new Client("http://guest:guest@localhost:15672/api");
+		Client client = new Client(adminUri());
 		moduleInputChannel.subscribe(new MessageHandler() {
 
 			@Override
@@ -1091,7 +1095,7 @@ public class RabbitBinderTests extends
 				"default", moduleInputChannel, consumerProperties);
 
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("", TEST_PREFIX + "dlqTestManual.default", "foo");
 
 		int n = 0;
@@ -1205,7 +1209,7 @@ public class RabbitBinderTests extends
 		output.send(new GenericMessage<>(1));
 
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.setReceiveTimeout(10000);
 
 		String streamDLQName = "bindertest.partDLQ.0.dlqPartGrp.dlq";
@@ -1356,7 +1360,7 @@ public class RabbitBinderTests extends
 		output.send(new GenericMessage<>(1));
 
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.setReceiveTimeout(10000);
 
 		String streamDLQName = "bindertest.partPubDLQ.0.dlqPartGrp.dlq";
@@ -1470,7 +1474,7 @@ public class RabbitBinderTests extends
 		output.send(new GenericMessage<Integer>(1));
 
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.setReceiveTimeout(10000);
 
 		String streamDLQName = "bindertest.partDLQ.1.dlqPartGrp.dlq";
@@ -1505,7 +1509,7 @@ public class RabbitBinderTests extends
 	@Test
 	public void testAutoBindDLQwithRepublish() throws Exception {
 		this.maxStackTraceSize = RabbitUtils
-				.getMaxFrame(rabbitAvailableRule.getResource()) - 20_000;
+				.getMaxFrame(rabbitTestSupport.getResource()) - 20_000;
 		assertThat(this.maxStackTraceSize).isGreaterThan(0);
 
 		RabbitTestBinder binder = getBinder();
@@ -1540,7 +1544,7 @@ public class RabbitBinderTests extends
 				consumerProperties);
 
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtest.foo", "foo");
 
 		template.setReceiveTimeout(10_000);
@@ -1598,7 +1602,7 @@ public class RabbitBinderTests extends
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
 				"foo.dlqpubtestTx", "foo", moduleInputChannel, consumerProperties);
 
-		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
+		RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource());
 		template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestTx.foo", "foo");
 
 		template.setReceiveTimeout(10_000);
@@ -1625,7 +1629,7 @@ public class RabbitBinderTests extends
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testAutoBindDLQwithRepublishSimpleConfirms() throws Exception {
-		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		CachingConnectionFactory ccf = this.rabbitTestSupport.getResource();
 		ccf.setPublisherReturns(true);
 		ccf.setPublisherConfirmType(ConfirmType.SIMPLE);
 		ccf.resetConnection();
@@ -1650,7 +1654,7 @@ public class RabbitBinderTests extends
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
 				"foo.dlqpubtestSimple", "foo", moduleInputChannel, consumerProperties);
 
-		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
+		RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource());
 		template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestSimple.foo", "foo");
 
 		template.setReceiveTimeout(10_000);
@@ -1676,7 +1680,7 @@ public class RabbitBinderTests extends
 	@SuppressWarnings("unchecked")
 	@Test
 	public void testAutoBindDLQwithRepublishCorrelatedConfirms() throws Exception {
-		CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
+		CachingConnectionFactory ccf = this.rabbitTestSupport.getResource();
 		ccf.setPublisherReturns(true);
 		ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
 		ccf.resetConnection();
@@ -1701,7 +1705,7 @@ public class RabbitBinderTests extends
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer(
 				"foo.dlqpubtestCorrelated", "foo", moduleInputChannel, consumerProperties);
 
-		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
+		RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource());
 		template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestCorrelated.foo", "foo");
 
 		template.setReceiveTimeout(10_000);
@@ -1894,11 +1898,11 @@ public class RabbitBinderTests extends
 		ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
 		Binding<MessageChannel> consumerBinding = binder.bindConsumer("propagate.0",
 				"propagate", input, consumerProperties);
-		RabbitAdmin admin = new RabbitAdmin(rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource());
 		admin.declareQueue(new Queue("propagate"));
 		admin.declareBinding(new org.springframework.amqp.core.Binding("propagate", DestinationType.QUEUE,
 				"propagate.1", "#", null));
-		RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
+		RabbitTemplate template = new RabbitTemplate(this.rabbitTestSupport.getResource());
 		template.convertAndSend("propagate.0.propagate", "foo");
 		output.send(input.receive(10_000));
 		org.springframework.amqp.core.Message received = template.receive("propagate", 10_000);
@@ -1997,7 +2001,7 @@ public class RabbitBinderTests extends
 		Binding<MessageChannel> durableConsumerBinding = binder.bindConsumer("latePubSub",
 				"lateDurableGroup", durablePubSubInputChannel, noDlqConsumerProperties);
 
-		proxy.start();
+		proxy.start(RABBITMQ.getAmqpPort());
 
 		moduleOutputChannel.send(MessageBuilder.withPayload("foo")
 				.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
@@ -2052,7 +2056,7 @@ public class RabbitBinderTests extends
 		proxy.stop();
 		cf.destroy();
 
-		this.rabbitAvailableRule.getResource().destroy();
+		this.rabbitTestSupport.getResource().destroy();
 
 		verifyAutoDeclareContextClear(binder);
 	}
@@ -2071,10 +2075,10 @@ public class RabbitBinderTests extends
 		bf.initializeBean(provisioner, "provisioner");
 		bf.registerSingleton("provisioner", provisioner);
 		context.addApplicationListener(provisioner);
-		RabbitAdmin admin = new RabbitAdmin(rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(rabbitTestSupport.getResource());
 		admin.declareQueue(new Queue("testBadUserDeclarationsFatal"));
 		// reset the connection and configure the "user" admin to auto declare queues...
-		rabbitAvailableRule.getResource().resetConnection();
+		rabbitTestSupport.getResource().resetConnection();
 		bf.initializeBean(admin, "rabbitAdmin");
 		bf.registerSingleton("rabbitAdmin", admin);
 		admin.afterPropertiesSet();
@@ -2113,7 +2117,7 @@ public class RabbitBinderTests extends
 		Binding<MessageChannel> producerBinding = binder.bindProducer("rke", output,
 				producerProperties);
 
-		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource());
 		Queue queue = new AnonymousQueue();
 		TopicExchange exchange = new TopicExchange("rke");
 		org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue)
@@ -2159,7 +2163,7 @@ public class RabbitBinderTests extends
 		Binding<MessageChannel> producerBinding = binder.bindProducer("rke", output,
 				producerProperties);
 
-		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource());
 		Queue queue = new AnonymousQueue();
 		DirectExchange exchange = new DirectExchange("rke");
 		org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue)
@@ -2199,7 +2203,7 @@ public class RabbitBinderTests extends
 		Binding<MessageChannel> producerBinding = binder.bindProducer("rkep", output,
 				producerProperties);
 
-		RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource());
+		RabbitAdmin admin = new RabbitAdmin(this.rabbitTestSupport.getResource());
 		Queue queue = new AnonymousQueue();
 		TopicExchange exchange = new TopicExchange("rkep");
 		org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue)
@@ -2243,7 +2247,7 @@ public class RabbitBinderTests extends
 		Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer(
 				"pollable", "group", inboundBindTarget, createConsumerProperties());
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("pollable.group", "testPollable");
 		boolean polled = inboundBindTarget.poll(m -> {
 			assertThat(m.getPayload()).isEqualTo("testPollable");
@@ -2270,7 +2274,7 @@ public class RabbitBinderTests extends
 		Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer(
 				"pollableRequeue", "group", inboundBindTarget, properties);
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("pollableRequeue.group", "testPollable");
 		try {
 			boolean polled = false;
@@ -2307,7 +2311,7 @@ public class RabbitBinderTests extends
 		Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer(
 				"pollableDlq", "group", inboundBindTarget, properties);
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("pollableDlq.group", "testPollable");
 		try {
 			int n = 0;
@@ -2344,7 +2348,7 @@ public class RabbitBinderTests extends
 		Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer(
 				"pollableDlqNoRetry", "group", inboundBindTarget, properties);
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("pollableDlqNoRetry.group", "testPollable");
 		try {
 			int n = 0;
@@ -2380,7 +2384,7 @@ public class RabbitBinderTests extends
 		Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer(
 				"pollableDlqRePub", "group", inboundBindTarget, properties);
 		RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.convertAndSend("pollableDlqRePub.group", "testPollable");
 		boolean polled = false;
 		int n = 0;
@@ -2512,7 +2516,7 @@ public class RabbitBinderTests extends
 	@Override
 	public Spy spyOn(final String queue) {
 		final RabbitTemplate template = new RabbitTemplate(
-				this.rabbitAvailableRule.getResource());
+				this.rabbitTestSupport.getResource());
 		template.setAfterReceivePostProcessors(
 				new DelegatingDecompressingPostProcessor());
 		return new Spy() {

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestContainer.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2022-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.
+ * 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.cloud.stream.binder.rabbit;
+
+import java.time.Duration;
+
+import org.testcontainers.containers.RabbitMQContainer;
+
+/**
+ * Provides a static {@link RabbitMQContainer} that can be shared across test classes.
+ *
+ * @author Chris Bono
+ */
+public class RabbitTestContainer {
+
+	private static final RabbitMQContainer RABBITMQ;
+	static {
+		String image = "rabbitmq:management";
+		String cache = System.getenv().get("IMAGE_CACHE");
+		if (cache != null) {
+			image = cache + image;
+		}
+		RABBITMQ = new RabbitMQContainer(image)
+			.withExposedPorts(5672, 15672, 5552)
+			.withEnv("RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS", "-rabbitmq_stream advertised_host localhost")
+			.withPluginsEnabled("rabbitmq_stream")
+			.withStartupTimeout(Duration.ofMinutes(2));
+		RABBITMQ.start();
+	}
+
+	/**
+	 * Should be called early by test that wants to ensure a shared {@link RabbitMQContainer} is up and running.
+	 */
+	public static RabbitMQContainer sharedInstance() {
+		return RABBITMQ;
+	}
+
+}

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RepublishUnitTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2020 the original author or authors.
+ * Copyright 2020-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.
@@ -20,7 +20,7 @@ import java.io.IOException;
 import java.util.Collections;
 
 import com.rabbitmq.client.Channel;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
@@ -62,7 +62,7 @@ public class RepublishUnitTests {
 		RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(cf, props, null);
 		RabbitConsumerProperties extension = new RabbitConsumerProperties();
 		ExtendedConsumerProperties<RabbitConsumerProperties> bindingProps =
-				new ExtendedConsumerProperties<RabbitConsumerProperties>(extension);
+				new ExtendedConsumerProperties<>(extension);
 		MessageHandler handler = binder.getErrorMessageHandler(mock(ConsumerDestination.class), "foo", bindingProps);
 		ErrorMessage message = new ErrorMessage(new RuntimeException("test"),
 				Collections.singletonMap(IntegrationMessageHeaderAccessor.SOURCE_DATA,

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-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.
@@ -32,6 +32,7 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.mockito.Mockito;
+import org.testcontainers.containers.RabbitMQContainer;
 
 import org.springframework.amqp.core.DeclarableCustomizer;
 import org.springframework.amqp.core.ExchangeTypes;
@@ -58,6 +59,7 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
 import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
 import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
 import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
+import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
 import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
 import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport;
@@ -89,17 +91,19 @@ import static org.mockito.Mockito.verify;
  * @author Gary Russell
  * @author Artem Bilan
  * @author Soby Chacko
+ * @author Chris Bono
  */
 public class RabbitBinderModuleTests {
 
+	private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
+
+	private static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(ConnectionFactory.class, Mockito.RETURNS_MOCKS);
+
 	@RegisterExtension
-	public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport();
+	private RabbitTestSupport rabbitTestSupport = new RabbitTestSupport(true, RABBITMQ.getAmqpPort(), RABBITMQ.getHttpPort());
 
 	private ConfigurableApplicationContext context;
 
-	public static final ConnectionFactory MOCK_CONNECTION_FACTORY = mock(
-			ConnectionFactory.class, Mockito.RETURNS_MOCKS);
-
 	@AfterEach
 	public void tearDown() {
 		if (context != null) {
@@ -114,47 +118,48 @@ public class RabbitBinderModuleTests {
 	@Test
 	public void testParentConnectionFactoryInheritedByDefault() throws Exception {
 		context = new SpringApplicationBuilder(SimpleProcessor.class)
-				.web(WebApplicationType.NONE).run("--server.port=0",
-						"--spring.cloud.function.definition=process",
-						"--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo",
-						"--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.single-active-consumer=true");
+			.web(WebApplicationType.NONE).run("--server.port=0",
+				"--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort(),
+				"--spring.cloud.function.definition=process",
+				"--spring.cloud.stream.rabbit.binder.connection-name-prefix=foo",
+				"--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.single-active-consumer=true");
 		BinderFactory binderFactory = context.getBean(BinderFactory.class);
 		Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
 		assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
 		DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
 		CachingConnectionFactory binderConnectionFactory = (CachingConnectionFactory) binderFieldAccessor
-				.getPropertyValue("connectionFactory");
+			.getPropertyValue("connectionFactory");
 		assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
 		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
 		assertThat(binderConnectionFactory).isSameAs(connectionFactory);
 
 		CompositeHealthContributor bindersHealthIndicator = context
-				.getBean("bindersHealthContributor", CompositeHealthContributor.class);
+			.getBean("bindersHealthContributor", CompositeHealthContributor.class);
 
 		assertThat(bindersHealthIndicator).isNotNull();
 
 		RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("rabbit");
 		assertThat(indicator).isNotNull();
 		assertThat(indicator.health().getStatus())
-				.isEqualTo(Status.UP);
+			.isEqualTo(Status.UP);
 
 		ConnectionFactory publisherConnectionFactory = binderConnectionFactory
-				.getPublisherConnectionFactory();
+			.getPublisherConnectionFactory();
 		assertThat(TestUtils.getPropertyValue(publisherConnectionFactory,
-				"connection.target")).isNull();
+			"connection.target")).isNull();
 		DirectChannel checkPf = new DirectChannel();
 		Binding<MessageChannel> binding = ((RabbitMessageChannelBinder) binder)
-				.bindProducer("checkPF", checkPf,
-						new ExtendedProducerProperties<>(new RabbitProducerProperties()));
+			.bindProducer("checkPF", checkPf,
+				new ExtendedProducerProperties<>(new RabbitProducerProperties()));
 		checkPf.send(new GenericMessage<>("foo".getBytes()));
 		binding.unbind();
 		assertThat(TestUtils.getPropertyValue(publisherConnectionFactory,
-				"connection.target")).isNotNull();
+			"connection.target")).isNotNull();
 
 		CachingConnectionFactory cf = this.context
-				.getBean(CachingConnectionFactory.class);
+			.getBean(CachingConnectionFactory.class);
 		ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf,
-				"connectionNameStrategy", ConnectionNameStrategy.class);
+			"connectionNameStrategy", ConnectionNameStrategy.class);
 		assertThat(cns.obtainNewConnectionName(cf)).isEqualTo("foo#2");
 		new RabbitAdmin(rabbitTestSupport.getResource()).deleteExchange("checkPF");
 		checkCustomizedArgs();
@@ -164,7 +169,7 @@ public class RabbitBinderModuleTests {
 	}
 
 	private void checkCustomizedArgs() throws MalformedURLException, URISyntaxException, InterruptedException {
-		Client client = new Client("http://guest:guest@localhost:15672/api");
+		Client client = new Client(String.format("http://guest:guest@localhost:%d/api", RABBITMQ.getHttpPort()));
 		List<BindingInfo> bindings = client.getBindingsBySource("/", "process-in-0");
 		int n = 0;
 		while (n++ < 100 && bindings == null || bindings.size() < 1) {
@@ -184,59 +189,60 @@ public class RabbitBinderModuleTests {
 	@SuppressWarnings("unchecked")
 	public void testParentConnectionFactoryInheritedByDefaultAndRabbitSettingsPropagated() {
 		context = new SpringApplicationBuilder(SimpleProcessor.class)
-				.web(WebApplicationType.NONE).run("--server.port=0",
-						"--spring.cloud.function.definition=process",
-						"--spring.cloud.stream.bindings.source.group=someGroup",
-						"--spring.cloud.stream.bindings.process-in-0.group=someGroup",
-						"--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.transacted=true",
-						"--spring.cloud.stream.rabbit.bindings.process-out-0.producer.transacted=true");
+			.web(WebApplicationType.NONE).run("--server.port=0",
+				"--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort(),
+				"--spring.cloud.function.definition=process",
+				"--spring.cloud.stream.bindings.source.group=someGroup",
+				"--spring.cloud.stream.bindings.process-in-0.group=someGroup",
+				"--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.transacted=true",
+				"--spring.cloud.stream.rabbit.bindings.process-out-0.producer.transacted=true");
 		BinderFactory binderFactory = context.getBean(BinderFactory.class);
 		Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
 		assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
 		BindingService bindingService = context.getBean(BindingService.class);
 		DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(
-				bindingService);
+			bindingService);
 		// @checkstyle:off
 		Map<String, List<Binding<MessageChannel>>> consumerBindings = (Map<String, List<Binding<MessageChannel>>>) channelBindingServiceAccessor
-				.getPropertyValue("consumerBindings");
+			.getPropertyValue("consumerBindings");
 		// @checkstyle:on
 		Binding<MessageChannel> inputBinding = consumerBindings.get("process-in-0").get(0);
 		assertThat(TestUtils.getPropertyValue(inputBinding, "lifecycle.beanName"))
-				.isEqualTo("setByCustomizer:someGroup");
+			.isEqualTo("setByCustomizer:someGroup");
 		SimpleMessageListenerContainer container = TestUtils.getPropertyValue(
-				inputBinding, "lifecycle.messageListenerContainer",
-				SimpleMessageListenerContainer.class);
+			inputBinding, "lifecycle.messageListenerContainer",
+			SimpleMessageListenerContainer.class);
 		assertThat(TestUtils.getPropertyValue(container, "beanName"))
-				.isEqualTo("setByCustomizerForQueue:process-in-0.someGroup,andGroup:someGroup");
+			.isEqualTo("setByCustomizerForQueue:process-in-0.someGroup,andGroup:someGroup");
 		assertThat(TestUtils.getPropertyValue(container, "transactional", Boolean.class))
-				.isTrue();
+			.isTrue();
 		Map<String, Binding<MessageChannel>> producerBindings = (Map<String, Binding<MessageChannel>>) TestUtils
-				.getPropertyValue(bindingService, "producerBindings");
+			.getPropertyValue(bindingService, "producerBindings");
 		Binding<MessageChannel> outputBinding = producerBindings.get("process-out-0");
 		assertThat(TestUtils.getPropertyValue(outputBinding,
-				"lifecycle.amqpTemplate.transactional", Boolean.class)).isTrue();
+			"lifecycle.amqpTemplate.transactional", Boolean.class)).isTrue();
 		assertThat(TestUtils.getPropertyValue(outputBinding, "lifecycle.beanName"))
-				.isEqualTo("setByCustomizer:process-out-0");
+			.isEqualTo("setByCustomizer:process-out-0");
 		DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
 		ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
-				.getPropertyValue("connectionFactory");
+			.getPropertyValue("connectionFactory");
 		assertThat(binderConnectionFactory).isInstanceOf(CachingConnectionFactory.class);
 		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
 		assertThat(binderConnectionFactory).isSameAs(connectionFactory);
 		CompositeHealthContributor bindersHealthIndicator = context
-				.getBean("bindersHealthContributor", CompositeHealthContributor.class);
+			.getBean("bindersHealthContributor", CompositeHealthContributor.class);
 
 		assertThat(bindersHealthIndicator).isNotNull();
 
 		RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("rabbit");
 		assertThat(indicator).isNotNull();
 		assertThat(indicator.health().getStatus())
-				.isEqualTo(Status.UP);
+			.isEqualTo(Status.UP);
 
 		CachingConnectionFactory cf = this.context
-				.getBean(CachingConnectionFactory.class);
+			.getBean(CachingConnectionFactory.class);
 		ConnectionNameStrategy cns = TestUtils.getPropertyValue(cf,
-				"connectionNameStrategy", ConnectionNameStrategy.class);
+			"connectionNameStrategy", ConnectionNameStrategy.class);
 		assertThat(cns.obtainNewConnectionName(cf)).startsWith("rabbitConnectionFactory");
 //		assertThat(TestUtils.getPropertyValue(consumerBindings.get("source").get(0),
 //				"target.source.h.advised.targetSource.target.beanName"))
@@ -246,25 +252,25 @@ public class RabbitBinderModuleTests {
 	@Test
 	public void testParentConnectionFactoryInheritedIfOverridden() {
 		context = new SpringApplicationBuilder(SimpleProcessor.class,
-				ConnectionFactoryConfiguration.class).web(WebApplicationType.NONE)
-						.run("--server.port=0");
+			ConnectionFactoryConfiguration.class).web(WebApplicationType.NONE)
+			.run("--server.port=0", "--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort());
 		BinderFactory binderFactory = context.getBean(BinderFactory.class);
 		Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
 		assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
 		DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
 		ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
-				.getPropertyValue("connectionFactory");
+			.getPropertyValue("connectionFactory");
 		assertThat(binderConnectionFactory).isSameAs(MOCK_CONNECTION_FACTORY);
 		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
 		assertThat(binderConnectionFactory).isSameAs(connectionFactory);
 		CompositeHealthContributor bindersHealthIndicator = context
-				.getBean("bindersHealthContributor", CompositeHealthContributor.class);
+			.getBean("bindersHealthContributor", CompositeHealthContributor.class);
 		assertThat(bindersHealthIndicator).isNotNull();
 		RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("rabbit");
 		assertThat(indicator).isNotNull();
 		// mock connection factory behaves as if down
 		assertThat(indicator.health().getStatus())
-				.isEqualTo(Status.DOWN);
+			.isEqualTo(Status.DOWN);
 	}
 
 	@Test
@@ -276,43 +282,44 @@ public class RabbitBinderModuleTests {
 		params.add("--spring.cloud.stream.binders.custom.type=rabbit");
 		params.add("--spring.cloud.stream.binders.custom.environment.foo=bar");
 		params.add("--server.port=0");
+		params.add("--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort());
 		params.add("--spring.rabbitmq.template.retry.enabled=true");
 		params.add("--spring.rabbitmq.template.retry.maxAttempts=2");
 		params.add("--spring.rabbitmq.template.retry.initial-interval=1000");
 		params.add("--spring.rabbitmq.template.retry.multiplier=1.1");
 		params.add("--spring.rabbitmq.template.retry.max-interval=3000");
 		context = new SpringApplicationBuilder(SimpleProcessor.class)
-				.web(WebApplicationType.NONE)
-				.run(params.toArray(new String[params.size()]));
+			.web(WebApplicationType.NONE)
+			.run(params.toArray(new String[0]));
 		BinderFactory binderFactory = context.getBean(BinderFactory.class);
 		// @checkstyle:off
 		@SuppressWarnings("unchecked")
 		Binder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> binder = (Binder<MessageChannel, ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>>) binderFactory
-				.getBinder(null, MessageChannel.class);
+			.getBinder(null, MessageChannel.class);
 		// @checkstyle:on
 		assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
 		DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
 		ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
-				.getPropertyValue("connectionFactory");
+			.getPropertyValue("connectionFactory");
 		ConnectionFactory connectionFactory = context.getBean(ConnectionFactory.class);
 		assertThat(binderConnectionFactory).isNotSameAs(connectionFactory);
 		CompositeHealthContributor bindersHealthIndicator = context
-				.getBean("bindersHealthContributor", CompositeHealthContributor.class);
-		assertThat(bindersHealthIndicator);
+			.getBean("bindersHealthContributor", CompositeHealthContributor.class);
+		assertThat(bindersHealthIndicator).isNotNull();
 
 		RabbitHealthIndicator indicator = (RabbitHealthIndicator) bindersHealthIndicator.getContributor("custom");
 		assertThat(indicator).isNotNull();
 		assertThat(indicator.health().getStatus()).isEqualTo(Status.UP);
 		String name = UUID.randomUUID().toString();
 		Binding<MessageChannel> binding = binder.bindProducer(name, new DirectChannel(),
-				new ExtendedProducerProperties<>(new RabbitProducerProperties()));
+			new ExtendedProducerProperties<>(new RabbitProducerProperties()));
 		RetryTemplate template = TestUtils.getPropertyValue(binding,
-				"lifecycle.amqpTemplate.retryTemplate", RetryTemplate.class);
+			"lifecycle.amqpTemplate.retryTemplate", RetryTemplate.class);
 		assertThat(template).isNotNull();
 		SimpleRetryPolicy retryPolicy = TestUtils.getPropertyValue(template,
-				"retryPolicy", SimpleRetryPolicy.class);
+			"retryPolicy", SimpleRetryPolicy.class);
 		ExponentialBackOffPolicy backOff = TestUtils.getPropertyValue(template,
-				"backOffPolicy", ExponentialBackOffPolicy.class);
+			"backOffPolicy", ExponentialBackOffPolicy.class);
 		assertThat(retryPolicy.getMaxAttempts()).isEqualTo(2);
 		assertThat(backOff.getInitialInterval()).isEqualTo(1000L);
 		assertThat(backOff.getMultiplier()).isEqualTo(1.1);
@@ -325,23 +332,23 @@ public class RabbitBinderModuleTests {
 	@Test
 	public void testCloudProfile() {
 		this.context = new SpringApplicationBuilder(SimpleProcessor.class,
-				MockCloudConfiguration.class).web(WebApplicationType.NONE)
-						.profiles("cloud").run();
+			MockCloudConfiguration.class).web(WebApplicationType.NONE)
+			.profiles("cloud").run();
 		BinderFactory binderFactory = this.context.getBean(BinderFactory.class);
 		Binder<?, ?, ?> binder = binderFactory.getBinder(null, MessageChannel.class);
 		assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class);
 		DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder);
 		ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor
-				.getPropertyValue("connectionFactory");
+			.getPropertyValue("connectionFactory");
 		ConnectionFactory connectionFactory = this.context
-				.getBean(ConnectionFactory.class);
+			.getBean(ConnectionFactory.class);
 
 		assertThat(binderConnectionFactory).isNotSameAs(connectionFactory);
 
 		assertThat(TestUtils.getPropertyValue(connectionFactory, "addresses"))
-				.isNotNull();
+			.isNotNull();
 		assertThat(TestUtils.getPropertyValue(binderConnectionFactory, "addresses"))
-				.isNull();
+			.isNull();
 
 		Cloud cloud = this.context.getBean(Cloud.class);
 
@@ -351,30 +358,31 @@ public class RabbitBinderModuleTests {
 	@Test
 	public void testExtendedProperties() {
 		context = new SpringApplicationBuilder(SimpleProcessor.class)
-				.web(WebApplicationType.NONE).run("--server.port=0",
-						"--spring.cloud.function.definition=process",
-						"--spring.cloud.stream.rabbit.default.producer.routing-key-expression=fooRoutingKey",
-						"--spring.cloud.stream.rabbit.default.consumer.exchange-type=direct",
-						"--spring.cloud.stream.rabbit.bindings.process-out-0.producer.batch-size=512",
-						"--spring.cloud.stream.rabbit.default.consumer.max-concurrency=4",
-						"--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.exchange-type=fanout");
+			.web(WebApplicationType.NONE).run("--server.port=0",
+				"--spring.rabbitmq.port=" + RABBITMQ.getAmqpPort(),
+				"--spring.cloud.function.definition=process",
+				"--spring.cloud.stream.rabbit.default.producer.routing-key-expression=fooRoutingKey",
+				"--spring.cloud.stream.rabbit.default.consumer.exchange-type=direct",
+				"--spring.cloud.stream.rabbit.bindings.process-out-0.producer.batch-size=512",
+				"--spring.cloud.stream.rabbit.default.consumer.max-concurrency=4",
+				"--spring.cloud.stream.rabbit.bindings.process-in-0.consumer.exchange-type=fanout");
 		BinderFactory binderFactory = context.getBean(BinderFactory.class);
 		Binder<?, ?, ?> rabbitBinder = binderFactory.getBinder(null,
-				MessageChannel.class);
+			MessageChannel.class);
 
 		RabbitProducerProperties rabbitProducerProperties = (RabbitProducerProperties) ((ExtendedPropertiesBinder) rabbitBinder)
-				.getExtendedProducerProperties("process-out-0");
+			.getExtendedProducerProperties("process-out-0");
 
 		assertThat(
-				rabbitProducerProperties.getRoutingKeyExpression().getExpressionString())
-						.isEqualTo("fooRoutingKey");
+			rabbitProducerProperties.getRoutingKeyExpression().getExpressionString())
+			.isEqualTo("fooRoutingKey");
 		assertThat(rabbitProducerProperties.getBatchSize()).isEqualTo(512);
 
 		RabbitConsumerProperties rabbitConsumerProperties = (RabbitConsumerProperties) ((ExtendedPropertiesBinder) rabbitBinder)
-				.getExtendedConsumerProperties("process-in-0");
+			.getExtendedConsumerProperties("process-in-0");
 
 		assertThat(rabbitConsumerProperties.getExchangeType())
-				.isEqualTo(ExchangeTypes.FANOUT);
+			.isEqualTo(ExchangeTypes.FANOUT);
 		assertThat(rabbitConsumerProperties.getMaxConcurrency()).isEqualTo(4);
 	}
 
@@ -384,7 +392,7 @@ public class RabbitBinderModuleTests {
 		@Bean
 		public ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer() {
 			return (c, q, g) -> ((AbstractMessageListenerContainer) c).setBeanName(
-					"setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g));
+				"setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g));
 		}
 
 		@Bean
@@ -432,7 +440,7 @@ public class RabbitBinderModuleTests {
 			Cloud cloud = mock(Cloud.class);
 
 			willReturn(new CachingConnectionFactory("localhost")).given(cloud)
-					.getSingletonServiceConnector(ConnectionFactory.class, null);
+				.getSingletonServiceConnector(ConnectionFactory.class, null);
 
 			return cloud;
 		}

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/AbstractIntegrationTests.java

@@ -1,62 +0,0 @@
-/*
- * Copyright 2021-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.
- * 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.cloud.stream.binder.rabbit.stream;
-
-import java.time.Duration;
-
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.utility.DockerImageName;
-
-/**
- * @author Gary Russell
- * @since 3.2
- *
- */
-public abstract class AbstractIntegrationTests {
-
-	static final GenericContainer<?> RABBITMQ;
-
-	static {
-		if (System.getProperty("spring.rabbit.use.local.server") == null) {
-			String image = "pivotalrabbitmq/rabbitmq-stream";
-			String cache = System.getenv().get("IMAGE_CACHE");
-			if (cache != null) {
-				image = cache + image;
-			}
-			RABBITMQ = new GenericContainer<>(DockerImageName.parse(image))
-						.withExposedPorts(5672, 15672, 5552)
-						.withStartupTimeout(Duration.ofMinutes(2));
-			RABBITMQ.start();
-		}
-		else {
-			RABBITMQ = null;
-		}
-	}
-
-	static int amqpPort() {
-		return RABBITMQ != null ? RABBITMQ.getMappedPort(5672) : 5672;
-	}
-
-	static int managementPort() {
-		return RABBITMQ != null ? RABBITMQ.getMappedPort(15672) : 15672;
-	}
-
-	static int streamPort() {
-		return RABBITMQ != null ? RABBITMQ.getMappedPort(5552) : 5552;
-	}
-
-}

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2021 the original author or authors.
+ * Copyright 2021-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.
@@ -26,8 +26,10 @@ import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.OffsetSpecification;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.RabbitMQContainer;
 
 import org.springframework.cloud.stream.binder.rabbit.RabbitStreamMessageHandler;
+import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.rabbit.stream.producer.RabbitStreamTemplate;
 
@@ -35,16 +37,18 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 /**
  * @author Gary Russell
+ * @author Chris Bono
  * @since 3.2
- *
  */
-public class RabbitStreamMessageHandlerTests extends AbstractIntegrationTests {
+public class RabbitStreamMessageHandlerTests {
+
+	private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance();
 
 	@Test
 	void convertAndSend() throws InterruptedException {
 		Environment env = Environment.builder()
 				.lazyInitialization(true)
-				.addressResolver(add -> new Address("localhost", streamPort()))
+				.addressResolver(add -> new Address("localhost", RABBITMQ.getMappedPort(5552)))
 				.build();
 		try {
 			env.deleteStream("stream.stream");

bom/spring-cloud-starter-parent/pom.xml

@@ -7,6 +7,7 @@
 		<groupId>org.springframework.boot</groupId>
 		<artifactId>spring-boot-starter-parent</artifactId>
 		<version>3.0.0-M2</version>
+		<relativePath/>
 	</parent>
 	<groupId>org.springframework.cloud</groupId>
 	<artifactId>spring-cloud-stream-starter-parent</artifactId>