This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 9886c7710017 CAMEL-25947: camel-spring-rabbitmq - confirm=auto waits
for the confirms of a connection factory with simple publisher confirms (#28465)
9886c7710017 is described below
commit 9886c771001723f417772fb900603ab475734044
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 26 13:32:23 2026 +0530
CAMEL-25947: camel-spring-rabbitmq - confirm=auto waits for the confirms of
a connection factory with simple publisher confirms (#28465)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../springrabbit/SpringRabbitMQProducer.java | 10 +-
.../SpringRabbitMQProducerConfirmTest.java | 154 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++
3 files changed, 172 insertions(+), 2 deletions(-)
diff --git
a/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java
b/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java
index c4b21f688f28..24c9c8a14d1f 100644
---
a/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java
+++
b/components/camel-spring-parent/camel-spring-rabbitmq/src/main/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducer.java
@@ -35,6 +35,7 @@ import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.RabbitMessageFuture;
import org.springframework.amqp.rabbit.connection.Connection;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
@@ -222,7 +223,10 @@ public class SpringRabbitMQProducer extends
DefaultAsyncProducer {
final String rk = routingKey;
boolean confirm;
if ("auto".equalsIgnoreCase(getEndpoint().getConfirm())) {
- confirm =
getEndpoint().getConnectionFactory().isPublisherConfirms();
+ // a connection factory with publisher confirms of type CORRELATED
answers isPublisherConfirms, one of type
+ // SIMPLE (the type that waitForConfirms is made for) answers
isSimplePublisherConfirms instead
+ ConnectionFactory cf = getEndpoint().getConnectionFactory();
+ confirm = cf.isPublisherConfirms() ||
cf.isSimplePublisherConfirms();
} else if ("enabled".equalsIgnoreCase(getEndpoint().getConfirm())) {
confirm = true;
} else {
@@ -242,7 +246,9 @@ public class SpringRabbitMQProducer extends
DefaultAsyncProducer {
}
if (!sent) {
- exchange.setException(new TimeoutException("Message not sent
within " + timeout + " millis"));
+ // waitForConfirms returns false only when RabbitMQ negatively
acknowledged the message; when no confirm
+ // arrives within the timeout it throws AmqpTimeoutException,
which the catch below sets on the exchange
+ exchange.setException(new TimeoutException("Message negatively
acknowledged by RabbitMQ"));
}
} catch (Exception e) {
exchange.setException(e);
diff --git
a/components/camel-spring-parent/camel-spring-rabbitmq/src/test/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducerConfirmTest.java
b/components/camel-spring-parent/camel-spring-rabbitmq/src/test/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducerConfirmTest.java
new file mode 100644
index 000000000000..69b022fe07b3
--- /dev/null
+++
b/components/camel-spring-parent/camel-spring-rabbitmq/src/test/java/org/apache/camel/component/springrabbit/SpringRabbitMQProducerConfirmTest.java
@@ -0,0 +1,154 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.camel.component.springrabbit;
+
+import java.util.concurrent.TimeoutException;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.amqp.AmqpTimeoutException;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import
org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.RETURNS_DEFAULTS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * With the default {@code confirm=auto}, an InOnly producer waits for the
publisher confirm when the connection factory
+ * has publisher confirms, of either type.
+ * <p>
+ * The connection factory is a real Spring {@link CachingConnectionFactory}:
it decides what
+ * {@code isPublisherConfirms()} (true for {@link ConfirmType#CORRELATED}
only) and {@code isSimplePublisherConfirms()}
+ * (true for {@link ConfirmType#SIMPLE} only) answer, and it puts the channel
in confirm mode. Below it, the RabbitMQ
+ * client is mocked; its {@link Channel#waitForConfirms(long)} answers as the
RabbitMQ Java client documents it:
+ * {@code true} when every message published since the last call was
acknowledged, {@code false} when one was negatively
+ * acknowledged (for example by a queue with {@code x-overflow=reject-publish}
that is full), and a
+ * {@link TimeoutException} when the timeout elapses. Spring's {@code
RabbitTemplate.waitForConfirms(long)} turns that
+ * {@link TimeoutException} into an {@link AmqpTimeoutException}.
+ */
+class SpringRabbitMQProducerConfirmTest {
+
+ private final Connection connection = mock(Connection.class);
+ private final Channel channel = mock(Channel.class);
+ private CamelContext context;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ when(connection.isOpen()).thenReturn(true);
+ when(connection.createChannel()).thenReturn(channel);
+ when(channel.isOpen()).thenReturn(true);
+ when(channel.getConnection()).thenReturn(connection);
+ context = new DefaultCamelContext();
+ }
+
+ @AfterEach
+ void tearDown() {
+ if (context != null) {
+ context.stop();
+ }
+ }
+
+ @Test
+ void negativeAcknowledgementFailsTheExchangeWithSimplePublisherConfirms()
throws Exception {
+ when(channel.waitForConfirms(anyLong())).thenReturn(false);
+
+ Exchange exchange = send(ConfirmType.SIMPLE);
+
+ assertInstanceOf(TimeoutException.class, exchange.getException());
+ assertEquals("Message negatively acknowledged by RabbitMQ",
exchange.getException().getMessage());
+ verify(channel).confirmSelect();
+ verify(channel).basicPublish(eq("orders"), eq("new"), eq(false),
any(), any());
+ }
+
+ @Test
+ void missingConfirmFailsTheExchangeWithSimplePublisherConfirms() throws
Exception {
+ when(channel.waitForConfirms(anyLong())).thenThrow(new
TimeoutException());
+
+ Exchange exchange = send(ConfirmType.SIMPLE);
+
+ assertInstanceOf(AmqpTimeoutException.class, exchange.getException());
+ assertInstanceOf(TimeoutException.class,
exchange.getException().getCause());
+ verify(channel).waitForConfirms(5000L);
+ }
+
+ @Test
+ void acknowledgedMessageIsSentWithSimplePublisherConfirms() throws
Exception {
+ when(channel.waitForConfirms(anyLong())).thenReturn(true);
+
+ Exchange exchange = send(ConfirmType.SIMPLE);
+
+ assertNull(exchange.getException());
+ verify(channel).waitForConfirms(5000L);
+ }
+
+ @Test
+ void acknowledgedMessageIsSentWithCorrelatedPublisherConfirms() throws
Exception {
+ // control, which passes without the fix too
+ when(channel.waitForConfirms(anyLong())).thenReturn(true);
+
+ Exchange exchange = send(ConfirmType.CORRELATED);
+
+ assertNull(exchange.getException());
+ verify(channel).waitForConfirms(5000L);
+ }
+
+ @Test
+ void noConfirmIsAwaitedWithoutPublisherConfirms() throws Exception {
+ // control, which passes without the fix too
+ Exchange exchange = send(ConfirmType.NONE);
+
+ assertNull(exchange.getException());
+ verify(channel).basicPublish(eq("orders"), eq("new"), eq(false),
any(), any());
+ verify(channel, never()).waitForConfirms(anyLong());
+ }
+
+ private Exchange send(ConfirmType confirmType) throws Exception {
+ ConnectionFactory rabbit = mock(ConnectionFactory.class, invocation ->
invocation.getMethod().getReturnType()
+ .equals(Connection.class) ? connection :
RETURNS_DEFAULTS.answer(invocation));
+ CachingConnectionFactory cf = new CachingConnectionFactory(rabbit);
+ cf.setPublisherConfirmType(confirmType);
+ context.getRegistry().bind("cf", cf);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:start").to("spring-rabbitmq:orders?routingKey=new&connectionFactory=#cf");
+ }
+ });
+ context.start();
+ ProducerTemplate template = context.createProducerTemplate();
+ Exchange exchange = template.send("direct:start", e ->
e.getIn().setBody("order 1"));
+ return exchange;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 39e4283398ab..7dcb1fd4ef7f 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -273,6 +273,16 @@ configuration (the `camel.component.google-mail.*`
options), kept until the comp
client the component had already created, for example the one of a
`google-mail:messages/modify` endpoint, so
credentials set only on an endpoint URI no longer reach the transformer: set
them on the component as well.
+=== camel-spring-rabbitmq - confirm=auto waits for simple publisher confirms
+
+With the default `confirm=auto`, an InOnly producer now also waits for the
publisher confirm when the connection
+factory has publisher confirms of type `SIMPLE` (for example
`spring.rabbitmq.publisher-confirm-type=simple`), as it
+did for type `CORRELATED`. Before, it sent such messages without waiting, so a
message that RabbitMQ negatively
+acknowledged, or did not confirm, was lost without an error. Each send now
waits up to `confirmTimeout` (5 seconds by
+default). It fails with a `java.util.concurrent.TimeoutException` when
RabbitMQ negatively acknowledges the message,
+and with an `org.springframework.amqp.AmqpTimeoutException` when RabbitMQ does
not confirm it within `confirmTimeout`.
+Set `confirm=disabled` to send without waiting, as before.
+
=== camel-jbang - camel validate and camel overview report the same endpoints
no route consumes
`camel validate source` and `camel overview` now use the same rules for a
`direct:`, `seda:` or `disruptor:` endpoint
