fix

Author: Caideyipi

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

@@ -68,6 +68,8 @@
 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_ENABLE_ANONYMOUS_ACCESS_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_ENABLE_ANONYMOUS_ACCESS_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_HISTORIZING_DEFAULT_VALUE;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_HISTORIZING_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_DEFAULT_VALUE;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_HTTPS_BIND_PORT_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.CONNECTOR_OPC_UA_MODEL_CLIENT_SERVER_VALUE;
@@ -98,6 +100,7 @@
 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_ENABLE_ANONYMOUS_ACCESS_KEY;
+import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_HISTORIZING_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_HTTPS_BIND_PORT_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_MODEL_KEY;
 import static org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant.SINK_OPC_UA_NODE_URL_KEY;
@@ -358,7 +361,14 @@ private void customizeClient(final String nodeUrl, final PipeParameters paramete
                     + File.separatorChar
                     + UUID.nameUUIDFromBytes(nodeUrl.getBytes(TSFileConfig.STRING_CHARSET))));
 
-    client = new IoTDBOpcUaClient(nodeUrl, policy, provider);
+    client =
+        new IoTDBOpcUaClient(
+            nodeUrl,
+            policy,
+            provider,
+            parameters.getBooleanOrDefault(
+                Arrays.asList(CONNECTOR_OPC_UA_HISTORIZING_KEY, SINK_OPC_UA_HISTORIZING_KEY),
+                CONNECTOR_OPC_UA_HISTORIZING_DEFAULT_VALUE));
     new ClientRunner(client, securityDir, password).run();
   }
 

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

@@ -67,14 +67,17 @@ public class IoTDBOpcUaClient {
   private final SecurityPolicy securityPolicy;
   private final IdentityProvider identityProvider;
   private OpcUaClient client;
+  private boolean historizing;
 
   public IoTDBOpcUaClient(
       final String nodeUrl,
       final SecurityPolicy securityPolicy,
-      final IdentityProvider identityProvider) {
+      final IdentityProvider identityProvider,
+      final boolean historizing) {
     this.nodeUrl = nodeUrl;
     this.securityPolicy = securityPolicy;
     this.identityProvider = identityProvider;
+    this.historizing = historizing;
   }
 
   public void run(final OpcUaClient client) throws Exception {
@@ -251,7 +254,7 @@ IdentityProvider getIdentityProvider() {
 
   /////////////////////////////// Attribute creator ///////////////////////////////
 
-  public static VariableAttributes createMeasurementAttributes(
+  private VariableAttributes createMeasurementAttributes(
       final String name, final NodeId objectType, final Variant initialValue) {
     return new VariableAttributes(
         Unsigned.uint(0xFFFF), // specifiedAttributes
@@ -266,11 +269,10 @@ public static VariableAttributes createMeasurementAttributes(
         AccessLevel.toValue(AccessLevel.READ_WRITE),
         AccessLevel.toValue(AccessLevel.READ_WRITE),
         500.0, // samplingInterval
-        false // historizing
-        );
+        historizing);
   }
 
-  public static ObjectAttributes createFolderAttributes(final String name) {
+  private static ObjectAttributes createFolderAttributes(final String name) {
     return new ObjectAttributes(
         Unsigned.uint(0xFFFF), // specifiedAttributes
         LocalizedText.english(name),

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeSinkConstant.java

@@ -211,6 +211,10 @@ public class PipeSinkConstant {
   public static final String CONNECTOR_OPC_UA_QUALITY_SECURITY_POLICY_AES256_SHA256_RSAPSS_VALUE =
       "AES256_SHA256_RSAPSS";
 
+  public static final String CONNECTOR_OPC_UA_HISTORIZING_KEY = "connector.opcua.historizing";
+  public static final String SINK_OPC_UA_HISTORIZING_KEY = "sink.opcua.historizing";
+  public static final boolean CONNECTOR_OPC_UA_HISTORIZING_DEFAULT_VALUE = false;
+
   public static final String CONNECTOR_LEADER_CACHE_ENABLE_KEY = "connector.leader-cache.enable";
   public static final String SINK_LEADER_CACHE_ENABLE_KEY = "sink.leader-cache.enable";
   public static final boolean CONNECTOR_LEADER_CACHE_ENABLE_DEFAULT_VALUE = true;