apache · jt2594838 · Jul 21, 2026 · Jul 21, 2026 · Jul 21, 2026
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -945,6 +945,9 @@ public final class DataNodePipeMessages {
       "Network failed to receive tsFile %s, status: %s";
   public static final String SECURITY_DIR = "security dir: {}";
   public static final String SECURITY_PKI_DIR = "security pki dir: {}";
+  public static final String
+      LOG_OPC_UA_ENDPOINT_SELECTED_CONFIGURED_ARG_ADVERTISED_ARG_EFFECTIVE_ARG_ALLOWENDPOINTREDIRECT_ARG_4FE076CB =
+          "OPC UA endpoint selected: configured={}, advertised={}, effective={}, allowEndpointRedirect={}.";
   public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
       "When %s or %s is true, specify a complete trust-store pair under the same "
           + "alias: %s and %s, %s and %s, or %s and %s";
@@ -1535,6 +1538,14 @@ public final class DataNodePipeMessages {
   public static final String UNABLE_CREATE_SECURITY_DIR = "Unable to create security dir: ";
   public static final String OPC_UA_SECURITY_DIR = "Security dir: {}";
   public static final String OPC_UA_SECURITY_PKI_DIR = "Security pki dir: {}";
+  public static final String
+      EXCEPTION_THE_ADVERTISED_HOST_MUST_BE_A_HOSTNAME_OR_IP_ADDRESS_WITHOUT_A_SCHEME_PORT_OR_PATH_6857C67A =
+          "The advertised host must be a hostname or IP address without a scheme, port, or path.";
+  public static final String
+      LOG_ADVERTISED_HOST_ARG_IS_NOT_PRESENT_IN_THE_LOADED_OPC_UA_SERVER_CERTIFICATE_SUBJECT_ALTERNATIVE_NAMES_SECURED_CLIENTS_MAY_REJECT_IT_REPLACE_OR_REGENERATE_THE_CERTIFICATE_AND_ESTABLISH_TRUST_AGAIN_912358AF =
+          "Advertised host {} is not present in the loaded OPC UA server certificate subject "
+              + "alternative names. Secured clients may reject it; replace or regenerate the "
+              + "certificate and establish trust again.";
 
   // ---------------------------------------------------------------------------
   // pipe – PipeDataNodePluginAgent

diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -892,6 +892,9 @@ public final class DataNodePipeMessages {
       "网络接收 TsFile %s 失败，状态：%s";
   public static final String SECURITY_DIR = "security 目录：{}";
   public static final String SECURITY_PKI_DIR = "security pki 目录：{}";
+  public static final String
+      LOG_OPC_UA_ENDPOINT_SELECTED_CONFIGURED_ARG_ADVERTISED_ARG_EFFECTIVE_ARG_ALLOWENDPOINTREDIRECT_ARG_4FE076CB =
+          "已选择 OPC UA endpoint：configured={}，advertised={}，effective={}，allowEndpointRedirect={}。";
   public static final String SSL_TRUST_STORE_PAIR_REQUIRED_WHEN_SSL_ENABLED =
       "当 %s 或 %s 为 true 时，请在同一别名下指定完整的 trust-store 参数对：%s 和 %s、%s 和 %s，或 %s 和 %s";
   public static final String SSL_KEY_STORE_PATH_AND_PASSWORD_MUST_BE_SPECIFIED_TOGETHER =
@@ -1439,6 +1442,13 @@ public final class DataNodePipeMessages {
       "安全目录：{}";
   public static final String OPC_UA_SECURITY_PKI_DIR =
       "安全 PKI 目录：{}";
+  public static final String
+      EXCEPTION_THE_ADVERTISED_HOST_MUST_BE_A_HOSTNAME_OR_IP_ADDRESS_WITHOUT_A_SCHEME_PORT_OR_PATH_6857C67A =
+          "advertised host 必须是不带 scheme、port 或 path 的 hostname 或 IP 地址。";
+  public static final String
+      LOG_ADVERTISED_HOST_ARG_IS_NOT_PRESENT_IN_THE_LOADED_OPC_UA_SERVER_CERTIFICATE_SUBJECT_ALTERNATIVE_NAMES_SECURED_CLIENTS_MAY_REJECT_IT_REPLACE_OR_REGENERATE_THE_CERTIFICATE_AND_ESTABLISH_TRUST_AGAIN_912358AF =
+          "advertised host {} 不在已加载的 OPC UA server 证书 subject alternative names 中。安全客户端可能拒绝该证书；"
+              + "请替换或重新生成证书并重新建立信任。";
 
   // ---------------------------------------------------------------------------
   // pipe – PipeDataNodePluginAgent

diff --git a/...b-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java b/...b-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaSink.java
@@ -70,6 +70,9 @@
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USERNAME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USER_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_IOTDB_USER_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_ADVERTISED_HOST_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_DEFAULT_VALUE;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_DEBOUNCE_TIME_MS_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_DEBOUNCE_TIME_MS_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_DEFAULT_QUALITY_BAD_VALUE;
@@ -112,6 +115,8 @@
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_PASSWORD_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USERNAME_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_IOTDB_USER_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_ADVERTISED_HOST_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_DEBOUNCE_TIME_MS_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_DEFAULT_QUALITY_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_ENABLE_ANONYMOUS_ACCESS_KEY;
@@ -268,6 +273,9 @@ private void customizeServer(final PipeParameters parameters) {
         parameters.getIntOrDefault(
             Arrays.asList(CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY, SINK_OPC_UA_HTTPS_BIND_PORT_KEY),
             CONNECTOR_OPC_UA_HTTPS_BIND_PORT_DEFAULT_VALUE);
+    final String advertisedHost =
+        parameters.getStringByKeys(
+            CONNECTOR_OPC_UA_ADVERTISED_HOST_KEY, SINK_OPC_UA_ADVERTISED_HOST_KEY);
 
     final String user =
         parameters.getStringOrDefault(
@@ -330,6 +338,7 @@ private void customizeServer(final PipeParameters parameters) {
                             new OpcUaServerBuilder()
                                 .setTcpBindPort(tcpBindPort)
                                 .setHttpsBindPort(httpsBindPort)
+                                .setAdvertisedHost(advertisedHost)
                                 .setUser(user)
                                 .setPassword(password)
                                 .setSecurityDir(securityDir)
@@ -345,6 +354,7 @@ private void customizeServer(final PipeParameters parameters) {
                         oldValue
                             .getRight()
                             .checkEquals(
+                                advertisedHost,
                                 user,
                                 password,
                                 securityDir,
@@ -399,6 +409,12 @@ private void customizeClient(final PipeParameters parameters) {
         parameters.getLongOrDefault(
             Arrays.asList(CONNECTOR_OPC_UA_TIMEOUT_SECONDS_KEY, SINK_OPC_UA_TIMEOUT_SECONDS_KEY),
             CONNECTOR_OPC_UA_TIMEOUT_SECONDS_DEFAULT_VALUE);
+    final boolean allowEndpointRedirect =
+        parameters.getBooleanOrDefault(
+            Arrays.asList(
+                CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY,
+                SINK_OPC_UA_ALLOW_ENDPOINT_REDIRECT_KEY),
+            CONNECTOR_OPC_UA_ALLOW_ENDPOINT_REDIRECT_DEFAULT_VALUE);
 
     synchronized (CLIENT_KEY_TO_REFERENCE_COUNT_AND_CLIENT_MAP) {
       client =
@@ -418,11 +434,20 @@ private void customizeClient(final PipeParameters parameters) {
                                       SINK_OPC_UA_HISTORIZING_KEY),
                                   CONNECTOR_OPC_UA_HISTORIZING_DEFAULT_VALUE));
                       final ClientRunner runner =
-                          new ClientRunner(result, securityDir, password, userName, timeoutSeconds);
+                          new ClientRunner(
+                              result,
+                              securityDir,
+                              password,
+                              userName,
+                              timeoutSeconds,
+                              allowEndpointRedirect);
                       runner.run();
                       return new Pair<>(new AtomicInteger(0), result);
                     }
-                    oldValue.getRight().checkEquals(userName, password, securityDir, policy);
+                    oldValue
+                        .getRight()
+                        .checkEquals(
+                            userName, password, securityDir, policy, allowEndpointRedirect);
                     return oldValue;
                   })
               .getRight();

diff --git a/...anode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java b/...anode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/ClientRunner.java
@@ -27,7 +27,10 @@
 import org.eclipse.milo.opcua.stack.client.security.DefaultClientCertificateValidator;
 import org.eclipse.milo.opcua.stack.core.security.DefaultTrustListManager;
 import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
+import org.eclipse.milo.opcua.stack.core.transport.TransportProfile;
 import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
+import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
+import org.eclipse.milo.opcua.stack.core.util.EndpointUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,7 +40,10 @@
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.security.Security;
+import java.util.List;
+import java.util.Locale;
 import java.util.Objects;
+import java.util.Optional;
 
 import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
 
@@ -54,6 +60,7 @@ public class ClientRunner {
   private final Path securityDir;
   private final String password;
   private final long timeoutSeconds;
+  private final boolean allowEndpointRedirect;
 
   // For conflict checking
   private final String user;
@@ -64,11 +71,22 @@ public ClientRunner(
       final String password,
       final String user,
       final long timeoutSeconds) {
+    this(configurableUaClient, securityDir, password, user, timeoutSeconds, false);
+  }
+
+  public ClientRunner(
+      final IoTDBOpcUaClient configurableUaClient,
+      final String securityDir,
+      final String password,
+      final String user,
+      final long timeoutSeconds,
+      final boolean allowEndpointRedirect) {
     this.configurableUaClient = configurableUaClient;
     this.securityDir = Paths.get(securityDir);
     this.password = password;
     this.user = user;
     this.timeoutSeconds = timeoutSeconds;
+    this.allowEndpointRedirect = allowEndpointRedirect;
     configurableUaClient.setRunner(this);
   }
 
@@ -93,7 +111,12 @@ private OpcUaClient createClient() throws Exception {
 
     return OpcUaClient.create(
         configurableUaClient.getNodeUrl(),
-        endpoints -> endpoints.stream().filter(configurableUaClient.endpointFilter()).findFirst(),
+        endpoints ->
+            selectEndpoint(
+                endpoints,
+                configurableUaClient.getNodeUrl(),
+                configurableUaClient.getSecurityPolicy(),
+                allowEndpointRedirect),
         configBuilder ->
             configBuilder
                 .setApplicationName(LocalizedText.english("Apache IoTDB OPC UA client"))
@@ -109,6 +132,66 @@ private OpcUaClient createClient() throws Exception {
                 .build());
   }
 
+  static Optional<EndpointDescription> selectEndpoint(
+      final List<EndpointDescription> endpoints,
+      final String configuredNodeUrl,
+      final SecurityPolicy securityPolicy,
+      final boolean allowEndpointRedirect) {
+    final String configuredScheme = normalizeScheme(EndpointUtil.getScheme(configuredNodeUrl));
+
+    return endpoints.stream()
+        .filter(endpoint -> securityPolicy.getUri().equals(endpoint.getSecurityPolicyUri()))
+        .filter(endpoint -> matchesConfiguredTransport(endpoint, configuredScheme))
+        .findFirst()
+        .map(
+            advertisedEndpoint -> {
+              final EndpointDescription effectiveEndpoint =
+                  allowEndpointRedirect
+                      ? advertisedEndpoint
+                      : EndpointUtil.updateUrl(
+                          advertisedEndpoint,
+                          EndpointUtil.getHost(configuredNodeUrl),
+                          EndpointUtil.getPort(configuredNodeUrl));
+              logger.info(
+                  DataNodePipeMessages
+                      .LOG_OPC_UA_ENDPOINT_SELECTED_CONFIGURED_ARG_ADVERTISED_ARG_EFFECTIVE_ARG_ALLOWENDPOINTREDIRECT_ARG_4FE076CB,
+                  configuredNodeUrl,
+                  advertisedEndpoint.getEndpointUrl(),
+                  effectiveEndpoint.getEndpointUrl(),
+                  allowEndpointRedirect);
+              return effectiveEndpoint;
+            });
+  }
+
+  private static boolean matchesConfiguredTransport(
+      final EndpointDescription endpoint, final String configuredScheme) {
+    if (Objects.isNull(configuredScheme)
+        || !Objects.equals(
+            configuredScheme, normalizeScheme(EndpointUtil.getScheme(endpoint.getEndpointUrl())))) {
+      return false;
+    }
+
+    final String transportProfileUri = endpoint.getTransportProfileUri();
+    if (Objects.isNull(transportProfileUri)) {
+      return true;
+    }
+    try {
+      return Objects.equals(
+          configuredScheme,
+          normalizeScheme(TransportProfile.fromUri(transportProfileUri).getScheme()));
+    } catch (final IllegalArgumentException ignored) {
+      // Preserve compatibility with servers that advertise a custom transport profile. The
+      // endpoint URL scheme still has to match the configured URL.
+      return true;
+    }
+  }
+
+  private static String normalizeScheme(final String scheme) {
+    return Objects.nonNull(scheme) && scheme.equalsIgnoreCase("opc.https")
+        ? "https"
+        : Objects.isNull(scheme) ? null : scheme.toLowerCase(Locale.ROOT);
+  }
+
   public void run() {
     try {
       final OpcUaClient client = createClient();
@@ -143,14 +226,16 @@ void checkEquals(
       final String user,
       final String password,
       final Path securityDir,
-      final SecurityPolicy securityPolicy) {
+      final SecurityPolicy securityPolicy,
+      final boolean allowEndpointRedirect) {
     checkEquals("user", this.user, user);
     checkEquals("password", this.password, password);
     checkEquals(
         "security dir",
         FileSystems.getDefault().getPath(this.securityDir.toAbsolutePath().toString()),
         FileSystems.getDefault().getPath(securityDir.toAbsolutePath().toString()));
     checkEquals("securityPolicy", configurableUaClient.getSecurityPolicy(), securityPolicy);
+    checkEquals("allow endpoint redirect", this.allowEndpointRedirect, allowEndpointRedirect);
   }
 
   private void checkEquals(final String attrName, Object thisAttr, Object thatAttr) {

diff --git a/...e/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java b/...e/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/client/IoTDBOpcUaClient.java
@@ -52,7 +52,6 @@
 import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesItem;
 import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResponse;
 import org.eclipse.milo.opcua.stack.core.types.structured.AddNodesResult;
-import org.eclipse.milo.opcua.stack.core.types.structured.EndpointDescription;
 import org.eclipse.milo.opcua.stack.core.types.structured.ObjectAttributes;
 import org.eclipse.milo.opcua.stack.core.types.structured.VariableAttributes;
 import org.slf4j.Logger;
@@ -65,7 +64,6 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ExecutionException;
-import java.util.function.Predicate;
 
 import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.convertToOpcDataType;
 import static org.apache.iotdb.db.pipe.sink.protocol.opcua.server.OpcUaNameSpace.timestampToUtc;
@@ -307,10 +305,6 @@ String getNodeUrl() {
     return nodeUrl;
   }
 
-  Predicate<EndpointDescription> endpointFilter() {
-    return e -> getSecurityPolicy().getUri().equals(e.getSecurityPolicyUri());
-  }
-
   SecurityPolicy getSecurityPolicy() {
     return securityPolicy;
   }
@@ -365,7 +359,9 @@ public void checkEquals(
       final String user,
       final String password,
       final String securityDir,
-      final SecurityPolicy securityPolicy) {
-    runner.checkEquals(user, password, Paths.get(securityDir), securityPolicy);
+      final SecurityPolicy securityPolicy,
+      final boolean allowEndpointRedirect) {
+    runner.checkEquals(
+        user, password, Paths.get(securityDir), securityPolicy, allowEndpointRedirect);
   }
 }
diff --git a/...rc/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java b/...rc/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaKeyStoreLoader.java
@@ -22,7 +22,7 @@
 import org.apache.iotdb.commons.utils.FileUtils;
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 
-import com.google.common.collect.Sets;
+import com.google.common.net.InetAddresses;
 import org.eclipse.milo.opcua.sdk.server.util.HostnameUtil;
 import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateBuilder;
 import org.eclipse.milo.opcua.stack.core.util.SelfSignedCertificateGenerator;
@@ -41,22 +41,21 @@
 import java.security.PrivateKey;
 import java.security.PublicKey;
 import java.security.cert.X509Certificate;
+import java.util.LinkedHashSet;
 import java.util.Set;
 import java.util.UUID;
-import java.util.regex.Pattern;
 
 class OpcUaKeyStoreLoader {
   private static final Logger LOGGER = LoggerFactory.getLogger(OpcUaKeyStoreLoader.class);
 
-  private static final Pattern IP_ADDR_PATTERN =
-      Pattern.compile("^(([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.){3}([01]?\\d\\d?|2[0-4]\\d|25[0-5])$");
-
   private static final String SERVER_ALIAS = "server-ai";
 
   private X509Certificate serverCertificate;
   private KeyPair serverKeyPair;
 
-  OpcUaKeyStoreLoader load(final Path baseDir, final char[] password) throws Exception {
+  OpcUaKeyStoreLoader load(
+      final Path baseDir, final char[] password, final Set<String> advertisedHostnames)
+      throws Exception {
     final KeyStore keyStore = KeyStore.getInstance("PKCS12");
 
     final File serverKeyStore = baseDir.resolve("iotdb-server.pfx").toFile();
@@ -90,14 +89,14 @@ OpcUaKeyStoreLoader load(final Path baseDir, final char[] password) throws Excep
               .setApplicationUri(applicationUri);
 
       // Get as many hostnames and IP addresses as we can list in the certificate.
-      final Set<String> hostnames =
-          Sets.union(
-              Sets.newHashSet(HostnameUtil.getHostname()),
-              HostnameUtil.getHostnames("0.0.0.0", false));
+      final Set<String> hostnames = new LinkedHashSet<>();
+      hostnames.add(HostnameUtil.getHostname());
+      hostnames.addAll(HostnameUtil.getHostnames("0.0.0.0", false));
+      hostnames.addAll(advertisedHostnames);
 
       hostnames.forEach(
           hostname -> {
-            if (IP_ADDR_PATTERN.matcher(hostname).matches()) {
+            if (InetAddresses.isInetAddress(hostname)) {
               builder.addIpAddress(hostname);
             } else {
               builder.addDnsName(hostname);

diff --git a/...ode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java b/...ode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/server/OpcUaNameSpace.java
@@ -703,13 +703,15 @@ public void onMonitoringModeChanged(final List<MonitoredItem> monitoredItems) {
   /////////////////////////////// Conflict detection ///////////////////////////////
 
   public void checkEquals(
+      final String advertisedHost,
       final String user,
       final String password,
       final String securityDir,
       final boolean enableAnonymousAccess,
       final Set<SecurityPolicy> securityPolicies,
       final long debounceTimeMs) {
     builder.checkEquals(
+        advertisedHost,
         user,
         password,
         Paths.get(securityDir),