@Override
  public void customize(
      final PipeParameters parameters, final PipeConnectorRuntimeConfiguration configuration)
      throws Exception {
    final int tcpBindPort =
        parameters.getIntOrDefault(
            Arrays.asList(CONNECTOR_OPC_UA_TCP_BIND_PORT_KEY, SINK_OPC_UA_TCP_BIND_PORT_KEY),
            CONNECTOR_OPC_UA_TCP_BIND_PORT_DEFAULT_VALUE);
    final int httpsBindPort =
        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 user =
        parameters.getStringOrDefault(
            Arrays.asList(
                CONNECTOR_IOTDB_USER_KEY,
                SINK_IOTDB_USER_KEY,
                CONNECTOR_IOTDB_USERNAME_KEY,
                SINK_IOTDB_USERNAME_KEY),
            CONNECTOR_IOTDB_USER_DEFAULT_VALUE);
    final String password =
        parameters.getStringOrDefault(
            Arrays.asList(CONNECTOR_IOTDB_PASSWORD_KEY, SINK_IOTDB_PASSWORD_KEY),
            CONNECTOR_IOTDB_PASSWORD_DEFAULT_VALUE);
    final String securityDir =
        IoTDBConfig.addDataHomeDir(
            parameters.getStringOrDefault(
                Arrays.asList(CONNECTOR_OPC_UA_SECURITY_DIR_KEY, SINK_OPC_UA_SECURITY_DIR_KEY),
                CONNECTOR_OPC_UA_SECURITY_DIR_DEFAULT_VALUE
                    + File.separatorChar
                    + httpsBindPort
                    + "_"
                    + tcpBindPort));
    final boolean enableAnonymousAccess =
        parameters.getBooleanOrDefault(
            Arrays.asList(
                CONNECTOR_OPC_UA_ENABLE_ANONYMOUS_ACCESS_KEY,
                SINK_OPC_UA_ENABLE_ANONYMOUS_ACCESS_KEY),
            CONNECTOR_OPC_UA_ENABLE_ANONYMOUS_ACCESS_DEFAULT_VALUE);
    final String placeHolder =
        parameters.getStringOrDefault(
            Arrays.asList(CONNECTOR_OPC_UA_PLACEHOLDER_KEY, SINK_OPC_UA_PLACEHOLDER_KEY),
            CONNECTOR_OPC_UA_PLACEHOLDER_DEFAULT_VALUE);

    synchronized (SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP) {
      serverKey = httpsBindPort + ":" + tcpBindPort;

      nameSpace =
          SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP
              .compute(
                  serverKey,
                  (key, oldValue) -> {
                    try {
                      if (Objects.isNull(oldValue)) {
                        final OpcUaServerBuilder builder =
                            new OpcUaServerBuilder()
                                .setTcpBindPort(tcpBindPort)
                                .setHttpsBindPort(httpsBindPort)
                                .setUser(user)
                                .setPassword(password)
                                .setSecurityDir(securityDir)
                                .setEnableAnonymousAccess(enableAnonymousAccess);
                        final OpcUaServer newServer = builder.build();
                        final DataRegion region =
                            StorageEngine.getInstance()
                                .getDataRegion(
                                    new DataRegionId(
                                        configuration.getRuntimeEnvironment().getRegionId()));
                        nameSpace =
                            new OpcUaNameSpace(
                                newServer,
                                parameters
                                    .getStringOrDefault(
                                        Arrays.asList(
                                            CONNECTOR_OPC_UA_MODEL_KEY, SINK_OPC_UA_MODEL_KEY),
                                        CONNECTOR_OPC_UA_MODEL_DEFAULT_VALUE)
                                    .equals(CONNECTOR_OPC_UA_MODEL_CLIENT_SERVER_VALUE),
                                builder,
                                Objects.nonNull(region)
                                    ? region.getDatabaseName()
                                    : "root.__temp_db",
                                placeHolder);
                        nameSpace.startup();
                        newServer.startup().get();
                        return new Pair<>(new AtomicInteger(0), nameSpace);
                      } else {
                        oldValue
                            .getRight()
                            .checkEquals(user, password, securityDir, enableAnonymousAccess);
                        return oldValue;
                      }
                    } catch (final PipeException e) {
                      throw e;
                    } catch (final Exception e) {
                      throw new PipeException("Failed to build and startup OpcUaServer", e);
                    }
                  })
              .getRight();
      SERVER_KEY_TO_REFERENCE_COUNT_AND_NAME_SPACE_MAP.get(serverKey).getLeft().incrementAndGet();
    }
  }
Send data in IoTDB based on Opc Ua protocol, using Eclipse Milo. All data are converted into
tablets, and then:

<p>1. In pub-sub mode, converted to eventNodes to send to the subscriber clients.

<p>2. In client-server mode, push the newest value to the local server.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/opcua/OpcUaConnector.java