[
https://issues.apache.org/jira/browse/CAMEL-25272?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18122174#comment-18122174
]
Claus Ibsen commented on CAMEL-25272:
-------------------------------------
Merged via https://github.com/apache/camel/pull/27292
_Claude Code on behalf of davsclaus_
> camel-rocketmq - the consumer acknowledges the message of a failed exchange
> (the message is lost), and an InOut send without replyToTopic completes
> successfully when the send fails
> ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25272
> URL: https://issues.apache.org/jira/browse/CAMEL-25272
> Project: Camel
> Issue Type: Bug
> Components: camel-rocketmq
> Reporter: shashank
> Assignee: shashank
> Priority: Major
> Fix For: 4.23.0
>
>
> Two defects in camel-rocketmq (code unchanged since the component was added
> in CAMEL-14831).
> h3. 1. The consumer acknowledges the message of a failed exchange
> {{RocketMQConsumer}} registers a {{MessageListenerConcurrently}} whose status
> acknowledges the message:
> {code:java}
> try {
> getProcessor().process(exchange);
> } catch (Exception e) {
> getExceptionHandler().handleException(e);
> return ConsumeConcurrentlyStatus.RECONSUME_LATER;
> }
> return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
> {code}
> A route that fails does not throw from {{process}}: the error handler sets
> the exception on the exchange. So the listener answers {{CONSUME_SUCCESS}}
> for every failed exchange, the broker takes the message as consumed, and it
> is never redelivered (no retry topic, no dead letter queue). Any failure of
> the route (a database down, a validation error) loses the message, which is
> what the push consumer's {{RECONSUME_LATER}} is there to prevent.
> h3. 2. An InOut send without replyToTopic completes successfully when the
> send fails
> {{RocketMQProducer.processInOut}} is used for every exchange whose pattern is
> out capable, for example a message coming from an HTTP consumer. Without
> {{replyToTopic}} there is no reply manager ({{replyManager}} is {{null}}),
> and the send callback does:
> {code:java}
> public void onException(Throwable e) {
> try {
> replyManager.cancelMessageKey(generateKey); // NullPointerException
> exchange.setException(e);
> } finally {
> callback.done(false);
> }
> }
> {code}
> The {{NullPointerException}} skips {{setException}}, the {{finally}}
> completes the exchange, and the exchange continues as if the message had been
> sent. (The NPE ends up in the client's async send thread.)
> h3. Reproduction (unit tests, no broker)
> * {{RocketMQConsumerFailureTest}}: the listener (made a package-private
> method {{consumeMessage}} so that it can be called without a broker) with a
> processor that sends the exchange to a route that throws: main answers
> {{CONSUME_SUCCESS}}, expected {{RECONSUME_LATER}}. Controls: a processor that
> throws, and a successful route ({{CONSUME_SUCCESS}}).
> * {{RocketMQProducerSendFailureTest}}: an empty body, which the RocketMQ
> client rejects in its asynchronous send ({{MQClientException}}, "the message
> body length is zero") before it contacts the name server; any asynchronous
> send failure takes the same path. InOnly: the exchange fails (control). InOut
> without {{replyToTopic}}: on main the exchange has no exception.
> h3. Proposed fix
> * Consumer: set a thrown exception on the exchange, and answer
> {{RECONSUME_LATER}} (after reporting to the exception handler) whenever the
> exchange has an exception; {{CONSUME_SUCCESS}} otherwise.
> * Producer: set the exception first, and cancel the message key only when
> there is a reply manager.
> With the fix the 5 new tests pass; without it 2 fail. The module's other
> tests are ITs that need Docker (not run). Note that the module skips tests on
> aarch64 ({{skipTests.aarch64}}, the RocketMQ container is x86 only), so these
> unit tests run on the x86 CI only; I ran them locally with
> {{-DskipTests.aarch64=false -DskipTests=false}}. The upgrade guide for 4.23
> gets a note: messages of failed exchanges are now redelivered (with an
> increasing delay, up to the max reconsume times of the group, 16 by default,
> then the dead letter queue).
> Affected: 4.14.x, 4.18.x and main (same code).
> Duplicate check (2026-10-02): JIRA text "rocketmq" (12 issues): only the
> component, upgrades and test issues (CAMEL-14831, 19832, 19610, 19684, 20368
> batching consumers, ...). GitHub pull requests "rocketmq": none about the
> consumer status or the producer callback (open #22358 touches only the reply
> manager's {{getOut}}).
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
