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 7d9ef52e7848 CAMEL-26082: camel-sjms - preserveMessageQos also applies
to InOnly sends (#28590)
7d9ef52e7848 is described below
commit 7d9ef52e78489c4f89792c63d314dcd817eae920
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 28 21:34:47 2026 +0530
CAMEL-26082: camel-sjms - preserveMessageQos also applies to InOnly sends
(#28590)
With preserveMessageQos the endpoint priority, delivery mode and time to
live are used for InOnly sends and for exception replies of a consumer when the
message carries no QoS headers, even with explicitQosEnabled=false, as
camel-jms does. Documented in the 4.24 upgrade guide.
Co-Authored-By: Claude <[email protected]>
---
.../apache/camel/component/sjms/SjmsEndpoint.java | 9 +-
.../apache/camel/component/sjms/SjmsTemplate.java | 43 ++++-
.../sjms/consumer/EndpointMessageListener.java | 4 +-
.../producer/InOnlyPreserveMessageQosTest.java | 214 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 20 ++
5 files changed, 276 insertions(+), 14 deletions(-)
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
index 55cd0de8f889..32c238cedb20 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsEndpoint.java
@@ -495,8 +495,8 @@ public class SjmsEndpoint extends DefaultEndpoint
public SjmsTemplate createInOnlyTemplate() {
SjmsTemplate template = new SjmsTemplate(getConnectionFactory(),
isTransacted(), getAcknowledgementMode().intValue());
- // configure qos if enabled
- if (isExplicitQosEnabled()) {
+ // configure qos if enabled, and with preserveMessageQos as the values
for a message without the headers
+ if (isExplicitQosEnabled() || isPreserveMessageQos()) {
int dm = isDeliveryPersistent() ? DeliveryMode.PERSISTENT :
DeliveryMode.NON_PERSISTENT;
if (getDeliveryMode() != null) {
dm = getDeliveryMode();
@@ -504,6 +504,7 @@ public class SjmsEndpoint extends DefaultEndpoint
template.setQoSSettings(dm, getPriority(), getTimeToLive());
}
template.setDestinationCreationStrategy(getDestinationCreationStrategy());
+ template.setPreserveMessageQos(isPreserveMessageQos());
return template;
}
@@ -515,8 +516,8 @@ public class SjmsEndpoint extends DefaultEndpoint
// must be auto-ack mode for InOut (request/reply mode)
SjmsTemplate template = new SjmsTemplate(getConnectionFactory(),
false, Session.AUTO_ACKNOWLEDGE);
- // configure qos if enabled
- if (isExplicitQosEnabled()) {
+ // configure qos if enabled, and with preserveMessageQos as the values
for a message without the headers
+ if (isExplicitQosEnabled() || isPreserveMessageQos()) {
int dm = isDeliveryPersistent() ? DeliveryMode.PERSISTENT :
DeliveryMode.NON_PERSISTENT;
if (getDeliveryMode() != null) {
dm = getDeliveryMode();
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsTemplate.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsTemplate.java
index 12bdc59b8220..4d85e9838887 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsTemplate.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/SjmsTemplate.java
@@ -139,7 +139,7 @@ public class SjmsTemplate {
this.message = messageCreator.createMessage(session);
MessageProducer producer = session.createProducer(dest);
try {
- send(producer, message);
+ send(producer, message, exchange.getIn());
} finally {
closeProducer(producer);
}
@@ -170,6 +170,20 @@ public class SjmsTemplate {
}
public void send(MessageProducer producer, Message message) throws
Exception {
+ send(producer, message, null);
+ }
+
+ /**
+ * Sends the message with the producer.
+ *
+ * @param producer the producer
+ * @param message the JMS message
+ * @param camelMessage the Camel message the JMS message was created from.
With preserveMessageQos, the priority and
+ * delivery mode of the JMS message are only used when
this message has the JMSPriority or
+ * JMSDeliveryMode header, otherwise those of the
endpoint. When null, those of the JMS message
+ * are used.
+ */
+ public void send(MessageProducer producer, Message message,
org.apache.camel.Message camelMessage) throws Exception {
if (preserveMessageQos) {
long ttl = message.getJMSExpiration();
if (ttl != 0) {
@@ -178,19 +192,28 @@ public class SjmsTemplate {
if (ttl <= 0) {
ttl = 1;
}
+ } else {
+ // the message has no expiration, so use the time to live of
the endpoint (timeToLive, or the
+ // requestTimeout for request/reply), which is 0 (never
expires) when none is configured
+ ttl = this.timeToLive;
}
- int priority = message.getJMSPriority();
- if (priority < 0 || priority > 9) {
+ // a new JMS message always has a priority and delivery mode (the
binding sets PERSISTENT, and the
+ // priority is the default of the JMS client, 0 with ActiveMQ
Classic), so they are only taken from the
+ // message when the Camel message has the header; otherwise the
endpoint values are used
+ int priority = this.priority;
+ if (camelMessage == null ||
camelMessage.getHeader(JmsConstants.JMS_PRIORITY) != null) {
+ int messagePriority = message.getJMSPriority();
// use priority from endpoint if not provided on message with
a valid range
- priority = this.priority;
+ if (messagePriority >= 0 && messagePriority <= 9) {
+ priority = messagePriority;
+ }
}
- // if a delivery mode was set as a JMS header then we have used a
temporary
- // property to store it - CamelJMSDeliveryMode. Otherwise we could
not keep
- // track whether it was set or not as getJMSDeliveryMode() will
default return 1 regardless
- // if it was set or not, so we can never tell if end user provided
it in a header
- int resolvedDeliveryMode = resolveDeliveryMode(message);
+ int resolvedDeliveryMode = this.deliveryMode;
+ if (camelMessage == null ||
camelMessage.getHeader(JmsConstants.JMS_DELIVERY_MODE) != null) {
+ resolvedDeliveryMode = resolveDeliveryMode(message);
+ }
producer.send(message, resolvedDeliveryMode, priority, ttl);
} else if (explicitQosEnabled) {
@@ -202,6 +225,8 @@ public class SjmsTemplate {
private static int resolveDeliveryMode(Message message) throws
JMSException {
int resolvedDeliveryMode;
+ // the sjms binding sets the JMSDeliveryMode header with
setJMSDeliveryMode, so the binding never creates this
+ // property; it is kept for a JMS message that carries it as a property
if (JmsMessageHelper.hasProperty(message,
JmsConstants.JMS_DELIVERY_MODE)) {
resolvedDeliveryMode =
message.getIntProperty(JmsConstants.JMS_DELIVERY_MODE);
// remove the temporary property
diff --git
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
index 3e16b22f85fb..012e353972ff 100644
---
a/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
+++
b/components/camel-sjms/src/main/java/org/apache/camel/component/sjms/consumer/EndpointMessageListener.java
@@ -45,6 +45,7 @@ import org.apache.camel.component.sjms.SjmsMessage;
import org.apache.camel.component.sjms.SjmsTemplate;
import org.apache.camel.component.sjms.jms.JmsMessageHelper;
import org.apache.camel.support.AsyncProcessorConverterHelper;
+import org.apache.camel.support.DefaultMessage;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -315,7 +316,8 @@ public class EndpointMessageListener implements
SessionMessageListener {
producer = messageProducerCreator.create(session,
endpoint, replyDestination);
- template.send(producer, reply);
+ // an exception reply has no Camel message: it has no QoS
headers, so it gets the QoS of the endpoint
+ template.send(producer, reply, out != null ? out : new
DefaultMessage(exchange));
} finally {
close(producer);
}
diff --git
a/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/InOnlyPreserveMessageQosTest.java
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/InOnlyPreserveMessageQosTest.java
new file mode 100644
index 000000000000..c5d336c4c833
--- /dev/null
+++
b/components/camel-sjms/src/test/java/org/apache/camel/component/sjms/producer/InOnlyPreserveMessageQosTest.java
@@ -0,0 +1,214 @@
+/*
+ * 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.sjms.producer;
+
+import java.util.Map;
+
+import jakarta.jms.DeliveryMode;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.sjms.support.JmsTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * preserveMessageQos=true applies the JMSPriority, JMSDeliveryMode and
JMSExpiration headers also to InOnly sends, not
+ * only to InOut sends, and a message without one of these headers gets the
priority, delivery mode or time to live of
+ * the endpoint.
+ */
+class InOnlyPreserveMessageQosTest extends JmsTestSupport {
+
+ private static final String QUEUE =
"sjms:queue:in.only.preserve.qos.InOnlyPreserveMessageQosTest";
+ private static final String EXPIRY_QUEUE =
"sjms:queue:in.only.preserve.qos.expiry.InOnlyPreserveMessageQosTest";
+ private static final String REQUEST_QUEUE =
"sjms:queue:in.only.preserve.qos.request.InOnlyPreserveMessageQosTest";
+ private static final String REPLY_QUEUE =
"sjms:queue:in.only.preserve.qos.reply.InOnlyPreserveMessageQosTest";
+ private static final String FAILING_QUEUE =
"sjms:queue:in.only.preserve.qos.failing.InOnlyPreserveMessageQosTest";
+ private static final String EXCEPTION_REPLY_QUEUE =
"in.only.preserve.qos.exception.reply.InOnlyPreserveMessageQosTest";
+
+ @Test
+ void priorityAndDeliveryModeFromTheHeadersAreUsed() throws Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("Hello World");
+ result.message(0).header("JMSPriority").isEqualTo(8);
+
result.message(0).header("JMSDeliveryMode").isEqualTo(DeliveryMode.NON_PERSISTENT);
+
+ template.sendBodyAndHeaders(QUEUE + "?preserveMessageQos=true", "Hello
World",
+ Map.of("JMSPriority", 8, "JMSDeliveryMode",
DeliveryMode.NON_PERSISTENT));
+
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void priorityAndDeliveryModeOfTheEndpointAreUsedWithoutHeaders() throws
Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("Hello World");
+ // a new JMS message has a priority (the default of the JMS client)
and the delivery mode set by the binding
+ // (PERSISTENT), which must not win over the endpoint when the route
did not set the headers
+ result.message(0).header("JMSPriority").isEqualTo(7);
+
result.message(0).header("JMSDeliveryMode").isEqualTo(DeliveryMode.NON_PERSISTENT);
+
+ template.sendBody(QUEUE +
"?preserveMessageQos=true&priority=7&deliveryPersistent=false", "Hello World");
+
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void headersWinOverTheEndpoint() throws Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("Hello World");
+ result.message(0).header("JMSPriority").isEqualTo(2);
+
result.message(0).header("JMSDeliveryMode").isEqualTo(DeliveryMode.PERSISTENT);
+
+ template.sendBodyAndHeaders(QUEUE +
"?preserveMessageQos=true&priority=7&deliveryPersistent=false", "Hello World",
+ Map.of("JMSPriority", 2, "JMSDeliveryMode", "PERSISTENT"));
+
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void priorityAndDeliveryModeOfTheEndpointAreUsedWithExplicitQosDisabled()
throws Exception {
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("Hello World");
+ result.message(0).header("JMSPriority").isEqualTo(7);
+
result.message(0).header("JMSDeliveryMode").isEqualTo(DeliveryMode.NON_PERSISTENT);
+
+ template.sendBody(QUEUE +
"?preserveMessageQos=true&explicitQosEnabled=false&priority=7&deliveryPersistent=false",
+ "Hello World");
+
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void
exceptionReplyOfAConsumerUsesThePriorityAndDeliveryModeOfTheEndpoint() throws
Exception {
+ MockEndpoint reply = getMockEndpoint("mock:exceptionReply");
+ reply.expectedMessageCount(1);
+ reply.message(0).header("JMSPriority").isEqualTo(7);
+
reply.message(0).header("JMSDeliveryMode").isEqualTo(DeliveryMode.NON_PERSISTENT);
+ reply.message(0).body().isInstanceOf(IllegalStateException.class);
+
+ // InOnly with a JMSReplyTo header: the failing consumer sends the
exception reply to the reply queue
+ template.sendBodyAndHeader(FAILING_QUEUE + "?preserveMessageQos=true",
"Hello World", "JMSReplyTo",
+ EXCEPTION_REPLY_QUEUE);
+
+ reply.assertIsSatisfied();
+ }
+
+ @Test
+ void requestUsesThePriorityAndDeliveryModeOfTheEndpointWithoutHeaders()
throws Exception {
+ MockEndpoint request = getMockEndpoint("mock:request");
+ request.expectedBodiesReceived("Hello World");
+ request.message(0).header("JMSPriority").isEqualTo(7);
+
request.message(0).header("JMSDeliveryMode").isEqualTo(DeliveryMode.NON_PERSISTENT);
+
+ Object reply = template.requestBody(REQUEST_QUEUE +
"?preserveMessageQos=true&priority=7&deliveryPersistent=false",
+ "Hello World");
+
+ request.assertIsSatisfied();
+ assertEquals("Bye World", reply);
+ }
+
+ @Test
+ void replyOfAConsumerUsesThePriorityHeader() {
+ Exchange out = template.request(REPLY_QUEUE, e ->
e.getIn().setBody("Hello World"));
+
+ assertEquals("Bye World", out.getMessage().getBody(String.class));
+ // the consumer endpoint has preserveMessageQos=true and its route
sets JMSPriority on the reply
+ assertEquals(9, out.getMessage().getHeader("JMSPriority",
Integer.class));
+ }
+
+ @Test
+ void expirationFromTheHeaderIsUsed() throws Exception {
+ long expiration = System.currentTimeMillis() + 600_000;
+
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("Hello World");
+ // without the header the message would never expire (JMSExpiration 0)
+ result.message(0).header("JMSExpiration").isGreaterThan(expiration -
60_000);
+
+ template.sendBodyAndHeader(QUEUE + "?preserveMessageQos=true", "Hello
World", "JMSExpiration", expiration);
+
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void timeToLiveOfTheEndpointIsUsedForAMessageWithoutExpiration() throws
Exception {
+ long before = System.currentTimeMillis();
+
+ MockEndpoint result = getMockEndpoint("mock:result");
+ result.expectedBodiesReceived("Hello World");
+
result.message(0).header("JMSExpiration").isGreaterThanOrEqualTo(before +
600_000);
+
+ template.sendBody(QUEUE +
"?preserveMessageQos=true&timeToLive=600000", "Hello World");
+
+ result.assertIsSatisfied();
+ }
+
+ @Test
+ void requestTimeoutIsUsedForARequestWithoutExpiration() throws Exception {
+ long before = System.currentTimeMillis();
+
+ MockEndpoint request = getMockEndpoint("mock:request");
+ request.expectedBodiesReceived("Hello World");
+
request.message(0).header("JMSExpiration").isGreaterThanOrEqualTo(before +
20_000);
+
+ Object reply = template.requestBody(REQUEST_QUEUE +
"?preserveMessageQos=true&requestTimeout=20000", "Hello World");
+
+ request.assertIsSatisfied();
+ assertEquals("Bye World", reply);
+ }
+
+ @Test
+ void anExpiredMessageIsNotDelivered() throws Exception {
+ MockEndpoint expiry = getMockEndpoint("mock:expiry");
+ expiry.expectedBodiesReceived("Fresh");
+ expiry.setAssertPeriod(500);
+
+ // expired one minute ago: sent with the shortest time to live, so the
broker drops it
+ template.sendBodyAndHeader(EXPIRY_QUEUE + "?preserveMessageQos=true",
"Stale", "JMSExpiration",
+ System.currentTimeMillis() - 60_000);
+ template.sendBody(EXPIRY_QUEUE, "Fresh");
+ // the consumer starts only after both messages were sent
+ context.getRouteController().startRoute("expiry");
+
+ expiry.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from(QUEUE).to("mock:result");
+
+
from(REQUEST_QUEUE).to("mock:request").transform().constant("Bye World");
+
+ from(REPLY_QUEUE +
"?preserveMessageQos=true").setHeader("JMSPriority", constant(9))
+ .transform().constant("Bye World");
+
+
from(EXPIRY_QUEUE).routeId("expiry").autoStartup(false).to("mock:expiry");
+
+ // an exception reply is an ObjectMessage, which needs
objectMessageEnabled on both sides
+ from(FAILING_QUEUE +
"?preserveMessageQos=true&transferException=true&objectMessageEnabled=true"
+ + "&priority=7&deliveryPersistent=false")
+ .throwException(new IllegalStateException("Forced"));
+ from("sjms:queue:" + EXCEPTION_REPLY_QUEUE +
"?objectMessageEnabled=true").to("mock:exceptionReply");
+ }
+ };
+ }
+}
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 7767b53870a0..d29de39c38f8 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
@@ -1945,6 +1945,26 @@ live, so it never expired, even when the endpoint has a
`timeToLive` and, for re
`requestTimeout` as time to live otherwise. Such a message now gets the time
to live of the endpoint, as the option
documents. Set `disableTimeToLive=true` to send it without time to live.
+=== camel-sjms - preserveMessageQos also applies to InOnly sends
+
+With `preserveMessageQos=true`, an InOnly send ignored the `JMSPriority`,
`JMSDeliveryMode` and `JMSExpiration`
+headers, and used the QoS of the endpoint (or of the JMS client); only
request/reply used them. An InOnly send now
+uses these headers, as camel-jms does. A message whose `JMSExpiration` is in
the past expires at once, and a message
+consumed from JMS and sent on keeps the priority, delivery mode and expiration
of the consumed message. The replies of
+a consumer endpoint with `preserveMessageQos=true` use the headers of the
reply message too; these are still the
+headers of the request unless the route changes or removes them. An exception
reply (`transferException=true`) has
+no headers and uses the QoS of the endpoint. `camel-sjms2` behaves the same.
+
+A header that is not set falls back to the endpoint, as the option documents,
also with `explicitQosEnabled=false`: to
+the `priority` (a `priority` of 0 is not applied, as with
`explicitQosEnabled`) and the `deliveryPersistent` (or
+`deliveryMode`) of the endpoint, and, for a message without `JMSExpiration`
(header missing or
+0), to the time to live of the endpoint (`timeToLive`, or the `requestTimeout`
for request/reply). This also changes
+request/reply: before, a request without `JMSPriority` was sent with the
priority of the new JMS message (the default
+of the JMS client, 0 with ActiveMQ Classic), a request without
`JMSDeliveryMode` was persistent whatever the endpoint
+said, and a request without `JMSExpiration` never expired. To send a request
without time to live, set
+`disableTimeToLive=true` and no `timeToLive`. Remove `preserveMessageQos=true`
to keep using only the QoS of the
+endpoint for InOnly sends.
+
=== camel-pulsar - the producer continues the sequence ids of the broker
The `pulsar` producer no longer passes the default `initialSequenceId=-1` to
the Pulsar client, so the client
