davsclaus commented on code in PR #25767:
URL: https://github.com/apache/camel/pull/25767#discussion_r3865369850
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java:
##########
@@ -139,6 +141,23 @@ public void deliveryComplete(IMqttToken token) {
}
}
+ private void restartRouteAsync() {
Review Comment:
No guard against overlapping restarts: if `subscribe()` keeps failing across
rapid successive `connectComplete(true, ...)` invocations,
`restartRouteAsync()` can be invoked multiple times concurrently, spawning
several threads that concurrently call `stopRoute`/`startRoute` on the same
route id. Consider a simple in-flight flag (e.g. an `AtomicBoolean`) so a
restart already in progress is not duplicated.
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java:
##########
@@ -139,6 +141,23 @@ public void deliveryComplete(IMqttToken token) {
}
}
+ private void restartRouteAsync() {
+ Thread restartThread = new Thread(() -> {
Review Comment:
Minor convention note: elsewhere in the codebase (e.g. `camel-jms`
`JmsProducer`) background threads are created via
`getEndpoint().getCamelContext().getExecutorServiceManager()` rather than a raw
`new Thread(...)`. That gets you consistent thread naming/pooling and makes the
thread visible to Camel's `ShutdownStrategy`. That said, spawning a separate
thread at all is the right call here - `MqttClient` is a synchronous wrapper,
and calling `disconnect()`/`connect()` back on the same thread currently
executing the `MqttCallback` risks deadlock, since disconnect blocks waiting
for completion signaled by that same client's internal thread.
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java:
##########
@@ -139,6 +141,23 @@ public void deliveryComplete(IMqttToken token) {
}
}
+ private void restartRouteAsync() {
+ Thread restartThread = new Thread(() -> {
+ try {
+ String routeId = getRoute().getRouteId();
+ LOG.info("Stopping route {} for restart after resubscribe
failure", routeId);
+
getEndpoint().getCamelContext().getRouteController().stopRoute(routeId);
Review Comment:
Restarting the route here can silently replace a user-supplied `MqttClient`.
`doStop()` unconditionally sets `client = null` in its `finally` block
regardless of `stopClient` - so if a caller supplied their own client via
`setClient()` (e.g. for custom persistence or a custom SSL socket factory), the
next `doStart()` sees `client == null`, sets `stopClient = true`, and
creates+connects a brand-new default `MqttClient`, discarding the caller's
client with no warning.
This quirk pre-dates this PR (it's only reachable today via a manual route
restart), but this change makes it reachable automatically and repeatedly any
time resubscribe fails. Worth either guarding the auto-restart to only apply
when the consumer owns the client (`stopClient == true`), or at minimum logging
when a restart is about to replace an externally-supplied client.
##########
components/camel-paho-mqtt5/src/main/java/org/apache/camel/component/paho/mqtt5/PahoMqtt5Consumer.java:
##########
@@ -139,6 +141,23 @@ public void deliveryComplete(IMqttToken token) {
}
}
+ private void restartRouteAsync() {
+ Thread restartThread = new Thread(() -> {
+ try {
+ String routeId = getRoute().getRouteId();
Review Comment:
Nit: `getRoute().getRouteId()` can just be `getRouteId()` - already
available directly on `DefaultConsumer` (also used again on line 156 for the
thread name).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
