GH-1439: Fix Memory Leak with Misconfiguration

Resolves #1439

Do not store pending confirms/returns if `RabbitTemplate` has confirms
enabled but the factory does not support confirms.

* Test polishing; include a returned message.

**cherry-pick to `2.4.x` & `2.3.x`**

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java

@@ -2437,17 +2437,22 @@ protected void sendToRabbit(Channel channel, String exchange, String routingKey,
 	private void setupConfirm(Channel channel, Message message, @Nullable CorrelationData correlationDataArg) {
 		if ((this.publisherConfirms || this.confirmCallback != null) && channel instanceof PublisherCallbackChannel) {
 
-			PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel;
-			CorrelationData correlationData = this.correlationDataPostProcessor != null
-					? this.correlationDataPostProcessor.postProcess(message, correlationDataArg)
-					: correlationDataArg;
 			long nextPublishSeqNo = channel.getNextPublishSeqNo();
-			message.getMessageProperties().setPublishSequenceNumber(nextPublishSeqNo);
-			publisherCallbackChannel.addPendingConfirm(this, nextPublishSeqNo,
-					new PendingConfirm(correlationData, System.currentTimeMillis()));
-			if (correlationData != null && StringUtils.hasText(correlationData.getId())) {
-				message.getMessageProperties().setHeader(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY,
-						correlationData.getId());
+			if (nextPublishSeqNo > 0) {
+				PublisherCallbackChannel publisherCallbackChannel = (PublisherCallbackChannel) channel;
+				CorrelationData correlationData = this.correlationDataPostProcessor != null
+						? this.correlationDataPostProcessor.postProcess(message, correlationDataArg)
+						: correlationDataArg;
+				message.getMessageProperties().setPublishSequenceNumber(nextPublishSeqNo);
+				publisherCallbackChannel.addPendingConfirm(this, nextPublishSeqNo,
+						new PendingConfirm(correlationData, System.currentTimeMillis()));
+				if (correlationData != null && StringUtils.hasText(correlationData.getId())) {
+					message.getMessageProperties().setHeader(PublisherCallbackChannel.RETURNED_MESSAGE_CORRELATION_KEY,
+							correlationData.getId());
+				}
+			}
+			else {
+				logger.debug("Factory does not have confirms enabled");
 			}
 		}
 		else if (channel instanceof ChannelProxy && ((ChannelProxy) channel).isConfirmSelected()) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-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.
@@ -48,6 +48,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.commons.logging.Log;
+import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -130,6 +131,7 @@ public void create() {
 		connectionFactoryWithReturnsEnabled.setPort(BrokerTestUtils.getPort());
 		connectionFactoryWithReturnsEnabled.setPublisherReturns(true);
 		templateWithReturnsEnabled = new RabbitTemplate(connectionFactoryWithReturnsEnabled);
+		templateWithReturnsEnabled.setMandatory(true);
 		connectionFactoryWithConfirmsAndReturnsEnabled = new CachingConnectionFactory();
 		connectionFactoryWithConfirmsAndReturnsEnabled.setHost("localhost");
 		connectionFactoryWithConfirmsAndReturnsEnabled.setChannelCacheSize(100);
@@ -320,6 +322,7 @@ public void testPublisherConfirmNotReceived() throws Exception {
 		Connection mockConnection = mock(Connection.class);
 		Channel mockChannel = mock(Channel.class);
 		given(mockChannel.isOpen()).willReturn(true);
+		given(mockChannel.getNextPublishSeqNo()).willReturn(1L);
 
 		given(mockConnectionFactory.newConnection(any(ExecutorService.class), anyString())).willReturn(mockConnection);
 		given(mockConnection.isOpen()).willReturn(true);
@@ -863,4 +866,32 @@ public void testWithFuture() throws Exception {
 		admin.deleteQueue(queue.getName());
 	}
 
+	@Test
+	void justReturns() throws InterruptedException {
+		CorrelationData correlationData = new CorrelationData();
+		CountDownLatch latch = new CountDownLatch(1);
+		this.templateWithReturnsEnabled.setReturnsCallback(returned -> {
+			latch.countDown();
+		});
+		this.templateWithReturnsEnabled.setConfirmCallback((correlationData1, ack, cause) -> {
+			// has callback but factory is not enabled
+		});
+		this.templateWithReturnsEnabled.convertAndSend("", ROUTE, "foo", correlationData);
+		ChannelProxy channel = (ChannelProxy) this.connectionFactoryWithReturnsEnabled.createConnection()
+				.createChannel(false);
+		assertThat(channel.getTargetChannel())
+				.extracting("pendingReturns")
+				.asInstanceOf(InstanceOfAssertFactories.MAP)
+				.isEmpty();
+		assertThat(channel.getTargetChannel())
+				.extracting("pendingConfirms")
+				.asInstanceOf(InstanceOfAssertFactories.MAP)
+				.extracting(map -> map.values().iterator().next())
+				.asInstanceOf(InstanceOfAssertFactories.MAP)
+				.isEmpty();
+
+		this.templateWithReturnsEnabled.convertAndSend("", "___JUNK___", "foo", correlationData);
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+	}
+
 }