[
https://issues.apache.org/jira/browse/CAMEL-25293?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25293:
--------------------------------
Fix Version/s: 4.23.0
> 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
> Priority: Minor
> Fix For: 4.23.0
>
>
> {{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)
