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 a0e6de82c9166040d848e7375b772e7b9d586ecc Author: Guillaume Nodet <[email protected]> AuthorDate: Thu Jul 30 16:32:52 2026 +0200 chore: improve IBM MQ reply-to diagnostics with DLQ check and queue sweep Remove LANG=C workaround (disproven hypothesis), add manual-bypass test using separate queues (DEV.QUEUE.3→4) to isolate reply-to mechanism, add Dead Letter Queue and queue sweep diagnostics to pinpoint where the reply message ends up with MQ 10.0. Co-Authored-By: Claude Opus 4.6 <[email protected]> --- .../component/jms/issues/JmsReplyToIbmMQTest.java | 114 +++++++++++++++------ .../services/IbmMQLocalContainerInfraService.java | 5 - 2 files changed, 84 insertions(+), 35 deletions(-) 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 c31ecd12aad3..97fa4bb2e48b 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 @@ -30,6 +30,7 @@ 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.JmsComponent; import org.apache.camel.component.jms.JmsTestHelper; @@ -38,7 +39,10 @@ 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.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; @@ -48,13 +52,18 @@ import static org.apache.camel.component.jms.JmsComponent.jmsComponentAutoAcknow 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") -public class JmsReplyToIbmMQTest extends CamelTestSupport { +@TestMethodOrder(MethodOrderer.OrderAnnotation.class) +class JmsReplyToIbmMQTest extends CamelTestSupport { private static final Logger LOG = LoggerFactory.getLogger(JmsReplyToIbmMQTest.class); @RegisterExtension - public static IbmMQService service = IbmMQServiceFactory.createService(); + static IbmMQService service = IbmMQServiceFactory.createService(); // Captures any error thrown during reply sending (normally swallowed by Spring DMLC) private final AtomicReference<Throwable> listenerError = new AtomicReference<>(); @@ -63,8 +72,34 @@ public class JmsReplyToIbmMQTest extends CamelTestSupport { // 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<>(); + + /** + * 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.4 via .to(). If this passes but testCustomJMSReplyToInOut + * fails, the issue is in EndpointMessageListener's reply mechanism. + */ + @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.4 (20s timeout) ==="); + String reply = consumer.receiveBody("jms:queue:DEV.QUEUE.4", 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); @@ -75,52 +110,64 @@ public class JmsReplyToIbmMQTest extends CamelTestSupport { LOG.info("=== DIAGNOSTIC: Verifying DEV.QUEUE.2 is accessible via raw JMS ==="); verifyQueueAccessible("DEV.QUEUE.2"); - // Step 2: Also verify DEV.QUEUE.1 works - LOG.info("=== DIAGNOSTIC: Verifying DEV.QUEUE.1 is accessible via raw JMS ==="); - verifyQueueAccessible("DEV.QUEUE.1"); - - // Step 3: Send the test message via Camel + // 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 4: Wait for the route to process the message + // 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: '{}' ===", - routeProcessed, routeReceivedBody.get()); + 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 5: Now try to receive the reply from DEV.QUEUE.2 + // 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) { - // Check if the listener error handler captured an exception + // Build detailed diagnostic report StringBuilder diagnosis = new StringBuilder(); diagnosis.append("Reply was null on DEV.QUEUE.2.\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); + 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); - } else { - // No error handler was triggered — maybe the reply was silently not sent - // Try a raw JMS receive on DEV.QUEUE.2 to see if message is there with different format - LOG.info("=== DIAGNOSTIC: No error captured. Trying raw JMS receive on DEV.QUEUE.2 ==="); - String rawReply = rawJmsReceive("DEV.QUEUE.2", 3000); - diagnosis.append("No listener error was captured within timeout.\n"); - diagnosis.append("Raw JMS receive on DEV.QUEUE.2: '").append(rawReply).append("'\n"); - diagnosis.append("This suggests the reply was silently not sent, " - + "or was sent to wrong destination, or body format issue"); - fail(diagnosis.toString()); } + + // 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 <= 5; 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"); + fail(diagnosis.toString()); } assertThat(reply).isEqualTo("My name is Camel"); @@ -135,15 +182,13 @@ public class JmsReplyToIbmMQTest extends CamelTestSupport { conn.start(); try (Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE)) { Queue queue = session.createQueue(queueName); - // Send a test message try (MessageProducer producer = session.createProducer(queue)) { TextMessage msg = session.createTextMessage("diagnostic-test-" + queueName); producer.send(msg); LOG.info("=== DIAGNOSTIC: Successfully sent to {} ===", queueName); } - // Receive it back - try (MessageConsumer consumer = session.createConsumer(queue)) { - jakarta.jms.Message received = consumer.receive(5000); + 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()); @@ -163,8 +208,8 @@ public class JmsReplyToIbmMQTest extends CamelTestSupport { try (Connection conn = cf.createConnection()) { conn.start(); try (Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE); - MessageConsumer consumer = session.createConsumer(session.createQueue(queueName))) { - jakarta.jms.Message received = consumer.receive(timeoutMs); + MessageConsumer jmsConsumer = session.createConsumer(session.createQueue(queueName))) { + jakarta.jms.Message received = jmsConsumer.receive(timeoutMs); if (received instanceof TextMessage tm) { return tm.getText(); } @@ -172,7 +217,7 @@ public class JmsReplyToIbmMQTest extends CamelTestSupport { } } } catch (Exception e) { - LOG.error("=== DIAGNOSTIC: Raw JMS receive failed ===", e); + LOG.error("=== DIAGNOSTIC: Raw JMS receive failed on {} ===", queueName, e); return "ERROR: " + e.getMessage(); } } @@ -182,6 +227,7 @@ public 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") @@ -190,6 +236,7 @@ public class JmsReplyToIbmMQTest extends CamelTestSupport { 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 -> { @@ -201,6 +248,13 @@ public class JmsReplyToIbmMQTest extends CamelTestSupport { routeProcessedLatch.countDown(); }) .to("mock:processed"); + + // Manual bypass route: consumes from DEV.QUEUE.3 (no replyTo), + // transforms, and explicitly sends to DEV.QUEUE.4 (separate queue to avoid interference) + from("jms:queue:DEV.QUEUE.3") + .routeId("manual-reply") + .transform(constant("My name is Camel")) + .to("jms:queue:DEV.QUEUE.4"); } }; } diff --git a/test-infra/camel-test-infra-ibmmq/src/main/java/org/apache/camel/test/infra/ibmmq/services/IbmMQLocalContainerInfraService.java b/test-infra/camel-test-infra-ibmmq/src/main/java/org/apache/camel/test/infra/ibmmq/services/IbmMQLocalContainerInfraService.java index 5f4a6f2dcf84..41b577c539a1 100644 --- a/test-infra/camel-test-infra-ibmmq/src/main/java/org/apache/camel/test/infra/ibmmq/services/IbmMQLocalContainerInfraService.java +++ b/test-infra/camel-test-infra-ibmmq/src/main/java/org/apache/camel/test/infra/ibmmq/services/IbmMQLocalContainerInfraService.java @@ -70,11 +70,6 @@ public class IbmMQLocalContainerInfraService implements IbmMQInfraService, Conta .withEnv("LICENSE", "accept") .withEnv("MQ_QMGR_NAME", IbmMQProperties.DEFAULT_QMGR_NAME) .withEnv("MQ_APP_PASSWORD", IbmMQProperties.DEFAULT_APP_PASSWORD) - // MQ 10.0 changed the default locale from C to C.utf8, which causes the - // queue manager to use CCSID 1208 (UTF-8) instead of 819 (ISO 8859-1). - // Force C locale to preserve CCSID 819 until the JMS reply-to mechanism - // is confirmed to work correctly with CCSID 1208. - .withEnv("LANG", "C") .withLogConsumer(new Slf4jLogConsumer(LOG)) // AND the listener-port and log-message checks; a plain chained waitingFor() would replace, // not combine, the strategies. WITH_INDIVIDUAL_TIMEOUTS_ONLY keeps each strategy's own timeout
