[To dev/1.3] Pipe: Added robustness to OPC UA measurement logic

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/opcua/OpcUaNameSpace.java

@@ -41,6 +41,7 @@
 import org.eclipse.milo.opcua.sdk.server.api.MonitoredItem;
 import org.eclipse.milo.opcua.sdk.server.model.nodes.objects.BaseEventTypeNode;
 import org.eclipse.milo.opcua.sdk.server.nodes.UaFolderNode;
+import org.eclipse.milo.opcua.sdk.server.nodes.UaNode;
 import org.eclipse.milo.opcua.sdk.server.nodes.UaVariableNode;
 import org.eclipse.milo.opcua.sdk.server.util.SubscriptionModel;
 import org.eclipse.milo.opcua.stack.core.Identifiers;
@@ -108,10 +109,10 @@ private void transferTabletForClientServerModel(final Tablet tablet) {
       throw new PipeRuntimeCriticalException("The segments of tablets must exist");
     }
     final StringBuilder currentStr = new StringBuilder();
-    UaFolderNode folderNode = null;
+    UaNode folderNode = null;
     NodeId folderNodeId;
     for (final String segment : segments) {
-      final UaFolderNode nextFolderNode;
+      final UaNode nextFolderNode;
 
       currentStr.append(segment);
       folderNodeId = newNodeId(currentStr.toString());
@@ -126,7 +127,12 @@ private void transferTabletForClientServerModel(final Tablet tablet) {
                 LocalizedText.english(segment));
         getNodeManager().addNode(nextFolderNode);
         if (Objects.nonNull(folderNode)) {
-          folderNode.addOrganizes(nextFolderNode);
+          folderNode.addReference(
+              new Reference(
+                  folderNode.getNodeId(),
+                  Identifiers.Organizes,
+                  nextFolderNode.getNodeId().expanded(),
+                  true));
         } else {
           nextFolderNode.addReference(
               new Reference(
@@ -138,14 +144,13 @@ private void transferTabletForClientServerModel(final Tablet tablet) {
         folderNode = nextFolderNode;
       } else {
         folderNode =
-            (UaFolderNode)
-                getNodeManager()
-                    .getNode(folderNodeId)
-                    .orElseThrow(
-                        () ->
-                            new PipeRuntimeCriticalException(
-                                String.format(
-                                    "The folder node for %s does not exist.", tablet.deviceId)));
+            getNodeManager()
+                .getNode(folderNodeId)
+                .orElseThrow(
+                    () ->
+                        new PipeRuntimeCriticalException(
+                            String.format(
+                                "The folder node for %s does not exist.", tablet.deviceId)));
       }
     }
 
@@ -189,7 +194,15 @@ private void transferTabletForClientServerModel(final Tablet tablet) {
                 .setValue(value)
                 .build();
         getNodeManager().addNode(measurementNode);
-        folderNode.addOrganizes(measurementNode);
+        if (Objects.nonNull(folderNode)) {
+          folderNode.addReference(
+              new Reference(
+                  folderNode.getNodeId(), Identifiers.Organizes, nodeId.expanded(), true));
+        } else {
+          measurementNode.addReference(
+              new Reference(
+                  nodeId, Identifiers.Organizes, Identifiers.ObjectsFolder.expanded(), false));
+        }
       } else {
         // This must exist
         measurementNode =