shashank created CAMEL-25293:
--------------------------------
Summary: camel-hivemq - the consumer discards the messages it
already acknowledged, and interrupts the ones being processed, when it stops
Key: CAMEL-25293
URL: https://issues.apache.org/jira/browse/CAMEL-25293
Project: Camel
Issue Type: Bug
Reporter: shashank
{{HiveMQConsumer}} subscribes with a callback that only creates the exchange
and submits it to the consumer's thread pool. The HiveMQ client sends the
PUBACK (QoS 1) or completes the QoS 2 flow when the callback returns, so the
broker considers the message delivered before the route has processed it.
{{doStop}} then shuts that pool down with
{{ExecutorServiceManager.shutdownNow}}:
* the messages still queued in the pool (up to the default profile's queue of
1000) are discarded without any log;
* the threads processing a message are interrupted, so these exchanges fail (or
their side effects stop half way).
{{HiveMQConsumer}} is neither {{Suspendable}} nor {{ShutdownAware}}, so on a
graceful shutdown or a route stop the shutdown strategy stops it immediately
and then only waits for the in-flight exchanges, which do not include the
queued ones. Every route stop, redeploy or shutdown under load therefore loses
QoS 1/2 messages that the broker will not send again.
h3. Reproduction
{{HiveMQConsumerStopTest}} (no broker: the endpoint gets a fake
{{HiveMQClientAdapter}} that keeps the subscribe callback, and the context a
thread pool factory that gives the consumer a pool of one thread): three
messages are delivered, the first one is being processed (it waits until the
pool is shut down), then the route is stopped. On main none of the three is
processed ({{Expecting actual: [] to contain exactly ["1", "2", "3"]}}): the
first is interrupted, the other two are discarded.
h3. Proposed fix
Keep the order of {{doStop}} (unsubscribe, stop the client, so no new message
arrives) and shut the pool down with {{shutdownGraceful}}, so the received
messages complete; the ExecutorServiceManager's shutdown await termination (10
seconds by default) still bounds the wait: after it, the pool is shut down at
once as before (running exchanges interrupted, still queued messages discarded,
with a WARN). Other consumers with their own pool do the same
({{KafkaConsumer}}, {{GooglePubsubConsumer}}, {{IggyConsumer}}). With the fix
the new test and the camel-hivemq unit tests pass (24 tests); the integration
tests need Docker and were not run.
The component documentation (section "Acknowledgement") says that the
acknowledgement is not tied to the success or failure of the route, so a
message whose exchange fails is not redelivered. That is about failed
exchanges; here the messages are dropped without reaching the route or its
error handler at all, and without a log, on an ordinary route stop or graceful
shutdown, which other Camel consumers complete.
Not changed: a crash still loses the messages queued in the pool, because they
are acknowledged before processing. Acknowledging after processing would need
the client's manual acknowledgement, and gives nothing with the default
{{cleanStart=true}} (or an MQTT 5 session that expires on disconnect); it can
be a separate improvement.
Affected: main only (camel-hivemq is new in 4.23, not released yet), so no
upgrade note.
Duplicate check (2026-10-02): JIRA text "hivemq" (6 issues: CAMEL-23503, 24998,
24999, 25134, 24770, 21577), none about the consumer's stop. GitHub pull
requests "hivemq": the component (#26006), MQTT 3.1.1 support (#27104, merged),
the header filter and ssl annotation PRs, test-infra and dependency bumps.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
