Support additional publisher confirm types

See gh-17848

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration.java

@@ -99,8 +99,8 @@ public class RabbitAutoConfiguration {
 			CachingConnectionFactory factory = new CachingConnectionFactory(
 					getRabbitConnectionFactoryBean(properties).getObject());
 			map.from(properties::determineAddresses).to(factory::setAddresses);
-			map.from(properties::isPublisherConfirms).to(factory::setPublisherConfirms);
 			map.from(properties::isPublisherReturns).to(factory::setPublisherReturns);
+			map.from(properties::getPublisherConfirmType).whenNonNull().to(factory::setPublisherConfirmType);
 			RabbitProperties.Cache.Channel channel = properties.getCache().getChannel();
 			map.from(channel::getSize).whenNonNull().to(factory::setChannelCacheSize);
 			map.from(channel::getCheckoutTimeout).whenNonNull().as(Duration::toMillis)

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitProperties.java

@@ -23,6 +23,7 @@ import java.util.List;
 
 import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.CacheMode;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.boot.context.properties.DeprecatedConfigurationProperty;
 import org.springframework.boot.convert.DurationUnit;
@@ -96,6 +97,11 @@ public class RabbitProperties {
 	 */
 	private boolean publisherReturns;
 
+	/**
+	 * The type of publisher confirms to use.
+	 */
+	private ConfirmType publisherConfirmType;
+
 	/**
 	 * Connection timeout. Set it to zero to wait forever.
 	 */
@@ -275,11 +281,16 @@ public class RabbitProperties {
 	}
 
 	public boolean isPublisherConfirms() {
-		return this.publisherConfirms;
+		return this.publisherConfirmType.equals(ConfirmType.CORRELATED);
 	}
 
+	@Deprecated
 	public void setPublisherConfirms(boolean publisherConfirms) {
-		this.publisherConfirms = publisherConfirms;
+		if (publisherConfirms) {
+			this.publisherConfirmType = ConfirmType.CORRELATED;
+			return;
+		}
+		this.publisherConfirmType = ConfirmType.NONE;
 	}
 
 	public boolean isPublisherReturns() {
@@ -294,6 +305,14 @@ public class RabbitProperties {
 		return this.connectionTimeout;
 	}
 
+	public void setPublisherConfirmType(ConfirmType publisherConfirmType) {
+		this.publisherConfirmType = publisherConfirmType;
+	}
+
+	public ConfirmType getPublisherConfirmType() {
+		return this.publisherConfirmType;
+	}
+
 	public void setConnectionTimeout(Duration connectionTimeout) {
 		this.connectionTimeout = connectionTimeout;
 	}

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfigurationTests.java

@@ -210,6 +210,34 @@ class RabbitAutoConfigurationTests {
 				});
 	}
 
+	@Test
+	void testConnectionFactorPublisherSettingsUsingConfirmType() {
+		this.contextRunner.withUserConfiguration(TestConfiguration.class)
+				.withPropertyValues("spring.rabbitmq.publisher-confirm-type=correlated",
+						"spring.rabbitmq.publisher-returns=true")
+				.run((context) -> {
+					CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
+					RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
+					assertThat(connectionFactory.isPublisherConfirms()).isTrue();
+					assertThat(connectionFactory.isPublisherReturns()).isTrue();
+					assertThat(getMandatory(rabbitTemplate)).isTrue();
+				});
+	}
+
+	@Test
+	void testConnectionFactorySimplePublisherSettingsUsingConfirmType() {
+		this.contextRunner.withUserConfiguration(TestConfiguration.class)
+				.withPropertyValues("spring.rabbitmq.publisher-confirm-type=simple",
+						"spring.rabbitmq.publisher-returns=true")
+				.run((context) -> {
+					CachingConnectionFactory connectionFactory = context.getBean(CachingConnectionFactory.class);
+					RabbitTemplate rabbitTemplate = context.getBean(RabbitTemplate.class);
+					assertThat(connectionFactory.isSimplePublisherConfirms()).isTrue();
+					assertThat(connectionFactory.isPublisherReturns()).isTrue();
+					assertThat(getMandatory(rabbitTemplate)).isTrue();
+				});
+	}
+
 	@Test
 	void testRabbitTemplateMessageConverters() {
 		this.contextRunner.withUserConfiguration(MessageConvertersConfiguration.class).run((context) -> {

spring-boot-project/spring-boot-dependencies/pom.xml

@@ -182,7 +182,7 @@
 		<slf4j.version>1.7.26</slf4j.version>
 		<snakeyaml.version>1.24</snakeyaml.version>
 		<solr.version>8.2.0</solr.version>
-		<spring-amqp.version>2.2.0.M4</spring-amqp.version>
+		<spring-amqp.version>2.2.0.BUILD-SNAPSHOT</spring-amqp.version>
 		<spring-batch.version>4.2.0.BUILD-SNAPSHOT</spring-batch.version>
 		<spring-cloud-connectors.version>2.0.6.RELEASE</spring-cloud-connectors.version>
 		<spring-data-releasetrain.version>Moore-BUILD-SNAPSHOT</spring-data-releasetrain.version>