[
https://issues.apache.org/jira/browse/CAMEL-25853?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25853:
-----------------------------
Description:
The toD optimisation of camel-paho, camel-paho-mqtt5 and camel-hivemq
({{PahoSendDynamicAware}}, {{PahoMqtt5SendDynamicAware}},
{{HiveMQSendDynamicAware}}) sends to one static endpoint and puts the resolved
topic in the override topic header ({{CamelMqttOverrideTopic}} for paho and
paho-mqtt5, {{CamelHiveMQOverrideTopic}} for hivemq) with its pre processor.
There is no post processor, and the producers read the header without removing
it (unlike camel-kafka, camel-jms, camel-sjms and camel-spring-rabbitmq, whose
producers remove their override header). The pre processor keeps an override
topic it finds on the message:
{code:java}
if (exchange.getMessage().getHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC) !=
null) {
return null;
}
{code}
So every later toD of the same exchange to the same component publishes to the
topic of the first toD, without any error or log line:
{code:java}
from("direct:start")
.toD("paho:status/${header.id}")
.toD("paho:audit/${header.id}"); // published to status/<id>
from("direct:loop")
.loop(3)
.toD("paho:devices/${exchangeProperty.CamelLoopIndex}/cmd") //
devices/0/cmd three times
.end();
{code}
The same happens when the second toD is in another route called with direct,
since the header travels with the message.
*Fix*: the post processor removes the override topic header that the pre
processor has set. A header the route set itself before the toD (the documented
way to override the topic) is left as it is.
*Tests*: new {{PahoToDTwiceTest}}, {{PahoMqtt5ToDTwiceTest}} and
{{HiveMQToDTwiceTest}} (2 tests each, no broker: a mocked paho {{MqttClient}}
on the component, and a recording client for hivemq) with the two routes above.
On main all six fail, for example {{Expecting actual: ["status/1", "status/1"]
to contain exactly (and in same order): ["status/1", "audit/1"]}} and
{{["devices/0/cmd", "devices/0/cmd", "devices/0/cmd"]}}. (the camel-hivemq
test; the paho tests use JUnit assertions, as their modules, and fail with
{{expected: <[status/1, audit/1]> but was: <[status/1, status/1]>}}). All pass
with the fix. Unit tests with the fix: camel-paho 30 run, 8 errors in
{{PahoComponentTest}} and {{PahoManualAcksTest}} (route start fails with
{{Connection lost (32109) - java.io.EOFException}} against the embedded broker;
the same 8 errors without the fix in my environment), camel-paho-mqtt5 19 run,
0 failures, camel-hivemq 26 run, 0 failures.
*How it was found*: a review of the {{SendDynamicAware}} implementations,
comparing what each one leaves on the message after the send with what its
producer reads. The JUnit tests above are the evidence; it was not run against
an MQTT broker.
*Affected*: camel-paho and camel-paho-mqtt5 on main, camel-4.23.x,
camel-4.22.x, camel-4.18.x and camel-4.14.x (same pre processor, no post
processor, the producer does not remove the header); camel-hivemq on main and
camel-4.23.x. Read, not run. The paho optimisation came with CAMEL-15992.
_Filed with Claude Code on behalf of allthingssecurity._
was:
The toD optimisation of camel-paho, camel-paho-mqtt5 and camel-hivemq
({{PahoSendDynamicAware}}, {{PahoMqtt5SendDynamicAware}},
{{HiveMQSendDynamicAware}}) sends to one static endpoint and puts the resolved
topic in the override topic header ({{CamelMqttOverrideTopic}} for paho and
paho-mqtt5, {{CamelHiveMQOverrideTopic}} for hivemq) with its pre processor.
There is no post processor, and the producers read the header without removing
it (unlike camel-kafka, camel-jms, camel-sjms and camel-spring-rabbitmq, whose
producers remove their override header). The pre processor keeps an override
topic it finds on the message:
{code:java}
if (exchange.getMessage().getHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC) !=
null) {
return null;
}
{code}
So every later toD of the same exchange to the same component publishes to the
topic of the first toD, without any error or log line:
{code:java}
from("direct:start")
.toD("paho:status/${header.id}")
.toD("paho:audit/${header.id}"); // published to status/<id>
from("direct:loop")
.loop(3)
.toD("paho:devices/${exchangeProperty.CamelLoopIndex}/cmd") //
devices/0/cmd three times
.end();
{code}
The same happens when the second toD is in another route called with direct,
since the header travels with the message.
*Fix*: the post processor removes the override topic header that the pre
processor has set. A header the route set itself before the toD (the documented
way to override the topic) is left as it is.
*Tests*: new {{PahoToDTwiceTest}}, {{PahoMqtt5ToDTwiceTest}} and
{{HiveMQToDTwiceTest}} (2 tests each, no broker: a mocked paho {{MqttClient}}
on the component, and a recording client for hivemq) with the two routes above.
On main all six fail, for example {{Expecting actual: ["status/1", "status/1"]
to contain exactly (and in same order): ["status/1", "audit/1"]}} and
{{["devices/0/cmd", "devices/0/cmd", "devices/0/cmd"]}}. All pass with the fix.
Unit tests with the fix: camel-paho 30 run, 8 errors in {{PahoComponentTest}}
and {{PahoManualAcksTest}} (route start fails with {{Connection lost (32109) -
java.io.EOFException}} against the embedded broker; the same 8 errors without
the fix in my environment), camel-paho-mqtt5 19 run, 0 failures, camel-hivemq
26 run, 0 failures.
*How it was found*: a review of the {{SendDynamicAware}} implementations,
comparing what each one leaves on the message after the send with what its
producer reads. The JUnit tests above are the evidence; it was not run against
an MQTT broker.
*Affected*: camel-paho and camel-paho-mqtt5 on main, camel-4.23.x,
camel-4.22.x, camel-4.18.x and camel-4.14.x (same pre processor, no post
processor, the producer does not remove the header); camel-hivemq on main and
camel-4.23.x. Read, not run. The paho optimisation came with CAMEL-15992.
_Filed with Claude Code on behalf of allthingssecurity._
> camel-paho / camel-paho-mqtt5 / camel-hivemq - a second toD of the same
> exchange publishes to the topic of the first toD
> ------------------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-25853
> URL: https://issues.apache.org/jira/browse/CAMEL-25853
> Project: Camel
> Issue Type: Bug
> Components: camel-paho, camel-paho-mqtt5
> Reporter: shashank
> Assignee: shashank
> Priority: Major
>
> The toD optimisation of camel-paho, camel-paho-mqtt5 and camel-hivemq
> ({{PahoSendDynamicAware}}, {{PahoMqtt5SendDynamicAware}},
> {{HiveMQSendDynamicAware}}) sends to one static endpoint and puts the
> resolved topic in the override topic header ({{CamelMqttOverrideTopic}} for
> paho and paho-mqtt5, {{CamelHiveMQOverrideTopic}} for hivemq) with its pre
> processor. There is no post processor, and the producers read the header
> without removing it (unlike camel-kafka, camel-jms, camel-sjms and
> camel-spring-rabbitmq, whose producers remove their override header). The pre
> processor keeps an override topic it finds on the message:
> {code:java}
> if (exchange.getMessage().getHeader(PahoConstants.CAMEL_PAHO_OVERRIDE_TOPIC)
> != null) {
> return null;
> }
> {code}
> So every later toD of the same exchange to the same component publishes to
> the topic of the first toD, without any error or log line:
> {code:java}
> from("direct:start")
> .toD("paho:status/${header.id}")
> .toD("paho:audit/${header.id}"); // published to status/<id>
> from("direct:loop")
> .loop(3)
> .toD("paho:devices/${exchangeProperty.CamelLoopIndex}/cmd") //
> devices/0/cmd three times
> .end();
> {code}
> The same happens when the second toD is in another route called with direct,
> since the header travels with the message.
> *Fix*: the post processor removes the override topic header that the pre
> processor has set. A header the route set itself before the toD (the
> documented way to override the topic) is left as it is.
> *Tests*: new {{PahoToDTwiceTest}}, {{PahoMqtt5ToDTwiceTest}} and
> {{HiveMQToDTwiceTest}} (2 tests each, no broker: a mocked paho {{MqttClient}}
> on the component, and a recording client for hivemq) with the two routes
> above. On main all six fail, for example {{Expecting actual: ["status/1",
> "status/1"] to contain exactly (and in same order): ["status/1", "audit/1"]}}
> and {{["devices/0/cmd", "devices/0/cmd", "devices/0/cmd"]}}. (the
> camel-hivemq test; the paho tests use JUnit assertions, as their modules, and
> fail with {{expected: <[status/1, audit/1]> but was: <[status/1,
> status/1]>}}). All pass with the fix. Unit tests with the fix: camel-paho 30
> run, 8 errors in {{PahoComponentTest}} and {{PahoManualAcksTest}} (route
> start fails with {{Connection lost (32109) - java.io.EOFException}} against
> the embedded broker; the same 8 errors without the fix in my environment),
> camel-paho-mqtt5 19 run, 0 failures, camel-hivemq 26 run, 0 failures.
> *How it was found*: a review of the {{SendDynamicAware}} implementations,
> comparing what each one leaves on the message after the send with what its
> producer reads. The JUnit tests above are the evidence; it was not run
> against an MQTT broker.
> *Affected*: camel-paho and camel-paho-mqtt5 on main, camel-4.23.x,
> camel-4.22.x, camel-4.18.x and camel-4.14.x (same pre processor, no post
> processor, the producer does not remove the header); camel-hivemq on main and
> camel-4.23.x. Read, not run. The paho optimisation came with CAMEL-15992.
> _Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
