GH-11194: Use simple confirms in AmqpOutboundEndpoint

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java

@@ -48,6 +48,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Ngoc Nhan
+ * @author Jun Cho
  *
  * @since 2.1
  */
@@ -95,6 +96,12 @@ public void setExpectReply(boolean expectReply) {
 	 * received within the confirm timeout or a negative acknowledgment or returned
 	 * message is received, an exception will be thrown. Does not apply to the gateway
 	 * since it blocks awaiting the reply.
+	 * <p>
+	 * When there is no correlation data for the message and the connection factory is
+	 * configured for simple publisher confirms, the message is sent within the scope of a
+	 * {@code RabbitTemplate.invoke()} operation and {@code waitForConfirmsOrDie()} is used
+	 * instead. A returned message does not fail the send in that case; it is published to
+	 * the return channel, if one is configured.
 	 * @param waitForConfirm true to block until the confirmation or timeout is received.
 	 * @since 5.2
 	 * @see #setConfirmTimeout(long)
@@ -156,6 +163,10 @@ protected void endpointInit() {
 						+ "does not support returned messages; none will be received");
 			}
 		}
+		if (this.waitForConfirm && !usesSimplePublisherConfirms() && !usesCorrelatedPublisherConfirms()) {
+			this.logger.warn("The 'waitForConfirm' is set to true but the underlying connection factory "
+					+ "does not support publisher confirms; no confirmations will be awaited");
+		}
 		Duration confirmTimeout = getConfirmTimeout();
 		if (confirmTimeout != null) {
 			this.waitForConfirmTimeout = confirmTimeout;
@@ -181,6 +192,10 @@ protected void doStop() {
 			multiSend(requestMessage, exchangeName, routingKey);
 			return null;
 		}
+		else if (this.waitForConfirm && correlationData == null && usesSimplePublisherConfirms()) {
+			sendAndWaitForConfirms(exchangeName, routingKey, requestMessage);
+			return null;
+		}
 		else {
 			send(exchangeName, routingKey, requestMessage, correlationData);
 			if (this.waitForConfirm && correlationData != null) {
@@ -208,6 +223,28 @@ private void multiSend(Message<?> requestMessage, @Nullable String exchangeName,
 		});
 	}
 
+	private void sendAndWaitForConfirms(@Nullable String exchangeName, @Nullable String routingKey,
+			Message<?> requestMessage) {
+
+		RabbitTemplate rabbitTemplateToUse = this.rabbitTemplate;
+		Assert.notNull(rabbitTemplateToUse, "The 'RabbitTemplate' must be provided to wait for confirms.");
+		rabbitTemplateToUse.<@Nullable Object>invoke(template -> {
+			doRabbitSend(exchangeName, routingKey, requestMessage, null, rabbitTemplateToUse);
+			template.waitForConfirmsOrDie(this.waitForConfirmTimeout.toMillis());
+			return null;
+		});
+	}
+
+	private boolean usesSimplePublisherConfirms() {
+		return this.rabbitTemplate != null
+				&& this.rabbitTemplate.getConnectionFactory().isSimplePublisherConfirms();
+	}
+
+	private boolean usesCorrelatedPublisherConfirms() {
+		return this.rabbitTemplate != null
+				&& this.rabbitTemplate.getConnectionFactory().isPublisherConfirms();
+	}
+
 	private void waitForConfirm(Message<?> requestMessage, CorrelationData correlationData) {
 		try {
 			Confirm confirm = correlationData.getFuture().get(this.waitForConfirmTimeout.toMillis(),

spring-integration-amqp/src/main/resources/org/springframework/integration/amqp/config/spring-integration-amqp.xsd

@@ -70,6 +70,11 @@
 									been received. Requires a template configured for returns. If a confirm is not
 									received within the confirm timeout or a negative acknowledgment or returned
 									message is received, an exception will be thrown.
+									When there is no correlation data for the message and the connection factory is
+									configured for simple publisher confirms, the message is sent within the scope of
+									a RabbitTemplate.invoke() operation and waitForConfirmsOrDie() is used instead.
+									A returned message does not fail the send in that case; it is published to the
+									return channel, if one is configured.
 								</xsd:documentation>
 							</xsd:appinfo>
 						</xsd:annotation>

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests2.java

@@ -25,6 +25,7 @@
 import org.junit.jupiter.api.Test;
 
 import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.AmqpIOException;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.core.QueueBuilder;
 import org.springframework.amqp.core.QueueBuilder.Overflow;
@@ -53,6 +54,7 @@
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Jun Cho
  *
  * @since 5.2
  *
@@ -63,14 +65,18 @@ public class AmqpOutboundEndpointTests2 implements RabbitTestContainer {
 
 	static final String QUEUE_TEST_CONFIRM_OK = "testConfirmOk";
 
+	static final String QUEUE_TEST_SIMPLE_CONFIRMS = "testSimpleConfirms";
+
 	@BeforeAll
 	static void initQueue() throws IOException, InterruptedException {
 		RABBITMQ.execInContainer("rabbitmqadmin", "declare", "queue", "name=" + QUEUE_TEST_CONFIRM_OK);
+		RABBITMQ.execInContainer("rabbitmqadmin", "declare", "queue", "name=" + QUEUE_TEST_SIMPLE_CONFIRMS);
 	}
 
 	@AfterAll
 	static void deleteQueue() throws IOException, InterruptedException {
 		RABBITMQ.execInContainer("rabbitmqadmin", "delete", "queue", "name=" + QUEUE_TEST_CONFIRM_OK);
+		RABBITMQ.execInContainer("rabbitmqadmin", "delete", "queue", "name=" + QUEUE_TEST_SIMPLE_CONFIRMS);
 	}
 
 	@Test
@@ -118,6 +124,31 @@ void testWithReject(@Autowired IntegrationFlow flow, @Autowired RabbitAdmin admi
 		admin.deleteQueue(queue.getName());
 	}
 
+	@Test
+	void simpleConfirmsOk(@Autowired IntegrationFlow simpleConfirmsFlow, @Autowired RabbitTemplate template) {
+		simpleConfirmsFlow.getInputChannel()
+				.send(new GenericMessage<>("test", Collections.singletonMap("rk", QUEUE_TEST_SIMPLE_CONFIRMS)));
+		assertThat(template.receive(QUEUE_TEST_SIMPLE_CONFIRMS)).isNotNull();
+	}
+
+	@Test
+	void simpleConfirmsWithReject(@Autowired IntegrationFlow simpleConfirmsFlow,
+			@Autowired RabbitAdmin simpleConfirmsAdmin) {
+
+		Queue queue = QueueBuilder.nonDurable().autoDelete().maxLength(1L).overflow(Overflow.rejectPublish).build();
+		String queueName = queue.getName();
+		simpleConfirmsAdmin.declareQueue(queue);
+		GenericMessage<String> message = new GenericMessage<>("test", Collections.singletonMap("rk", queueName));
+		try {
+			simpleConfirmsFlow.getInputChannel().send(message);
+			assertThatThrownBy(() -> simpleConfirmsFlow.getInputChannel().send(message))
+					.hasCauseInstanceOf(AmqpIOException.class);
+		}
+		finally {
+			simpleConfirmsAdmin.deleteQueue(queueName);
+		}
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	@EnableIntegration
 	public static class Config {
@@ -138,6 +169,14 @@ public IntegrationFlow flow2(RabbitTemplate template) {
 					.routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class)));
 		}
 
+		@Bean
+		public IntegrationFlow simpleConfirmsFlow(RabbitTemplate simpleConfirmsTemplate) {
+			return f -> f.handle(Amqp.outboundAdapter(simpleConfirmsTemplate)
+					.exchangeName("")
+					.routingKeyFunction(msg -> msg.getHeaders().get("rk", String.class))
+					.waitForConfirm(true));
+		}
+
 		@Bean
 		public CachingConnectionFactory cf() {
 			CachingConnectionFactory ccf = new CachingConnectionFactory(RabbitTestContainer.amqpPort());
@@ -154,6 +193,23 @@ public RabbitTemplate template(ConnectionFactory cf) {
 			return rabbitTemplate;
 		}
 
+		@Bean
+		public CachingConnectionFactory simpleConfirmsCf() {
+			CachingConnectionFactory ccf = new CachingConnectionFactory(RabbitTestContainer.amqpPort());
+			ccf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE);
+			return ccf;
+		}
+
+		@Bean
+		public RabbitTemplate simpleConfirmsTemplate(ConnectionFactory simpleConfirmsCf) {
+			return new RabbitTemplate(simpleConfirmsCf);
+		}
+
+		@Bean
+		public RabbitAdmin simpleConfirmsAdmin(ConnectionFactory simpleConfirmsCf) {
+			return new RabbitAdmin(simpleConfirmsCf);
+		}
+
 		@Bean
 		public RabbitAdmin admin(ConnectionFactory cf) {
 			return new RabbitAdmin(cf);

src/reference/antora/modules/ROOT/pages/amqp/outbound-channel-adapter.adoc

@@ -123,10 +123,15 @@ Pending confirms are checked every 50% of this value, so the actual time a nack
 Also see xref:amqp/alternative-confirms-returns.adoc[Alternative Mechanism for Publisher Confirms and Returns].
 Default none (nacks will not be generated).
 <14> When set to true, the calling thread will block, waiting for a publisher confirmation.
-This requires a `RabbitTemplate` configured for `confirms` as well as a `confirm-correlation-expression`.
+This requires a `RabbitTemplate` configured for `confirms`.
+With correlated publisher confirms, the message also needs correlation data, which is usually provided by a `confirm-correlation-expression`.
 The thread will block for up to `confirm-timeout` (or 5 seconds by default).
 If a timeout occurs, a `MessageTimeoutException` will be thrown.
 If returns are enabled and a message is returned, or any other exception occurs while awaiting the confirmation, a `MessageHandlingException` will be thrown, with an appropriate message.
+With simple publisher confirms and no correlation data, the adapter sends the message within the scope of a `RabbitTemplate.invoke()` operation and calls `waitForConfirmsOrDie()` instead.
+A returned message then does not fail the send; it is published to the `return-channel`, if one is configured.
+This option cannot be combined with a `BoundRabbitChannelAdvice` that shares the same `RabbitTemplate`, since both open a scoped operation.
+Also see xref:amqp/strict-ordering.adoc[Strict Message Ordering].
 <15> The channel to which returned messages are sent.
 When provided, the underlying AMQP template is configured to return undeliverable messages to the adapter.
 When there is no `ErrorMessageStrategy` configured, the message is constructed from the data received from AMQP, with the following additional headers: `amqp_returnReplyCode`, `amqp_returnReplyText`, `amqp_returnExchange`, `amqp_returnRoutingKey`.

src/reference/antora/modules/ROOT/pages/amqp/strict-ordering.adoc

@@ -55,5 +55,9 @@ Notice that the same `RabbitTemplate` (which implements `RabbitOperations`) is u
 The advice runs the downstream flow within the template's `invoke` method so that all operations run on the same channel.
 If the optional timeout is provided, when the flow completes, the advice calls the `waitForConfirmsOrDie` method, which throws an exception if the confirmations are not received within the specified time.
 
+NOTE: The outbound adapter opens a scoped operation of its own when `multi-send` is used, and also when `wait-for-confirm` is set with simple publisher confirms.
+Do not use those options on an adapter whose flow is already wrapped by a `BoundRabbitChannelAdvice` that shares the same `RabbitTemplate`, because nested scopes are not supported.
+The advice already waits for the confirmations of all the messages sent within its scope.
+
 IMPORTANT: There must be no thread hands-off in the downstream flow (`QueueChannel`, `ExecutorChannel`, and others).
 

src/reference/antora/modules/ROOT/pages/whats-new.adoc

@@ -22,3 +22,11 @@ Java 17 is still the baseline, but Java 25 is supported.
 
 The `DefaultErrorMessageStrategy` now preserves the original headers when building the `ErrorMessage`.
 See xref:error-handling.adoc[] for more information.
+
+[[x7.2-amqp-changes]]
+=== AMQP Support Changes
+
+The `wait-for-confirm` option on the outbound channel adapter now also works with simple publisher confirms.
+When the message has no correlation data, the send is performed within the scope of a `RabbitTemplate.invoke()` operation and `waitForConfirmsOrDie()` is used.
+A warning is now logged when the option is set but the connection factory supports no publisher confirms at all.
+See xref:amqp/outbound-channel-adapter.adoc[] for more information.