shashank created CAMEL-25272:
--------------------------------
Summary: 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
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)
