This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch fix/ibmmq-10-upgrade-with-lang-fix in repository https://gitbox.apache.org/repos/asf/camel.git
commit dd5f83d7e782e45f7480c866461340e1967fca24 Author: Guillaume Nodet <[email protected]> AuthorDate: Thu Jul 30 18:13:30 2026 +0200 fix: handle reserved JMS vendor properties when copying headers to outgoing messages IBM MQ 10.0 marks JMS_IBM_MsgToken as a reserved read-only property that cannot be set by applications. When Camel copies headers from an incoming JMS message to an outgoing reply or forwarded message, it attempts to set all JMS_IBM_* properties on the new message. This causes a MessageFormatException (JMSCC0050) that crashes the reply send, resulting in null replies. The fix wraps the property-setting call in JmsBinding.appendJmsProperty() with a try-catch that gracefully skips any vendor-specific property that the JMS provider considers reserved or read-only. This is a defensive approach that works for any JMS provider, not just IBM MQ. Also removes diagnostic trace logging and test instrumentation that was added during investigation, and modernizes the test to follow project conventions (package-private visibility, AssertJ assertions). Co-Authored-By: Claude Opus 4.6 <[email protected]> --- .../component/jms/EndpointMessageListener.java | 47 +-- .../org/apache/camel/component/jms/JmsBinding.java | 16 +- .../component/jms/issues/JmsReplyToIbmMQTest.java | 326 +-------------------- .../src/test/resources/log4j2-test.properties | 7 - 4 files changed, 28 insertions(+), 368 deletions(-) diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java index 22dec9607379..29be0c3b4838 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/EndpointMessageListener.java @@ -80,13 +80,6 @@ public class EndpointMessageListener implements SessionAwareMessageListener { // and disableReplyTo hasn't been explicit enabled sendReply = replyDestination != null && !disableReplyTo; - if (LOG.isTraceEnabled()) { - LOG.trace("onMessage: replyDestination={} (type={}), disableReplyTo={}, sendReply={}", - replyDestination, - replyDestination != null ? replyDestination.getClass().getName() : "null", - disableReplyTo, sendReply); - } - // we should also not send back reply to ourself if this destination and replyDestination is the same Destination destination = JmsMessageHelper.getJMSDestination(message); if (destination != null && sendReply && !endpoint.isReplyToSameDestinationAllowed() @@ -95,15 +88,6 @@ public class EndpointMessageListener implements SessionAwareMessageListener { destination); sendReply = false; } - if (LOG.isTraceEnabled()) { - LOG.trace("onMessage: after same-dest check: destination={} (type={}), " - + "replyToSameDestAllowed={}, equals={}, sendReply={}", - destination, - destination != null ? destination.getClass().getName() : "null", - endpoint.isReplyToSameDestinationAllowed(), - destination != null ? destination.equals(replyDestination) : "N/A", - sendReply); - } final Exchange exchange = createExchange(message, session, replyDestination); if (ObjectHelper.isNotEmpty(eagerPoisonBody) && eagerLoadingOfProperties) { @@ -237,13 +221,6 @@ public class EndpointMessageListener implements SessionAwareMessageListener { } private void handleReplyIfNeeded(RuntimeCamelException rce) { - if (LOG.isTraceEnabled()) { - LOG.trace("handleReplyIfNeeded: rce={}, sendReply={}, replyDestination={} (type={}), " - + "exchangePattern={}, exchangeFailed={}, exchangeException={}", - rce, sendReply, replyDestination, - replyDestination != null ? replyDestination.getClass().getName() : "null", - exchange.getPattern(), exchange.isFailed(), exchange.getException()); - } if (rce != null || !sendReply) { return; } @@ -257,28 +234,14 @@ public class EndpointMessageListener implements SessionAwareMessageListener { body = exchange.getMessage(); } - if (LOG.isTraceEnabled()) { - LOG.trace("handleReplyIfNeeded: body={}, bodyContent={}, cause={}", - body != null ? body.getClass().getName() : "null", - body != null ? body.getBody() : "null", - cause); - } - if (body != null || cause != null) { LOG.trace("onMessage.sendReply START"); - try { - if (replyDestination instanceof Destination destination) { - sendReply(destination, message, exchange, body, cause); - } else { - sendReply((String) replyDestination, message, exchange, body, cause); - } - LOG.trace("onMessage.sendReply END"); - } catch (Exception e) { - LOG.warn("onMessage.sendReply FAILED with exception", e); - throw e; + if (replyDestination instanceof Destination destination) { + sendReply(destination, message, exchange, body, cause); + } else { + sendReply((String) replyDestination, message, exchange, body, cause); } - } else { - LOG.trace("handleReplyIfNeeded: NOT sending reply (body=null and cause=null)"); + LOG.trace("onMessage.sendReply END"); } } diff --git a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java index f14c46dc016e..6a2f36c5a7c7 100644 --- a/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java +++ b/components/camel-jms/src/main/java/org/apache/camel/component/jms/JmsBinding.java @@ -467,8 +467,20 @@ public class JmsBinding { if (value != null) { // must encode to safe JMS header name before setting property on jmsMessage String key = jmsKeyFormatStrategy.encodeKey(headerName); - // set the property - JmsMessageHelper.setProperty(jmsMessage, key, value); + try { + // set the property + JmsMessageHelper.setProperty(jmsMessage, key, value); + } catch (JMSException e) { + // Some JMS providers (e.g. IBM MQ) mark certain vendor-specific properties + // as read-only/reserved (set by the provider, not the application). When Camel + // copies headers from an incoming message to an outgoing reply or forwarded + // message, these reserved properties cause an exception. We skip them gracefully + // since they are provider-assigned metadata that should not be propagated. + if (LOG.isDebugEnabled()) { + LOG.debug("Could not set JMS property '{}' on outgoing message, skipping ({})", + key, e.getMessage()); + } + } } else if (LOG.isDebugEnabled()) { // okay the value is not a primitive or string so we cannot sent it over the wire LOG.debug("Ignoring non primitive header: {} of class: {} with value: {}", diff --git a/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsReplyToIbmMQTest.java b/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsReplyToIbmMQTest.java index e53a109ad8f2..fe7999fc4024 100644 --- a/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsReplyToIbmMQTest.java +++ b/components/camel-jms/src/test/java/org/apache/camel/component/jms/issues/JmsReplyToIbmMQTest.java @@ -16,313 +16,37 @@ */ package org.apache.camel.component.jms.issues; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; - -import jakarta.jms.Connection; import jakarta.jms.ConnectionFactory; -import jakarta.jms.JMSException; -import jakarta.jms.MessageConsumer; -import jakarta.jms.MessageProducer; -import jakarta.jms.Queue; -import jakarta.jms.Session; -import jakarta.jms.TextMessage; import org.apache.camel.CamelContext; -import org.apache.camel.ExchangePattern; import org.apache.camel.builder.RouteBuilder; -import org.apache.camel.component.jms.EndpointMessageListener; -import org.apache.camel.component.jms.JmsComponent; -import org.apache.camel.component.jms.JmsConsumer; -import org.apache.camel.component.jms.JmsEndpoint; import org.apache.camel.component.jms.JmsTestHelper; -import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.test.infra.ibmmq.common.ConnectionFactoryHelper; import org.apache.camel.test.infra.ibmmq.services.IbmMQService; import org.apache.camel.test.infra.ibmmq.services.IbmMQServiceFactory; import org.apache.camel.test.junit6.CamelTestSupport; -import org.apache.logging.log4j.core.config.Configurator; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.MethodOrderer; -import org.junit.jupiter.api.Order; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.TestMethodOrder; import org.junit.jupiter.api.condition.DisabledOnOs; import org.junit.jupiter.api.extension.RegisterExtension; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.jms.core.JmsOperations; import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknowledge; import static org.assertj.core.api.Assertions.assertThat; -import static org.junit.jupiter.api.Assertions.fail; -/** - * Tests JMS reply-to functionality with IBM MQ. The route consumes from DEV.QUEUE.1 with replyTo configured to - * DEV.QUEUE.2, transforms the message, and the reply should appear on DEV.QUEUE.2. - */ @DisabledOnOs(architectures = { "aarch64", "aarch_64" }, disabledReason = "IBM MQ has no Linux ARM64 native image") -@TestMethodOrder(MethodOrderer.OrderAnnotation.class) class JmsReplyToIbmMQTest extends CamelTestSupport { - private static final Logger LOG = LoggerFactory.getLogger(JmsReplyToIbmMQTest.class); - @RegisterExtension static IbmMQService service = IbmMQServiceFactory.createService(); - // Captures any error thrown during reply sending (normally swallowed by Spring DMLC) - private final AtomicReference<Throwable> listenerError = new AtomicReference<>(); - private final CountDownLatch errorLatch = new CountDownLatch(1); - - // Captures whether the route actually received and processed the message - private final CountDownLatch routeProcessedLatch = new CountDownLatch(1); - private final AtomicReference<String> routeReceivedBody = new AtomicReference<>(); - private final AtomicReference<ExchangePattern> routeExchangePattern = new AtomicReference<>(); - - @BeforeAll - static void enableTraceLogging() { - // Enable TRACE for EndpointMessageListener so we can see exactly what happens - // during the reply flow (onMessage START, process END, sendReply START/END) - Configurator.setLevel("org.apache.camel.component.jms.EndpointMessageListener", - org.apache.logging.log4j.Level.TRACE); - } - - /** - * Test 1: Verifies that explicit send via Camel producer (not reply-to mechanism) works with IBM MQ. Sends to - * DEV.QUEUE.3, route transforms and forwards to DEV.QUEUE.2 via .to(). If this passes but testCustomJMSReplyToInOut - * fails, the issue is in EndpointMessageListener's reply mechanism. Runs first so DEV.QUEUE.2 is drained before the - * main reply-to test. - */ - @Test - @Order(1) - void testManualReplyBypass() throws Exception { - JmsTestHelper.waitForJmsConsumerRoutes(context, "manual-reply"); - - LOG.info("=== MANUAL-BYPASS: Sending test message to DEV.QUEUE.3 ==="); - template.sendBody("jms:queue:DEV.QUEUE.3", "Manual test"); - - LOG.info("=== MANUAL-BYPASS: Waiting for reply on DEV.QUEUE.2 (20s timeout) ==="); - String reply = consumer.receiveBody("jms:queue:DEV.QUEUE.2", 20000, String.class); - LOG.info("=== MANUAL-BYPASS: Received reply: '{}' ===", reply); - - assertThat(reply).isEqualTo("My name is Camel"); - } - - /** - * Test 2: The main reply-to test. Verifies that the EndpointMessageListener correctly sends the reply to - * DEV.QUEUE.2 after route processing. - */ @Test - @Order(2) - void testCustomJMSReplyToInOut() throws Exception { - MockEndpoint mock = getMockEndpoint("mock:processed"); - mock.expectedMessageCount(1); - + void testCustomJMSReplyToInOut() { JmsTestHelper.waitForJmsConsumerRoutes(context, "request"); - // Step 0: Inspect the EndpointMessageListener's reply configuration - LOG.info("=== DIAGNOSTIC: Inspecting EndpointMessageListener state ==="); - String listenerState = inspectListenerState(); - LOG.info(listenerState); - - // Step 1: Verify DEV.QUEUE.2 is accessible by doing a manual JMS send/receive - LOG.info("=== DIAGNOSTIC: Verifying DEV.QUEUE.2 is accessible via raw JMS ==="); - verifyQueueAccessible("DEV.QUEUE.2"); - - // Step 1.5: Verify the reply JmsTemplate can send to DEV.QUEUE.2 - LOG.info("=== DIAGNOSTIC: Testing reply JmsTemplate send to DEV.QUEUE.2 ==="); - String templateTestResult = testReplyTemplate(); - LOG.info("=== DIAGNOSTIC: Reply JmsTemplate test result: {} ===", templateTestResult); - - // Step 2: Send the test message via Camel - LOG.info("=== DIAGNOSTIC: Sending test message to DEV.QUEUE.1 via Camel ==="); template.sendBody("jms:queue:DEV.QUEUE.1", "What is your name?"); - // Step 3: Wait for the route to process the message - LOG.info("=== DIAGNOSTIC: Waiting for route to process the message ==="); - boolean routeProcessed = routeProcessedLatch.await(15, TimeUnit.SECONDS); - LOG.info("=== DIAGNOSTIC: Route processed: {}, received body: '{}', pattern: {} ===", - routeProcessed, routeReceivedBody.get(), routeExchangePattern.get()); - - if (!routeProcessed) { - fail("Route did not process the message within 15 seconds — message may not have arrived at DEV.QUEUE.1"); - } - - // Step 4: Now try to receive the reply from DEV.QUEUE.2 - LOG.info("=== DIAGNOSTIC: Waiting for reply on DEV.QUEUE.2 (20s timeout) ==="); - String reply = consumer.receiveBody("jms:queue:DEV.QUEUE.2", 20000, String.class); - LOG.info("=== DIAGNOSTIC: Received reply: '{}' ===", reply); - - if (reply == null) { - // Build detailed diagnostic report - StringBuilder diagnosis = new StringBuilder(); - diagnosis.append("Reply was null on DEV.QUEUE.2.\n"); - diagnosis.append("Listener state: ").append(listenerState).append("\n"); - diagnosis.append("Reply template test: ").append(templateTestResult).append("\n"); - diagnosis.append("Route processed: ").append(routeProcessed).append("\n"); - diagnosis.append("Route received body: '").append(routeReceivedBody.get()).append("'\n"); - diagnosis.append("Exchange pattern: ").append(routeExchangePattern.get()).append("\n"); - - // Check if the listener error handler captured an exception - if (errorLatch.await(2, TimeUnit.SECONDS)) { - Throwable error = listenerError.get(); - diagnosis.append("Listener error handler captured: ").append(error).append("\n"); - LOG.error("=== DIAGNOSTIC: Reply was null — listener error handler captured exception ===", error); - fail(diagnosis.toString(), error); - } - - // No error was captured — check where the message might have gone - diagnosis.append("No listener error was captured.\n"); - - // Check DEV.QUEUE.2 with raw JMS (bypass Camel type converter) - String rawQ2 = rawJmsReceive("DEV.QUEUE.2", 3000); - diagnosis.append("Raw JMS receive on DEV.QUEUE.2: '").append(rawQ2).append("'\n"); - - // Check Dead Letter Queue for bounced messages - String dlqResult = rawJmsReceive("DEV.DEAD.LETTER.QUEUE", 1000); - diagnosis.append("DEV.DEAD.LETTER.QUEUE: '").append(dlqResult).append("'\n"); - - // Check all DEV.QUEUE.* for stray messages - for (int i = 1; i <= 3; i++) { - String stray = rawJmsReceive("DEV.QUEUE." + i, 500); - if (!"null".equals(stray)) { - diagnosis.append("Unexpected message on DEV.QUEUE.").append(i).append(": '").append(stray) - .append("'\n"); - } - } - - diagnosis.append( - "This suggests the reply was silently not sent, or was sent to wrong destination, or body format issue.\n"); - diagnosis.append( - "Check TRACE logs in target/surefire-reports/ for EndpointMessageListener onMessage.sendReply lines."); - fail(diagnosis.toString()); - } - + String reply + = consumer.receiveBody("jms:queue:DEV.QUEUE.2", 20000, String.class); assertThat(reply).isEqualTo("My name is Camel"); - - MockEndpoint.assertIsSatisfied(context); - } - - /** - * Inspects the EndpointMessageListener to check its replyToDestination, disableReplyTo, and template configuration. - */ - private String inspectListenerState() { - try { - // Get the consumer from the route - org.apache.camel.Route route = context.getRoute("request"); - if (route == null) { - return "ERROR: route 'request' not found"; - } - - JmsConsumer jmsConsumer = (JmsConsumer) route.getConsumer(); - if (jmsConsumer == null) { - return "ERROR: consumer is null"; - } - - JmsEndpoint endpoint = (JmsEndpoint) route.getEndpoint(); - - StringBuilder state = new StringBuilder(); - state.append("JmsEndpoint config: replyTo='").append(endpoint.getConfiguration().getReplyTo()) - .append("', disableReplyTo=").append(endpoint.getConfiguration().isDisableReplyTo()); - - EndpointMessageListener listener = jmsConsumer.getEndpointMessageListener(); - if (listener == null) { - state.append(", listener=null"); - return state.toString(); - } - - state.append(", listener.replyToDestination=").append(listener.getReplyToDestination()); - state.append(" (type=") - .append(listener.getReplyToDestination() != null - ? listener.getReplyToDestination().getClass().getName() - : "null") - .append(")"); - state.append(", listener.disableReplyTo=").append(listener.isDisableReplyTo()); - - JmsOperations replyTemplate = listener.getTemplate(); - state.append(", template=").append(replyTemplate != null ? replyTemplate.getClass().getName() : "null"); - - return state.toString(); - } catch (Exception e) { - return "ERROR inspecting listener: " + e; - } - } - - /** - * Tests whether the reply JmsTemplate can successfully send to DEV.QUEUE.2 (the same template the - * EndpointMessageListener uses for replies). - */ - private String testReplyTemplate() { - try { - org.apache.camel.Route route = context.getRoute("request"); - JmsConsumer jmsConsumer = (JmsConsumer) route.getConsumer(); - EndpointMessageListener listener = jmsConsumer.getEndpointMessageListener(); - JmsOperations replyTemplate = listener.getTemplate(); - - // Send a test message using the same template the reply mechanism uses - replyTemplate.send("DEV.QUEUE.2", session -> { - TextMessage msg = session.createTextMessage("template-test-msg"); - LOG.info("=== DIAGNOSTIC: Reply template creating message in session {} ===", - session.getClass().getName()); - return msg; - }); - LOG.info("=== DIAGNOSTIC: Reply template send to DEV.QUEUE.2 succeeded ==="); - - // Read it back - String received = rawJmsReceive("DEV.QUEUE.2", 5000); - return "SEND_OK, received_back='" + received + "'"; - } catch (Exception e) { - LOG.error("=== DIAGNOSTIC: Reply template send FAILED ===", e); - return "SEND_FAILED: " + e.getClass().getName() + ": " + e.getMessage(); - } - } - - private void verifyQueueAccessible(String queueName) throws JMSException { - ConnectionFactory cf = ConnectionFactoryHelper.createConnectionFactory( - service.queueManager(), service.channel(), service.listenerPort()); - try (Connection conn = cf.createConnection()) { - conn.start(); - try (Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE)) { - Queue queue = session.createQueue(queueName); - try (MessageProducer producer = session.createProducer(queue)) { - TextMessage msg = session.createTextMessage("diagnostic-test-" + queueName); - producer.send(msg); - LOG.info("=== DIAGNOSTIC: Successfully sent to {} ===", queueName); - } - try (MessageConsumer jmsConsumer = session.createConsumer(queue)) { - jakarta.jms.Message received = jmsConsumer.receive(5000); - if (received instanceof TextMessage tm) { - LOG.info("=== DIAGNOSTIC: Successfully received from {}: '{}' ===", - queueName, tm.getText()); - } else { - LOG.warn("=== DIAGNOSTIC: Received non-text or null from {}: {} ===", - queueName, received); - } - } - } - } - } - - private String rawJmsReceive(String queueName, long timeoutMs) { - try { - ConnectionFactory cf = ConnectionFactoryHelper.createConnectionFactory( - service.queueManager(), service.channel(), service.listenerPort()); - try (Connection conn = cf.createConnection()) { - conn.start(); - try (Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); - MessageConsumer jmsConsumer = session.createConsumer(session.createQueue(queueName))) { - jakarta.jms.Message received = jmsConsumer.receive(timeoutMs); - if (received instanceof TextMessage tm) { - return tm.getText(); - } - return received != null ? received.getClass().getName() : "null"; - } - } - } catch (Exception e) { - LOG.error("=== DIAGNOSTIC: Raw JMS receive failed on {} ===", queueName, e); - return "ERROR: " + e.getMessage(); - } } @Override @@ -330,35 +54,10 @@ class JmsReplyToIbmMQTest extends CamelTestSupport { return new RouteBuilder() { @Override public void configure() { - // Main reply-to route: consumes from DEV.QUEUE.1, reply goes to DEV.QUEUE.2 from("jms:queue:DEV.QUEUE.1?replyTo=queue:DEV.QUEUE.2") .routeId("request") - .to("log:hello?showAll=true&multiline=true") - .process(exchange -> { - String body = exchange.getIn().getBody(String.class); - LOG.info("=== DIAGNOSTIC: Route processing message, body='{}', pattern={} ===", - body, exchange.getPattern()); - routeReceivedBody.set(body); - routeExchangePattern.set(exchange.getPattern()); - }) - .transform(constant("My name is Camel")) - .process(exchange -> { - LOG.info( - "=== DIAGNOSTIC: After transform, body='{}', pattern={}, hasOut={} ===", - exchange.getMessage().getBody(), - exchange.getPattern(), - exchange.hasOut()); - routeProcessedLatch.countDown(); - }) - .to("mock:processed"); - - // Manual bypass route: consumes from DEV.QUEUE.3 (no replyTo), - // transforms, and explicitly sends to DEV.QUEUE.2 via Camel producer. - // This tests that producing to DEV.QUEUE.2 works (bypassing reply-to mechanism). - from("jms:queue:DEV.QUEUE.3") - .routeId("manual-reply") - .transform(constant("My name is Camel")) - .to("jms:queue:DEV.QUEUE.2"); + .to("log:hello") + .transform(constant("My name is Camel")); } }; } @@ -366,17 +65,10 @@ class JmsReplyToIbmMQTest extends CamelTestSupport { @Override protected CamelContext createCamelContext() throws Exception { CamelContext camelContext = super.createCamelContext(); - ConnectionFactory connectionFactory = ConnectionFactoryHelper.createConnectionFactory( - service.queueManager(), service.channel(), service.listenerPort()); - JmsComponent jms = jmsComponentAutoAcknowledge(connectionFactory); - // Install a custom error handler to capture the actual exception - // instead of Spring DMLC silently logging it - jms.getConfiguration().setErrorHandler(t -> { - LOG.error("=== DIAGNOSTIC: JMS error handler caught exception ===", t); - listenerError.set(t); - errorLatch.countDown(); - }); - camelContext.addComponent("jms", jms); + ConnectionFactory connectionFactory + = ConnectionFactoryHelper.createConnectionFactory( + service.queueManager(), service.channel(), service.listenerPort()); + camelContext.addComponent("jms", jmsComponentAutoAcknowledge(connectionFactory)); return camelContext; } } diff --git a/components/camel-jms/src/test/resources/log4j2-test.properties b/components/camel-jms/src/test/resources/log4j2-test.properties index 5748418d7431..a521512aad36 100644 --- a/components/camel-jms/src/test/resources/log4j2-test.properties +++ b/components/camel-jms/src/test/resources/log4j2-test.properties @@ -27,10 +27,3 @@ appender.stdout.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n rootLogger.level = INFO rootLogger.appenderRef.out.ref = out -rootLogger.appenderRef.stdout.ref = stdout - -# Enable TRACE for EndpointMessageListener to see the reply flow details -logger.eml.name = org.apache.camel.component.jms.EndpointMessageListener -logger.eml.level = TRACE -logger.eml.appenderRef.out.ref = out -logger.eml.appenderRef.stdout.ref = stdout
