fix: support yutong mqtt tls hostname override

modules/apps/yutong-mqtt-app/src/main/resources/application.yml

@@ -47,6 +47,7 @@ lingniu:
             ca-pem: ${YUTONG_MQTT_TLS_CA_PEM:}
             client-pem: ${YUTONG_MQTT_TLS_CLIENT_PEM:}
             client-key: ${YUTONG_MQTT_TLS_CLIENT_KEY:}
+            hostname-verification-enabled: ${YUTONG_MQTT_TLS_HOSTNAME_VERIFICATION_ENABLED:true}
     pipeline:
       disruptor:
         ring-buffer-size: ${PIPELINE_RING_BUFFER_SIZE:131072}

modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttEndpointManager.java

@@ -3,6 +3,7 @@ package com.lingniu.ingest.inbound.mqtt.client;
 import com.hivemq.client.mqtt.datatypes.MqttQos;
 import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
 import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;
+import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck;
 import com.lingniu.ingest.api.ProtocolId;
 import com.lingniu.ingest.api.pipeline.RawFrame;
 import com.lingniu.ingest.core.dispatcher.Dispatcher;
@@ -25,6 +26,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -62,10 +64,14 @@ public final class MqttEndpointManager implements AutoCloseable {
     }
 
     public void start() {
+        log.info("mqtt endpoint manager starting, endpoints={}", props.getEndpoints().size());
         for (MqttInboundProperties.Endpoint ep : props.getEndpoints()) {
             try {
                 // 多 endpoint 互不影响：单个连接初始化失败会派发 operationalError，但不会阻止其他 endpoint。
+                log.info("mqtt endpoint [{}] initializing uri={} topic={} qos={} tls={}",
+                        ep.getName(), ep.getUri(), ep.getTopic(), ep.getQos(), hasCustomTls(ep.getTls()));
                 Mqtt3AsyncClient client = buildClient(ep);
+                log.info("mqtt endpoint [{}] client built", ep.getName());
                 clients.add(client);
                 connectAndSubscribe(client, ep);
             } catch (Exception e) {
@@ -116,7 +122,10 @@ public final class MqttEndpointManager implements AutoCloseable {
                     .password(ep.getPassword() == null ? new byte[0] : ep.getPassword().getBytes())
                     .applySimpleAuth();
         }
-        connectBuilder.send().whenComplete((ack, err) -> {
+        log.info("mqtt endpoint [{}] connect sending", ep.getName());
+        CompletableFuture<Mqtt3ConnAck> timeoutProbe = connectBuilder.send()
+                .orTimeout(Math.max(15, ep.getConnectionTimeoutSeconds() + 5L), TimeUnit.SECONDS);
+        timeoutProbe.whenComplete((ack, err) -> {
             if (err != null) {
                 log.error("mqtt endpoint [{}] connect failed", ep.getName(), err);
                 publishOperationalError(ep, "connect", err.getMessage() == null ? err.getClass().getSimpleName() : err.getMessage());

modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/client/MqttTlsSupport.java

@@ -34,6 +34,9 @@ public final class MqttTlsSupport {
             if (tls != null && hasText(tls.getClientPem()) && hasText(tls.getClientKey())) {
                 builder.keyManagerFactory(keyManagerFactory(Path.of(tls.getClientPem()), Path.of(tls.getClientKey())));
             }
+            if (tls != null && !tls.isHostnameVerificationEnabled()) {
+                builder.hostnameVerifier((hostname, session) -> true);
+            }
             return builder.build();
         } catch (Exception e) {
             throw new IllegalStateException("mqtt tls config build failed", e);

modules/inbound/inbound-mqtt/src/main/java/com/lingniu/ingest/inbound/mqtt/config/MqttInboundProperties.java

@@ -74,11 +74,14 @@ public class MqttInboundProperties {
         private String caPem;
         private String clientPem;
         private String clientKey;
+        private boolean hostnameVerificationEnabled = true;
         public String getCaPem() { return caPem; }
         public void setCaPem(String caPem) { this.caPem = caPem; }
         public String getClientPem() { return clientPem; }
         public void setClientPem(String clientPem) { this.clientPem = clientPem; }
         public String getClientKey() { return clientKey; }
         public void setClientKey(String clientKey) { this.clientKey = clientKey; }
+        public boolean isHostnameVerificationEnabled() { return hostnameVerificationEnabled; }
+        public void setHostnameVerificationEnabled(boolean hostnameVerificationEnabled) { this.hostnameVerificationEnabled = hostnameVerificationEnabled; }
     }
 }

modules/inbound/inbound-mqtt/src/test/java/com/lingniu/ingest/inbound/mqtt/MqttTlsSupportTest.java

@@ -46,6 +46,20 @@ class MqttTlsSupportTest {
 
         assertThat(config.getTrustManagerFactory()).isPresent();
         assertThat(config.getKeyManagerFactory()).isPresent();
+        assertThat(config.getHostnameVerifier()).isEmpty();
+    }
+
+    @Test
+    void canDisableHostnameVerificationForLegacyMutualTlsBrokers() throws Exception {
+        Path ca = tempDir.resolve("ca.pem");
+        Files.writeString(ca, CERT);
+        MqttInboundProperties.Tls tls = new MqttInboundProperties.Tls();
+        tls.setCaPem(ca.toString());
+        tls.setHostnameVerificationEnabled(false);
+
+        MqttClientSslConfig config = MqttTlsSupport.build(tls);
+
+        assertThat(config.getHostnameVerifier()).isPresent();
     }
 
     private static final String CERT = """