Croway commented on code in PR #25905: URL: https://github.com/apache/camel/pull/25905#discussion_r3894207638
########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQEndpoint.java: ########## @@ -0,0 +1,132 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.hivemq; + +import java.nio.charset.StandardCharsets; + +import com.hivemq.client.mqtt.MqttClient; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientBuilder; +import com.hivemq.client.mqtt.mqtt5.message.auth.Mqtt5SimpleAuth; +import com.hivemq.client.mqtt.mqtt5.message.auth.Mqtt5SimpleAuthBuilder; +import org.apache.camel.Category; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.spi.EndpointServiceLocation; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriEndpoint; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriPath; +import org.apache.camel.support.DefaultEndpoint; + +@UriEndpoint(firstVersion = "4.23.0", scheme = "hivemq", title = "HiveMQ", syntax = "hivemq:topic", + category = { Category.MESSAGING, Category.IOT }, headersClass = HiveMQConstants.class) +public class HiveMQEndpoint extends DefaultEndpoint implements EndpointServiceLocation { + + /** + * The MQTT topic name or pattern to subscribe to or publish on. + */ + @UriPath + @Metadata(required = true) + private String topic; + + /** + * The HiveMQ component configuration options. + */ + @UriParam + @Metadata(description = "To use a custom HiveMQConfiguration") + private HiveMQConfiguration configuration; + + public HiveMQEndpoint(String uri, HiveMQComponent component, HiveMQConfiguration configuration, String topic) { + super(uri, component); + this.configuration = configuration; + this.topic = topic; + } + + @Override + public Producer createProducer() throws Exception { + return new HiveMQProducer(this); + } + + @Override + public Consumer createConsumer(Processor processor) throws Exception { + HiveMQConsumer consumer = new HiveMQConsumer(this, processor); + configureConsumer(consumer); + return consumer; + } + + public Mqtt5AsyncClient createClient() { + Mqtt5ClientBuilder builder = MqttClient.builder() + .serverHost(configuration.getHost()) + .serverPort(configuration.getPort()) + .automaticReconnectWithDefaultConfig() + .useMqttVersion5(); + + if (configuration.getClientId() != null) { + builder.identifier(configuration.getClientId()); + } + + if (configuration.isSsl()) { + builder.sslWithDefaultConfig(); + } + + if (configuration.getUsername() != null) { + Mqtt5SimpleAuthBuilder.Complete authBuilder + = Mqtt5SimpleAuth.builder().username(configuration.getUsername()); + if (configuration.getPassword() != null) { + authBuilder.password(configuration.getPassword().getBytes(StandardCharsets.UTF_8)); + } + builder.simpleAuth(authBuilder.build()); + } + + return builder.buildAsync(); + } + + public void connect(Mqtt5AsyncClient client) { + client.connectWith() + .cleanStart(configuration.isCleanStart()) + .send() + .join(); Review Comment: This `join()` has no timeout, and with `automaticReconnectWithDefaultConfig()` enabled the connect future **never completes** when the broker is unreachable — the client keeps retrying internally and the caller blocks forever. This is long-standing documented behavior of the client: hivemq/hivemq-mqtt-client#302 (closed as stale, not fixed) and the still-open follow-up hivemq/hivemq-mqtt-client#622. Since this is called from `doStart()` of both the consumer and the producer, a broker that is down at deployment time now hangs Camel context startup indefinitely — the reply on the earlier thread mentioned a bounded connection timeout, but none is set in the code. Suggestion: use `client.connectWith().cleanStart(...).send().toCompletableFuture().get(timeout, TimeUnit.SECONDS)` (or `orTimeout`) with a sensible default, and surface the timeout as a `connectTimeout` URI option. On timeout, call `disconnect()` so the auto-reconnector does not keep running behind a failed start. _Claude Code on behalf of Croway_ ########## components/camel-hivemq/src/main/java/org/apache/camel/component/hivemq/HiveMQConsumer.java: ########## @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.hivemq; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicBoolean; + +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.support.DefaultConsumer; + +public class HiveMQConsumer extends DefaultConsumer { + + private final HiveMQEndpoint endpoint; + private Mqtt5AsyncClient client; + private ExecutorService executor; + + public HiveMQConsumer(HiveMQEndpoint endpoint, Processor processor) { + super(endpoint, processor); + this.endpoint = endpoint; + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + executor = endpoint.getCamelContext().getExecutorServiceManager().newDefaultThreadPool(this, "HiveMQConsumer"); + client = endpoint.createClient(); + endpoint.connect(client); + + client.subscribeWith() + .topicFilter(endpoint.getTopic()) + .qos(endpoint.getConfiguration().getQos()) + .callback(this::onMessage) + .send() + .join(); + } + + @Override + protected void doStop() throws Exception { + if (client != null && client.getState().isConnected()) { Review Comment: `client.getState().isConnected()` is only true in the `CONNECTED` state. With automatic reconnect enabled, a client that is currently in `DISCONNECTED_RECONNECT` / `CONNECTING_RECONNECT` (broker briefly down, network blip) fails this check, so `disconnect()` is skipped — and the auto-reconnector stays alive: the client will reconnect and resubscribe **after** Camel has stopped the route, leaking a connection and continuing to consume messages that no route processes. Suggestion: call `client.disconnect()` whenever the state is not `DISCONNECTED` (or unconditionally and tolerate the already-disconnected error), so the reconnector is cancelled. The same pattern is in `HiveMQProducer.doStop()`. _Claude Code on behalf of Croway_ -- 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]
